Merge OutputQueue and QueueHandlers.

OutputQueue is now abstract, and the different queue methods
are subclasses.

This allows the same flexibility as before (subclasses can change
the type of items they add, implement their own throttling logic,
and say how the queue is ordered).

Organising it this way makes the package a lot easier to understand
as there's no longer a messy separation of concerns between
OutputQueue and QueueHandler (e.g. the output stream is no longer
in two places with two send methods, all of the definitions
relating to how the queue works are in one place, etc).

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

@@ -48,6 +48,7 @@ import com.dmdirc.parser.irc.IRCReader.ReadLine;
 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;
@@ -275,7 +276,7 @@ public class IRCParser extends BaseSocketAwareParser implements SecureParser, En
         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;
         }
@@ -662,9 +663,7 @@ public class IRCParser extends BaseSocketAwareParser implements SecureParser, En
             userModes.clear();
             chanPrefix = DEFAULT_CHAN_PREFIX;
             // Clear output queue.
-            if (out != null) {
-                out.clearQueue();
-            }
+            out.clearQueue();
             setServerName("");
             networkName = "";
             lastLine = null;
@@ -1026,7 +1025,7 @@ public class IRCParser extends BaseSocketAwareParser implements SecureParser, En
      * @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

@@ -26,43 +26,34 @@ import com.dmdirc.parser.common.QueuePriority;
 
 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);
     }
 
     /**
@@ -83,15 +74,6 @@ public class OutputQueue {
         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
@@ -110,11 +92,9 @@ public class OutputQueue {
         // 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()) {
@@ -155,18 +135,14 @@ public class OutputQueue {
     }
 
     /**
-     * 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();
     }
 
@@ -181,8 +157,8 @@ public class OutputQueue {
 
     /**
      * 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
      */
@@ -192,44 +168,58 @@ public class OutputQueue {
 
     /**
      * 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

@@ -22,20 +22,29 @@
 
 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

@@ -1,69 +0,0 @@
-/*
- * 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

@@ -1,103 +0,0 @@
-/*
- * 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

@@ -24,15 +24,13 @@ package com.dmdirc.parser.irc.outputqueue;
 
 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;
@@ -50,24 +48,10 @@ public class SimpleRateLimitedQueueHandler extends QueueHandler {
     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());
     }
 
     /**
@@ -164,7 +148,7 @@ public class SimpleRateLimitedQueueHandler extends QueueHandler {
     }
 
     @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();
@@ -179,7 +163,7 @@ public class SimpleRateLimitedQueueHandler extends QueueHandler {
                 // 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 {
@@ -193,21 +177,19 @@ public class SimpleRateLimitedQueueHandler extends QueueHandler {
             }
         }
 
-        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;
                     }
                 }
@@ -215,11 +197,14 @@ public class SimpleRateLimitedQueueHandler extends QueueHandler {
                 if (doSleep) {
                     try {
                         Thread.sleep(waitTime);
-                    } catch (InterruptedException ex) { /* Do Nothing. */ }
+                    } catch (InterruptedException ex) {
+                        /* Do Nothing. */
+                    }
                 }
             }
         } catch (InterruptedException ex) {
             // Do nothing
         }
     }
+
 }