/**
 * 
 */
package mathsquared.resultswizard2;

import java.util.Queue;

/**
 * Feeds data from a queue to another object, and from the object to another queue. The object must be able to {@linkplain CommandProcessor process commands}.
 * 
 * @author MathSquared
 * 
 */
public class QueueCommunicator implements Runnable {
    private Queue rx; // receiving
    private Queue tx; // sending
    private CommandProcessor comm;

    public static final long SLEEP_TIME = 100; // time to sleep when rx is empty

    public QueueCommunicator (Queue rx, Queue tx, CommandProcessor comm) {
        this.rx = rx;
        this.tx = tx;
        this.comm = comm;
    }

    public void run () {
        boolean aborted = false;
        while (!aborted) {
            // Wait until we have data
            while (rx.size() == 0) {
                try {
                    Thread.sleep(SLEEP_TIME);
                } catch (InterruptedException e) {
                    // ignore
                }
            }

            // Process the data

            Object pop = rx.poll();

            // This is an object stream, so we must check and cast (and ensure that we're not about to NPE, just to be safe)
            if (pop != null && pop instanceof Command) {
                Command cmd = (Command) pop;
                Command send = comm.processMessage(cmd); // Get the command to send back
                if (send != null) { // Only send the command if there's actually something to send (SQProxy's queues choke on null)
                    tx.add(send);
                }
                // Ensure that we're not aborting (POISON or XMIT_ERROR_RESTART invalidate the streams)
                if (cmd.getType().equals(Message.POISON) || cmd.getType().equals(Message.XMIT_ERROR_RESTART)) {
                    aborted = true;
                }
            }
        }
    }
}

@@ -512,155 +512,1323 @@
     
-public QueueCommunicator (Queue rx, Queue tx, CommandProcessor comm) %7B%0A        this.rx = rx;%0A        this.tx = tx;%0A        this.comm = comm;%0A    %7D%0A
+/**%0A     * Creates a QueueCommunicator communicating between two given queues and a given %7B@link CommandProcessor%7D. Note that the QueueCommunicator must be %7B@linkplain #run() run%7D to have any effect.%0A     * %0A     * @param rx a queue from which to obtain messages received from the outside%0A     * @param tx a queue to which to send messages intended for return to the sender of any messages popped from %3Ccode%3Erx%3C/code%3E%0A     * @param comm a CommandProcessor that will receive the commands from %3Ccode%3Erx%3C/code%3E and generate new commands to send to %3Ccode%3Etx%3C/code%3E%0A     */%0A    public QueueCommunicator (Queue rx, Queue tx, CommandProcessor comm) %7B%0A        this.rx = rx;%0A        this.tx = tx;%0A        this.comm = comm;%0A    %7D%0A%0A    /**%0A     * Runs the QueueCommunicator until aborted.%0A     * %0A     * %3Cp%3E%0A     * A QueueCommunicator is aborted when a Command is removed from the queue given by %3Ccode%3Erx%3C/code%3E whose %7B@link Command#getType() getType()%7D method returns %7B@link Message#POISON%7D or %7B@link Message#XMIT_ERROR_RESTART%7D. If the CommandProcessor designates a response to these commands, it will be sent prior to the QueueCommunicator being aborted.%0A     * %3C/p%3E%0A     * %0A     * %3Cp%3E%0A     * After being aborted, a QueueCommunicator may be restarted, with the same queues, by calling this method again.%0A     * %3C/p%3E%0A     */
 %0A   

Document QueueCommunicator