Fix Order of QueuePriorities

OutputHandlers are now incharge of creating QueueItems and comparing them.
Parser now allows for replacing the OutputQueue entirely
Default OutputQueue allows replacing of the OutputHandler (by giving it a new QueueFactory that outputs the alternative QueueHandler)
Sending PING is now High Priority
QueueItems now sort by a unique itemNumber not time, as priority queue doesn't guarentee order for itesm of teh same priority created at the same time
Add a rate-limited OutputQueue

src/com/dmdirc/parser/common/QueuePriority.java

@@ -28,10 +28,10 @@ package com.dmdirc.parser.common;
  * @author shane
  */
 public enum QueuePriority {
-    /** Low priority. */
-    LOW,
+    /** High priority. */
+    HIGH,
     /** Normal priority. */
     NORMAL,
-    /** High Priority. */
-    HIGH;
+    /** Low Priority. */
+    LOW;
 }

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

@@ -294,11 +294,34 @@ public class IRCParser implements SecureParser, Runnable {
      * @param myDetails Client information.
      */
     public IRCParser(final MyInfo myDetails, final ServerInfo serverDetails) {
+        out = new OutputQueue();
         if (myDetails != null) { this.me = myDetails; }
         if (serverDetails != null) { this.server = serverDetails; }
         resetState();
     }
 
+    /**
+     * Get the current OutputQueue
+     *
+     * @return the current OutputQueue
+     */
+    public OutputQueue getOut() {
+        return out;
+    }
+
+    /**
+     * Set the OutputQueue
+     *
+     * @param out the new current OutputQueue
+     */
+    public void setOut(final OutputQueue out) throws IRCParserException {
+        if (currentSocketState == SocketState.CLOSED) {
+            this.out = out;
+        } else {
+            throw new IRCParserException("OutputQueue can only be changed when disconnected.");
+        }
+    }
+
     /**
      * Get the current Value of bindIP.
      *
@@ -710,7 +733,7 @@ public class IRCParser implements SecureParser, Runnable {
         }
 
         callDebugInfo(DEBUG_SOCKET, "\t-> Opening socket output stream PrintWriter");
-        out = new OutputQueue(socket.getOutputStream());
+        out.setOutputStream(socket.getOutputStream());
         out.setQueueEnabled(true);
         callDebugInfo(DEBUG_SOCKET, "\t-> Opening socket input stream BufferedReader");
         in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
@@ -725,7 +748,7 @@ public class IRCParser implements SecureParser, Runnable {
             sendString("PASS " + server.getPassword());
         }
         sendString("NICK " + me.getNickname());
-    thinkNickname = me.getNickname();
+        thinkNickname = me.getNickname();
         String localhost;
         try {
             localhost = InetAddress.getLocalHost().getHostAddress();
@@ -1853,7 +1876,7 @@ public class IRCParser implements SecureParser, Runnable {
                 setPingNeeded(true);
                 pingCountDown = pingCountDownLength;
                 lastPingValue = String.valueOf(System.currentTimeMillis());
-                if (sendString("PING " + lastPingValue)) {
+                if (sendString("PING " + lastPingValue, QueuePriority.HIGH)) {
                     callPingSent();
                 }
             }

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

@@ -0,0 +1,27 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+
+package com.dmdirc.parser.irc;
+
+/**
+ * IRC Parser Exception!
+ *
+ * @author shane
+ */
+class IRCParserException extends Exception {
+
+    /** Version of this class. */
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * Create a new IRCParserException
+     *
+     * @param message Reason for exception
+     */
+    public IRCParserException(final String message) {
+        super(message);
+    }
+
+}

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

@@ -35,7 +35,7 @@ import java.util.concurrent.PriorityBlockingQueue;
  */
 public class OutputQueue {
     /** PrintWriter for sending output. */
-    private PrintWriter out;
+    private PrintWriter out = null;
 
     /** Is queueing enabled? */
     private boolean queueEnabled = true;
@@ -44,14 +44,23 @@ public class OutputQueue {
     private BlockingQueue<QueueItem> queue = new PriorityBlockingQueue<QueueItem>();
 
     /** Thread for the sending queue. */
-    private Thread queueThread;
+    private QueueHandler queueHandler;
+
+    /** The QueueFactory for this OutputQueue. */
+    private QueueFactory queueFactory = PriorityQueueHandler.getFactory();
+    // private QueueFactory queueFactory = SimpleRateLimitedQueueHandler.getFactory();
 
     /**
      * Create a new OutputQueue
-     * 
-     * @param outputStream PrintWriter to use to actually send stuff.
      */
-    public OutputQueue (final OutputStream outputStream) {
+    public OutputQueue() { }
+
+    /**
+     * Set the output stream for this queue.
+     *
+     * @param outputStream Output Stream to use.
+     */
+    public void setOutputStream(final OutputStream outputStream) {
         this.out = new PrintWriter(outputStream, true);
     }
 
@@ -64,6 +73,27 @@ public class OutputQueue {
         return queueEnabled;
     }
 
+    /**
+     * Set the QueueFactory.
+     * Changing this will not change an existing QueueHandler unless queueing is
+     * disabled and reenabled.
+     * If this is called before the first lien of output is queued then there is
+     * no need to disable and reenable the queue.
+     *
+     * @param manager New QueueFactory to use.
+     */
+    public void setQueueManager(final QueueFactory manager) {
+        queueFactory = manager;
+    }
+
+    /**
+     * Get the QueueFactory.
+     * @return The current QueueFactory.
+     */
+    public QueueFactory getQueueManager() {
+        return queueFactory;
+    }
+
     /**
      * Set if queueing is enabled.
      * if this is changed from enabled to disabled, all currently queued items
@@ -72,12 +102,15 @@ public class OutputQueue {
      * @param queueEnabled new value for queueEnabled
      */
     public void setQueueEnabled(final boolean queueEnabled) {
+        if (out == null) {
+            throw new NullPointerException("No output stream has been set.");
+        }
         final boolean old = this.queueEnabled;
         this.queueEnabled = queueEnabled;
 
         if (old != queueEnabled && old) {
-            queueThread.interrupt();
-            queueThread = null;
+            queueHandler.interrupt();
+            queueHandler = null;
 
             while (!queue.isEmpty()) {
                 try {
@@ -94,14 +127,23 @@ public class OutputQueue {
      */
     public void clearQueue() {
         this.queueEnabled = false;
-        if (queueThread != null) {
-            queueThread.interrupt();
-            queueThread = null;
+        if (queueHandler != null) {
+            queueHandler.interrupt();
+            queueHandler = null;
         }
 
         queue.clear();
     }
 
+    /**
+     * Get the number of items currently in the queue.
+     *
+     * @return Number of items in the queue.
+     */
+    public int queueCount() {
+        return queue.size();
+    }
+
     /**
      * Send the given line.
      * If queueing is enabled, this will queue it, else it will send it
@@ -122,26 +164,18 @@ public class OutputQueue {
      * @param priority Priority of item (ignored if queue is disabled)
      */
     public void sendLine(final String line, final QueuePriority priority) {
+        if (out == null) {
+            throw new NullPointerException("No output stream has been set.");
+        }
         if (queueEnabled) {
-            queue.add(new QueueItem(line, priority));
-
-            if (queueThread == null || !queueThread.isAlive()) {
-                queueThread = getQueueHandler(queue, out);
-                queueThread.start();
+            if (queueHandler == null || !queueHandler.isAlive()) {
+                queueHandler = queueFactory.getQueueHandler(this, queue, out);
+                queueHandler.start();
             }
+
+            queue.add(queueHandler.getQueueItem(line, priority));
         } else {
             out.printf("%s\r\n", line);
         }
     }
-
-    /**
-     * Get a new QueueHandler instance as needed.
-     * 
-     * @param queue The queue to handle.
-     * @param out Where to send crap.
-     * @return the new queue handler object.
-     */
-    private QueueHandler getQueueHandler(final BlockingQueue<QueueItem> queue, final PrintWriter out) {
-        return new PriorityQueueHandler(this, queue, out);
-    }
 }

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

@@ -31,8 +31,28 @@ import java.util.concurrent.BlockingQueue;
  * @author shane
  */
 public class PriorityQueueHandler extends QueueHandler {
+    /**
+     * Get a QueueFactory that produces PriorityQueueHandlers
+     *
+     * @return a QueueFactory that produces PrirortyQueueHandlers.
+     */
+    public static QueueFactory getFactory() {
+        return new QueueFactory(){
+            /** {@inheritDoc} */
+            @Override
+            public QueueHandler getQueueHandler(final OutputQueue outputQueue, final BlockingQueue<QueueItem> queue, final PrintWriter out) {
+                return new PriorityQueueHandler(outputQueue, queue, out);
+            }
+        };
+    }
 
-
+    /**
+     * Create a new PriorityQueueHandler
+     *
+     * @param outputQueue Owner of this Queue Handler
+     * @param queue Queue to use
+     * @param out Output Stream to use
+     */
     public PriorityQueueHandler(final OutputQueue outputQueue, final BlockingQueue<QueueItem> queue, final PrintWriter out) {
         super(outputQueue, queue, out);
     }

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

@@ -0,0 +1,28 @@
+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+
+package com.dmdirc.parser.irc.outputqueue;
+
+import java.io.PrintWriter;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * A QueueFactory produces QueueHandlers for OutputQueue.
+ *
+ * @author shane
+ */
+public interface QueueFactory {
+
+    /**
+     * Get a new QueueHandler instance as needed.
+     *
+     * @param outputQueue the OutputQueue that will own this QueueHandler
+     * @param queue The queue to handle.
+     * @param out Where to send crap.
+     * @return the new queue handler object.
+     */
+    public QueueHandler getQueueHandler(final OutputQueue outputQueue, final BlockingQueue<QueueItem> queue, final PrintWriter out);
+
+}

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

@@ -22,6 +22,7 @@
 
 package com.dmdirc.parser.irc.outputqueue;
 
+import com.dmdirc.parser.common.QueuePriority;
 import java.io.PrintWriter;
 import java.util.concurrent.BlockingQueue;
 
@@ -62,7 +63,61 @@ public abstract class QueueHandler extends Thread {
         out.printf("%s\r\n", line);
     }
 
-    /** {@inheritDoc} */
+    /**
+     * 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(this, line, priority);
+    }
+
+    /**
+     * Compare two QueueItems for sorting purposes.
+     * This is called by the default QueueItem in its compareTo method. The
+     * calling object will be the first parameter, the object to compare it to
+     * will be second.
+     * This allows QueueHandlers to sort items differently if needed.
+     *
+     * The default implementation works as follows:
+     * Compare based on priorty firstly, if the priorities are the same,
+     * compare based on the order the items were added to the queue.
+     *
+     * If an item has been in the queue longer than 10 seconds, it will not
+     * check its priority and soley position itself based on adding order.
+     *
+     * @param mainObject Main object we are comparing against.
+     * @param otherObject Object we are comparing to.
+     * @return A QueueItem for teh given parameters
+     */
+    public int compareQueueItem(final QueueItem mainObject, final QueueItem otherObject) {
+        if (mainObject.getTime() < 10 * 1000 && mainObject.getPriority().compareTo(otherObject.getPriority()) != 0) {
+            return mainObject.getPriority().compareTo(otherObject.getPriority());
+        }
+
+        if (mainObject.getItemNumber() > otherObject.getItemNumber()) {
+            return 1;
+        } else if (mainObject.getItemNumber() < otherObject.getItemNumber()) {
+            return -1;
+        } else {
+            // This can't happen.
+            return 0;
+        }
+    }
+
+    /**
+     * 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();
 }

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

@@ -30,15 +30,24 @@ import com.dmdirc.parser.common.QueuePriority;
  * @author shane
  */
 public class QueueItem implements Comparable<QueueItem> {
+    /** Global Item Number*/
+    private static long number = 0L;
+
     /** Line to send */
     private final String line;
 
     /** Time this line was added. */
     private final long time;
 
+    /** Item Number */
+    private final long itemNumber;
+
     /** What is the priority of this line? */
     private final QueuePriority priority;
 
+    /** Our handler. */
+    private final QueueHandler handler;
+
     /**
      * Get the value of line
      *
@@ -57,6 +66,15 @@ public class QueueItem implements Comparable<QueueItem> {
         return time;
     }
 
+    /**
+     * Get the number of this item
+     *
+     * @return the value of itemNumber
+     */
+    public long getItemNumber() {
+        return itemNumber;
+    }
+
     /**
      * Get the value of priority
      *
@@ -67,42 +85,31 @@ public class QueueItem implements Comparable<QueueItem> {
     }
 
     /**
-     * Create a new QueuePriority
+     * Create a new QueueItem
      *
+     * @param handler Handler for this QueueItem
      * @param line Line to send
      * @param priority
      */
-    public QueueItem(final String line, final QueuePriority priority) {
+    public QueueItem(final QueueHandler handler, final String line, final QueuePriority priority) {
+        this.handler = handler;
         this.line = line;
         this.priority = priority;
 
         this.time = System.currentTimeMillis();
+        this.itemNumber = number++;
     }
 
     /**
      * Compare objects.
-     * Compare based on priorty firstly, if the priorities are the same,
-     * compare based on time added.
+     * This will use the compareQueueItem method of the current QueueHandler.
      *
-     * If an item has been in the queue longer than 10 seconds, it will not
-     * check its priority and soley position itself based on time.
-     * 
      * @param o Object to compare to
      * @return Position of this item in reference to the given item.
      */
     @Override
     public int compareTo(final QueueItem o) {
-        if (this.getTime() < 10 * 1000 && this.getPriority().compareTo(o.getPriority()) > 0) {
-            return 1;
-        }
-
-        if (this.getTime() < o.getTime()) {
-            return 1;
-        } else if (this.getTime() > o.getTime()) {
-            return -1;
-        } else {
-            return 0;
-        }
+        return handler.compareQueueItem(this, o);
     }
 
     /** {@inheritDoc} */

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

@@ -0,0 +1,258 @@
+/*
+ *  Copyright (c) 2006-2009 Chris Smith, Shane Mc Cormack, Gregory Holmes
+ * 
+ *  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.concurrent.BlockingQueue;
+
+/**
+ * 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.
+ *
+ * @author shane
+ */
+public class SimpleRateLimitedQueueHandler extends QueueHandler {
+    /**
+     * Get a QueueFactory that produces PriorityQueueHandlers
+     *
+     * @return a QueueFactory that produces PrirortyQueueHandlers.
+     */
+    public static QueueFactory getFactory() {
+        return new QueueFactory(){
+            /** {@inheritDoc} */
+            @Override
+            public QueueHandler getQueueHandler(final OutputQueue outputQueue, final BlockingQueue<QueueItem> queue, final PrintWriter out) {
+                return new SimpleRateLimitedQueueHandler(outputQueue, queue, out);
+            }
+        };
+    }
+
+    /** Current count. */
+    private int count = 0;
+
+    /** Time last item was added. */
+    private long lastItemTime = 0L;
+
+    /** Are we limiting? */
+    private boolean isLimiting = false;
+
+    /** How many items are allowed before limiting? */
+    private int items = 4;
+
+    /** How many microseconds do we care about when checking for items? */
+    private int limitTime = 4000;
+
+    /** How long to wait in between each item when limiting? */
+    private int waitTime = 3000;
+
+    /** Always update the lastItemTime or only if its been > limitTime? */
+    private boolean alwaysUpdateTime = true;
+
+    /**
+     * Create a new SimpleRateLimitedQueueHandler
+     *
+     * @param outputQueue Owner of this Queue Handler
+     * @param queue Queue to use
+     * @param out Output Stream to use
+     */
+    public SimpleRateLimitedQueueHandler(final OutputQueue outputQueue, final BlockingQueue<QueueItem> queue, final PrintWriter out) {
+        super(outputQueue, queue, out);
+    }
+
+    /**
+     * Get the number of items needed to activate rate limiting.
+     * 
+     * @return Number of items needed to activate rate limiting.
+     */
+    public int getItems() {
+        return items;
+    }
+
+    /**
+     * Set the number of items needed to activate rate limiting.
+     *
+     * @param items Number of items needed to activate rate limiting.
+     */
+    public void setItems(final int items) {
+        this.items = items;
+    }
+
+    /**
+     * Get the length of time that is used when checking for rate limiting. (If
+     * more than getItems() number of lines are added less that this time apart
+     * from each other then rate limiting is activated.)
+     *
+     * @return Number of items needed to activate rate limiting.
+     */
+    public int getLimitTime() {
+        return limitTime;
+    }
+
+    /**
+     * Set the length of time that is used when checking for rate limiting. (If
+     * more than getItems() number of lines are added less that this time apart
+     * from each other then rate limiting is activated.)
+     *
+     * @param limitTime Number of items needed to activate rate limiting.
+     */
+    public void setLimitTime(final int limitTime) {
+        this.limitTime = limitTime;
+    }
+
+    /**
+     * Get the length of time that we wait inbetween lines when limiting.
+     *
+     * @return length of time that we wait inbetween lines when limiting.
+     */
+    public int getWaitTime() {
+        return waitTime;
+    }
+
+    /**
+     * Set the length of time that we wait inbetween lines when limiting.
+     *
+     * @param waitTime length of time that we wait inbetween lines when limiting.
+     */
+    public void setWaitTime(final int waitTime) {
+        this.waitTime = waitTime;
+    }
+
+    /**
+     * Will the internal "lastItemTime" be updated every time an item is added,
+     * or only after limitTime has passed?
+     *
+     * If true, assuming the default settings) items sent at 0, 3, 6, 9 will
+     * activate rate limiting, if false it would need to be 0, 1, 2, 3.
+     *
+     * @return is LastItemTime always updated?
+     */
+    public boolean getAlwaysUpdateTime() {
+        return alwaysUpdateTime;
+    }
+
+    /**
+     * Set if the internal "lastItemTime" should be updated every time an item
+     * is added, or only after limitTime has passed?
+     *
+     * If true, assuming the default settings) items sent at 0, 3, 6, 9 will
+     * activate rate limiting, if false it would need to be 0, 1, 2, 3.
+     * 
+     * @param alwaysUpdateTime Should LastItemTime always updated?
+     */
+    public void setAlwaysUpdateTime(final boolean alwaysUpdateTime) {
+        this.alwaysUpdateTime = alwaysUpdateTime;
+    }
+
+    /**
+     * Are we currently limiting?
+     * 
+     * @return True if limiting is active.
+     */
+    public boolean isLimiting() {
+        return isLimiting;
+    }
+
+    /**
+     * Compare queue items, if priorities differ, then  higher priority items
+     * will always be put further ahead in the queue (This queue ignores the
+     * 10-second rule of the normal queue) otherwise the normal comparison is
+     * used.
+     */
+    @Override
+    public int compareQueueItem(final QueueItem mainObject, final QueueItem otherObject) {
+        if (mainObject.getPriority().compareTo(otherObject.getPriority()) != 0) {
+            return mainObject.getPriority().compareTo(otherObject.getPriority());
+        } else {
+            return super.compareQueueItem(mainObject, otherObject);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public QueueItem getQueueItem(String line, QueuePriority priority) {
+        // Was the last line added less than limitTime ago?
+        synchronized (this) {
+            final boolean overTime = (lastItemTime + limitTime > System.currentTimeMillis());
+            if (overTime) {
+                // If we are not currently limiting, and this is the items-th item
+                // added in the last limitTime, start limiting.
+                if (!isLimiting) {
+                    if (++count > (items - 1)) {
+                        System.out.println("++ Begin Limiting");
+                        isLimiting = true;
+                        count = 0;
+                    }
+                }
+            } else if (!isLimiting) {
+                // If it has been more than limitTime seconds since the last line
+                // and we are not currently limiting, reset the count.
+                count = 0;
+            } else {
+                // 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 (queue.size() == 0) {
+                    isLimiting = false;
+                }
+            }
+            if (alwaysUpdateTime || overTime) {
+                lastItemTime = System.currentTimeMillis();
+            }
+        }
+        
+        return super.getQueueItem(line, priority);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public void run() {
+        try {
+            while (outputQueue.isQueueEnabled()) {
+                final QueueItem item = queue.take();
+                
+                sendLine(item.getLine());
+
+                final boolean doSleep;
+                synchronized (this) {
+                    doSleep = isLimiting;
+                    if (isLimiting) {
+                        if (queue.size() == 0) {
+                            isLimiting = false;
+                        }
+                    }
+                }
+
+                if (doSleep) {
+                    try {
+                        Thread.sleep(waitTime);
+                    } catch (InterruptedException ex) { /* Do Nothing. */ }
+                }
+            }
+        } catch (InterruptedException ex) {
+            // Do nothing
+        }
+    }
+}