Naively use a maximum-length queue in the WorkPool.

Author: Simon MacMullen

src/com/rabbitmq/client/impl/WorkPool.java

@@ -3,10 +3,10 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedList;
 import java.util.Map;
-import java.util.Queue;
 import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * This is a generic implementation of the <q>Channels</q> specification
@@ -75,7 +75,9 @@ public class WorkPool<K, W> {
         /** The set of clients which have work <i>in progress</i>. */
         private final Set<K> inProgress = new HashSet<K>();
         /** The pool of registered clients, with their work queues. */
-        private final Map<K, LinkedList<W>> pool = new HashMap<K, LinkedList<W>>();
+        private final Map<K, BlockingQueue<W>> pool = new HashMap<K, BlockingQueue<W>>();
+
+    private int MAX_QUEUE_LENGTH = 1000;
 
     /**
      * Add client <code><b>key</b></code> to pool of item queues, with an empty queue.
@@ -87,7 +89,7 @@ public class WorkPool<K, W> {
     public void registerKey(K key) {
         synchronized (this.monitor) {
             if (!this.pool.containsKey(key)) {
-                this.pool.put(key, new LinkedList<W>());
+                this.pool.put(key, new LinkedBlockingQueue<W>(MAX_QUEUE_LENGTH));
             }
         }
     }
@@ -129,7 +131,7 @@ public K nextWorkBlock(Collection<W> to, int size) {
         synchronized (this.monitor) {
             K nextKey = readyToInProgress();
             if (nextKey != null) {
-                LinkedList<W> queue = this.pool.get(nextKey);
+                BlockingQueue<W> queue = this.pool.get(nextKey);
                 drainTo(queue, to, size);
             }
             return nextKey;
@@ -144,7 +146,7 @@ public K nextWorkBlock(Collection<W> to, int size) {
      * @param maxElements to take from deList
      * @return number of elements actually taken
      */
-    private static <W> int drainTo(LinkedList<W> deList, Collection<W> c, int maxElements) {
+    private static <W> int drainTo(BlockingQueue<W> deList, Collection<W> c, int maxElements) {
         int n = 0;
         while (n < maxElements) {
             W first = deList.poll();
@@ -167,9 +169,14 @@ private static <W> int drainTo(LinkedList<W> deList, Collection<W> c, int maxEle
      */
     public boolean addWorkItem(K key, W item) {
         synchronized (this.monitor) {
-            Queue<W> queue = this.pool.get(key);
+            BlockingQueue<W> queue = this.pool.get(key);
             if (queue != null) {
-                queue.offer(item);
+                try {
+                    queue.put(item);
+                } catch (InterruptedException e) {
+                    // ok
+                }
+
                 if (isDormant(key)) {
                     dormantToReady(key);
                     return true;
@@ -205,7 +212,7 @@ public boolean finishWorkBlock(K key) {
     }
 
     private boolean moreWorkItems(K key) {
-        LinkedList<W> leList = this.pool.get(key);
+        BlockingQueue<W> leList = this.pool.get(key);
         return (leList==null ? false : !leList.isEmpty());
     }