Merge pull request #128 from csmith/master

Merge OutputQueue and QueueHandlers.

irc/src/com/dmdirc/parser/irc/IRCParser.java

 import com.dmdirc.parser.irc.events.IRCDataInEvent;
 import com.dmdirc.parser.irc.events.IRCDataOutEvent;
 import com.dmdirc.parser.irc.outputqueue.OutputQueue;
+import com.dmdirc.parser.irc.outputqueue.PriorityOutputQueue;
 
 import java.io.IOException;
 import java.net.InetAddress;
         myProcessingManager = graph.get(ProcessingManager.class);
         myself = new IRCClientInfo(this, userModes, "myself").setFake(true);
 
-        out = new OutputQueue();
+        out = new PriorityOutputQueue();
         if (myDetails != null) {
             this.me = myDetails;
         }
             userModes.clear();
             chanPrefix = DEFAULT_CHAN_PREFIX;
             // Clear output queue.
-            if (out != null) {
-                out.clearQueue();
-            }
+            out.clearQueue();
             setServerName("");
             networkName = "";
             lastLine = null;
      * @return True if line was sent, else false.
      */
     protected boolean sendString(final String line, final QueuePriority priority, final boolean fromParser) {
-        if (out == null || getSocketState() != SocketState.OPEN) {
+        if (getSocketState() != SocketState.OPEN) {
             return false;
         }
         callDataOut(line, fromParser);

irc/src/com/dmdirc/parser/irc/outputqueue/OutputQueue.java

 
 import java.io.OutputStream;
 import java.io.PrintWriter;
+import java.util.Comparator;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.PriorityBlockingQueue;
 
 /**
  * This class handles the Parser output Queue.
  */
-public class OutputQueue {
+public abstract class OutputQueue {
 
     /** PrintWriter for sending output. */
     private PrintWriter out;
     /** Is queueing enabled? */
     private boolean queueEnabled = true;
     /** Are we discarding all futher input? */
-    private boolean discarding = false;
+    private boolean discarding;
     /** The output queue! */
-    private BlockingQueue<QueueItem> queue = new PriorityBlockingQueue<>();
-    /** Object for synchronising access to the {@link #queueHandler}. */
-    private final Object queueHandlerLock = new Object();
-    /** Thread for the sending queue. */
-    private QueueHandler queueHandler;
-    /** The QueueHandlerFactory for this OutputQueue. */
-    private final QueueHandlerFactory queueHandlerFactory;
+    private final BlockingQueue<QueueItem> queue;
+    /** The thread on which we will send items. */
+    private Thread sendingThread;
 
     /**
-     * Creates a new output queue using the default handler.
-     */
-    public OutputQueue() {
-        this(PriorityQueueHandler.getFactory());
-    }
-
-    /**
-     * Creates a new output queue using a handler from the given factory.
+     * Creates a new instance of {@link OutputQueue} that will sort items using the given
+     * comparator.
      *
-     * @param queueHandlerFactory The factory to use to create {@link QueueHandler}s.
+     * @param itemComparator The comparator to use to sort queued items.
      */
-    public OutputQueue(final QueueHandlerFactory queueHandlerFactory) {
-        this.queueHandlerFactory = queueHandlerFactory;
+    protected OutputQueue(final Comparator<QueueItem> itemComparator) {
+        queue = new PriorityBlockingQueue<>(10, itemComparator);
     }
 
     /**
         return queueEnabled;
     }
 
-    /**
-     * Get the QueueHandler.
-     *
-     * @return The current QueueHandler if there is one, else null.
-     */
-    public QueueHandler getQueueHandler() {
-        return queueHandler;
-    }
-
     /**
      * Set if queueing is enabled.
      * if this is changed from enabled to disabled, all currently queued items
         // If the new value is not the same as the old one, and we used to be enabled
         // then flush the queue.
         if (old != queueEnabled && old) {
-            synchronized (queueHandlerLock) {
-                if (queueHandler != null) {
-                    queueHandler.interrupt();
-                    queueHandler = null;
-                }
+            if (sendingThread != null) {
+                sendingThread.interrupt();
+                sendingThread = null;
             }
 
             while (!queue.isEmpty()) {
     }
 
     /**
-     * Clear the queue and stop the thread that is sending stuff.
+     * Clears any pending items.
      */
     public void clearQueue() {
-        this.queueEnabled = false;
-
-        synchronized (queueHandlerLock) {
-            if (queueHandler != null) {
-                queueHandler.interrupt();
-                queueHandler = null;
-            }
+        queueEnabled = false;
+        if (sendingThread != null) {
+            sendingThread.interrupt();
+            sendingThread = null;
         }
-
         queue.clear();
     }
 
 
     /**
      * Send the given line.
-     * If queueing is enabled, this will queue it, else it will send it
-     * immediately.
+     *
+     * <p>If queueing is enabled, this will queue it, else it will send it immediately.
      *
      * @param line Line to send
      */
 
     /**
      * Send the given line.
-     * If queueing is enabled, this will queue it, else it will send it
-     * immediately.
+     *
+     * <p>If queueing is enabled, this will queue it, else it will send it immediately.
      *
      * @param line Line to send
      * @param priority Priority of item (ignored if queue is disabled)
      */
     public void sendLine(final String line, final QueuePriority priority) {
-        if (discarding) { return; }
-        if (out == null) {
-            throw new NullPointerException("No output stream has been set.");
+        if (discarding) {
+            return;
         }
 
-        if (queueEnabled && priority != QueuePriority.IMMEDIATE) {
-            synchronized (queueHandlerLock) {
-                if (queueHandler == null || !queueHandler.isAlive()) {
-                    setQueueHandler(queueHandlerFactory.getQueueHandler(this, out));
-                    queueHandler.start();
-                }
-
-                queue.add(queueHandler.getQueueItem(line, priority));
-            }
+        if (queueEnabled && priority == QueuePriority.IMMEDIATE) {
+            send(line);
         } else {
-            out.printf("%s\r\n", line);
+            if (sendingThread == null || !sendingThread.isAlive()) {
+                sendingThread = new Thread(this::handleQueuedItems, "IRC Parser queue handler");
+                sendingThread.start();
+            }
+
+            enqueue(line, priority);
         }
     }
 
     /**
-     * Sets the hanlder that this queue will use. This will cause the existing {@link #queue}
-     * to be replaced with a new version with an updated comparator.
+     * Sends queued items to the output channel, blocking or waiting as necessary.
      *
-     * @param queueHandler The new queue handler to use.
+     * <p>This is a long running method that will be executed in a separate thread.
      */
-    private void setQueueHandler(final QueueHandler queueHandler) {
-        this.queueHandler = queueHandler;
-        final BlockingQueue<QueueItem> newQueue = new PriorityBlockingQueue<>(
-                10, queueHandler.getQueueItemComparator());
-        newQueue.addAll(queue);
-        queue = newQueue;
+    protected abstract void handleQueuedItems();
+
+    /**
+     * Enqueues a new line to be sent.
+     *
+     * @param line The raw line to be sent to the server.
+     * @param priority The priority at which the line should be sent.
+     */
+    protected void enqueue(final String line, final QueuePriority priority) {
+        queue.add(new QueueItem(line, priority));
     }
 
+    /**
+     * Sends a line immediately to the server.
+     *
+     * @param line The line to be sent.
+     */
+    protected void send(final String line) {
+        if (out == null) {
+            throw new IllegalStateException("No output stream has been set.");
+        }
+
+        out.printf("%s\r\n", line);
+    }
+
+
 }

irc/src/com/dmdirc/parser/irc/outputqueue/QueueHandlerFactory.java → irc/src/com/dmdirc/parser/irc/outputqueue/PriorityOutputQueue.java

 
 package com.dmdirc.parser.irc.outputqueue;
 
-import java.io.PrintWriter;
+import java.time.Duration;
 
 /**
- * A QueueHandlerFactory produces QueueHandlers for OutputQueue.
+ * This does no rate limiting, and just sends based on priority.
  */
-public interface QueueHandlerFactory {
+public class PriorityOutputQueue extends OutputQueue {
 
     /**
-     * Get a new QueueHandler instance as needed.
-     *
-     * @param outputQueue the OutputQueue that will own this QueueHandler
-     * @param out Where to send crap.
-     * @return the new queue handler object.
+     * Create a new PriorityOutputQueue.
      */
-    QueueHandler getQueueHandler(final OutputQueue outputQueue, final PrintWriter out);
+    public PriorityOutputQueue() {
+        super(QueueComparators.byPriorityThenNumber(Duration.ofSeconds(10)));
+    }
+
+    @Override
+    protected void handleQueuedItems() {
+        try {
+            while (isQueueEnabled()) {
+                send(getQueue().take().getLine());
+            }
+        } catch (InterruptedException ex) {
+            // Do nothing
+        }
+    }
 
 }

irc/src/com/dmdirc/parser/irc/outputqueue/PriorityQueueHandler.java

-/*
- * Copyright (c) 2006-2015 DMDirc Developers
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- * SOFTWARE.
- */
-
-package com.dmdirc.parser.irc.outputqueue;
-
-import java.io.PrintWriter;
-import java.time.Duration;
-
-/**
- * This does no rate limiting, and just sends based on priority.
- */
-public class PriorityQueueHandler extends QueueHandler {
-
-    /**
-     * Create a new PriorityQueueHandler.
-     *
-     * @param outputQueue Owner of this Queue Handler
-     * @param out Output Stream to use
-     */
-    public PriorityQueueHandler(
-            final OutputQueue outputQueue,
-            final PrintWriter out) {
-        super(outputQueue,
-                QueueComparators.byPriorityThenNumber(Duration.ofSeconds(10)),
-                out);
-    }
-
-    /**
-     * Get a QueueHandlerFactory that produces PriorityQueueHandlers.
-     *
-     * @return a QueueHandlerFactory that produces PrirortyQueueHandlers.
-     */
-    public static QueueHandlerFactory getFactory() {
-        return PriorityQueueHandler::new;
-    }
-
-    @Override
-    public void run() {
-        try {
-            while (outputQueue.isQueueEnabled()) {
-                final QueueItem item = outputQueue.getQueue().take();
-
-                sendLine(item.getLine());
-            }
-        } catch (InterruptedException ex) {
-            // Do nothing
-        }
-    }
-
-}

irc/src/com/dmdirc/parser/irc/outputqueue/QueueHandler.java

-/*
- * Copyright (c) 2006-2015 DMDirc Developers
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- * SOFTWARE.
- */
-
-package com.dmdirc.parser.irc.outputqueue;
-
-import com.dmdirc.parser.common.QueuePriority;
-
-import java.io.PrintWriter;
-import java.util.Comparator;
-
-/**
- * Sending queue.
- */
-public abstract class QueueHandler extends Thread {
-
-    /** The output queue that owns us. */
-    protected OutputQueue outputQueue;
-    /** Comparator to use to sort queue items. */
-    private final Comparator<QueueItem> comparator;
-    /** Where to send the output. */
-    private final PrintWriter out;
-
-    /**
-     * Create a new Queue Thread.
-     *
-     * @param outputQueue the OutputQueue that owns us.
-     * @param comparator Comparator to use to sort items in the queue.
-     * @param out Writer to send to.
-     */
-    public QueueHandler(
-            final OutputQueue outputQueue,
-            final Comparator<QueueItem> comparator,
-            final PrintWriter out) {
-        super("IRC Parser queue handler");
-
-        this.comparator = comparator;
-        this.out = out;
-        this.outputQueue = outputQueue;
-    }
-
-    /**
-     * Send the given item.
-     *
-     * @param line Line to send.
-     */
-    public void sendLine(final String line) {
-        out.printf("%s\r\n", line);
-    }
-
-    /**
-     * Get a new QueueItem for the given line and priority.
-     * By default this will just create a new QueueItem with the given
-     * parameters, but QueueHandlers are free to override it if they need to
-     * instead produce subclasses of QueueItem or do anything else with the
-     * data given.
-     *
-     * @param line Line to send
-     * @param priority Priority of the line.
-     * @return A QueueItem for teh given parameters
-     */
-    public QueueItem getQueueItem(final String line, final QueuePriority priority) {
-        return new QueueItem(line, priority);
-    }
-
-    /**
-     * Gets the comparator to use to sort queue items.
-     *
-     * @return The comparator to use to sort queue items.
-     */
-    public Comparator<QueueItem> getQueueItemComparator() {
-        return comparator;
-    }
-
-    /**
-     * This is the main even loop of the queue.
-     * It needs to handle pulling items out of the queue and calling
-     * sendLine.
-     *
-     * It also needs to handle any delays in sending that it deems needed.
-     */
-    @Override
-    public abstract void run();
-
-}

irc/src/com/dmdirc/parser/irc/outputqueue/SimpleRateLimitedQueueHandler.java → irc/src/com/dmdirc/parser/irc/outputqueue/SimpleRateLimitedOutputQueue.java

 
 import com.dmdirc.parser.common.QueuePriority;
 
-import java.io.PrintWriter;
-
 /**
  * This is a simple rate limiting queue.
  * If more than 4 items are added in 4 seconds it will start limiting.
  * The first 4 items will be sent un-limited and then limiting will commence at
  * a rate of 1 per second.
  */
-public class SimpleRateLimitedQueueHandler extends QueueHandler {
+public class SimpleRateLimitedOutputQueue extends OutputQueue {
 
     /** Current count. */
     private int count;
     private boolean alwaysUpdateTime = true;
 
     /**
-     * Create a new SimpleRateLimitedQueueHandler.
-     *
-     * @param outputQueue Owner of this Queue Handler
-     * @param out Output Stream to use
-     */
-    public SimpleRateLimitedQueueHandler(
-            final OutputQueue outputQueue,
-            final PrintWriter out) {
-        super(outputQueue, QueueComparators.byPriorityThenNumber(), out);
-    }
-
-    /**
-     * Get a QueueHandlerFactory that produces PriorityQueueHandlers.
-     *
-     * @return a QueueHandlerFactory that produces PrirortyQueueHandlers.
+     * Create a new SimpleRateLimitedOutputQueue.
      */
-    public static QueueHandlerFactory getFactory() {
-        return SimpleRateLimitedQueueHandler::new;
+    public SimpleRateLimitedOutputQueue() {
+        super(QueueComparators.byPriorityThenNumber());
     }
 
     /**
     }
 
     @Override
-    public QueueItem getQueueItem(final String line, final QueuePriority priority) {
+    protected void enqueue(final String line, final QueuePriority priority) {
         // Was the last line added less than limitTime ago?
         synchronized (this) {
             final boolean overTime = lastItemTime + limitTime > System.currentTimeMillis();
                 // It has been longer than limitTime and we are still shown as
                 // limiting, check to see if the queue is empty or not, if it is
                 // disable limiting.
-                if (outputQueue.getQueue().isEmpty()) {
+                if (getQueue().isEmpty()) {
                     isLimiting = false;
                 }
             } else {
             }
         }
 
-        return super.getQueueItem(line, priority);
+        super.enqueue(line, priority);
     }
 
     @Override
-    public void run() {
+    protected void handleQueuedItems() {
         try {
-            while (outputQueue.isQueueEnabled()) {
-                final QueueItem item = outputQueue.getQueue().take();
-
-                sendLine(item.getLine());
+            while (isQueueEnabled()) {
+                sendLine(getQueue().take().getLine());
 
                 final boolean doSleep;
                 synchronized (this) {
                     doSleep = isLimiting;
-                    if (isLimiting && outputQueue.getQueue().isEmpty()) {
+                    if (isLimiting && getQueue().isEmpty()) {
                         isLimiting = false;
                     }
                 }
                 if (doSleep) {
                     try {
                         Thread.sleep(waitTime);
-                    } catch (InterruptedException ex) { /* Do Nothing. */ }
+                    } catch (InterruptedException ex) {
+                        /* Do Nothing. */
+                    }
                 }
             }
         } catch (InterruptedException ex) {
             // Do nothing
         }
     }
+
 }