Merge default in.

Author: Steve Powell

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

@@ -94,7 +94,6 @@ private void scheduleShutdownProcessing() {
         final Set<CountDownLatch> sdSet = new HashSet<CountDownLatch>(shutdownSet);
         final ConsumerWorkService ssWorkService = workService;
         Thread shutdownThread = new Thread( new Runnable() {
-            @Override
             public void run() {
                 for (CountDownLatch latch : sdSet) {
                     try { latch.await(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (Throwable e) { }

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

@@ -1,7 +1,6 @@
 package com.rabbitmq.client.impl;
 
 import java.util.Collection;
-import java.util.Deque;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedList;
@@ -74,7 +73,7 @@ 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, Deque<W>> pool = new HashMap<K, Deque<W>>();
+        private final Map<K, LinkedList<W>> pool = new HashMap<K, LinkedList<W>>();
 
     /**
      * Add client <code><b>key</b></code> to pool of item queues, with an empty queue.
@@ -128,25 +127,25 @@ public K nextWorkBlock(Collection<W> to, int size) {
         synchronized (this.monitor) {
             K nextKey = readyToInProgress();
             if (nextKey != null) {
-                Deque<W> queue = this.pool.get(nextKey);
+                LinkedList<W> queue = this.pool.get(nextKey);
                 drainTo(queue, to, size);
             }
             return nextKey;
         }
     }
 
     /**
-     * Private implementation of <code><b>drainTo</b></code> (not implemented for <code><b>Deque&lt;W&gt;</b></code>s).
+     * Private implementation of <code><b>drainTo</b></code> (not implemented for <code><b>LinkedList&lt;W&gt;</b></code>s).
      * @param <W> element type
-     * @param deque to take (poll) elements from
+     * @param deList to take (poll) elements from
      * @param c to add elements to
-     * @param maxElements to take from deque
+     * @param maxElements to take from deList
      * @return number of elements actually taken
      */
-    private static <W> int drainTo(Deque<W> deque, Collection<W> c, int maxElements) {
+    private static <W> int drainTo(LinkedList<W> deList, Collection<W> c, int maxElements) {
         int n = 0;
         while (n < maxElements) {
-            W first = deque.poll();
+            W first = deList.poll();
             if (first == null)
                 break;
             c.add(first);
@@ -206,8 +205,8 @@ public boolean finishWorkBlock(K key) {
     }
 
     private boolean moreWorkItems(K key) {
-        Deque<W> deque = this.pool.get(key);
-        return (deque==null ? false : !deque.isEmpty());
+        LinkedList<W> leList = this.pool.get(key);
+        return (leList==null ? false : !leList.isEmpty());
     }
 
     /* State identification functions */

test/src/com/rabbitmq/client/test/Bug20004Test.java

@@ -28,9 +28,9 @@
  * an independent thread.
  */
 public class Bug20004Test extends BrokerTestCase {
-    public volatile Exception caughtException = null;
-    public volatile boolean completed = false;
-    public volatile boolean created = false;
+    private volatile Exception caughtException = null;
+    private volatile boolean completed = false;
+    private volatile boolean created = false;
 
     protected void releaseResources()
         throws IOException

test/src/com/rabbitmq/client/test/ConfirmBase.java

@@ -11,60 +11,51 @@
 //  The Original Code is RabbitMQ.
 //
 //  The Initial Developer of the Original Code is VMware, Inc.
-//  Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
+//  Copyright (c) 2011 VMware, Inc.  All rights reserved.
 //
 
 package com.rabbitmq.client.test;
 
-import com.rabbitmq.client.ConfirmListener;
-import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.ShutdownSignalException;
 
 import java.io.IOException;
-import java.util.Collections;
-import java.util.SortedSet;
-import java.util.TreeSet;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.TimeUnit;
 
-public class ConfirmBase extends BrokerTestCase {
-
-    protected SortedSet<Long> unconfirmedSet;
-
-    @Override
-    protected void setUp() throws IOException {
-        super.setUp();
-        unconfirmedSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
-        channel.addConfirmListener(new ConfirmListener() {
-                public void handleAck(long seqNo, boolean multiple) {
-                    if (!unconfirmedSet.contains(seqNo)) {
-                        fail("got duplicate ack: " + seqNo);
-                    }
-                    if (multiple) {
-                        unconfirmedSet.headSet(seqNo + 1).clear();
-                    } else {
-                        unconfirmedSet.remove(seqNo);
-                    }
-                }
+import junit.framework.AssertionFailedError;
 
-                public void handleNack(long seqNo, boolean multiple) {
-                    fail("got a nack");
-                }
-            });
-        channel.confirmSelect();
-    }
-
-    protected void waitAcks() throws InterruptedException {
-        while (unconfirmedSet.size() > 0)
-            Thread.sleep(10);
+public class ConfirmBase extends BrokerTestCase {
+    protected void waitForConfirms() throws Exception
+    {
+        waitForConfirms("ConfirmBase.waitForConfirms");
     }
 
-    protected void publish(String exchangeName, String queueName,
-                           boolean persistent, boolean mandatory,
-                           boolean immediate)
-        throws IOException {
-        unconfirmedSet.add(channel.getNextPublishSeqNo());
-        channel.basicPublish(exchangeName, queueName, mandatory, immediate,
-                             persistent ? MessageProperties.PERSISTENT_BASIC
-                                        : MessageProperties.BASIC,
-                             "nop".getBytes());
+    protected void waitForConfirms(final String testTitle) throws Exception
+    {
+        try {
+            FutureTask<?> waiter = new FutureTask<Object>(new Runnable() {
+                    public void run() {
+                        try {
+                            channel.waitForConfirmsOrDie();
+                        } catch (IOException e) {
+                            throw (ShutdownSignalException)e.getCause();
+                        } catch (InterruptedException _) {
+                            fail(testTitle + ": interrupted");
+                        }
+                    }
+                }, null);
+            (Executors.newSingleThreadExecutor()).execute(waiter);
+            waiter.get(10, TimeUnit.SECONDS);
+        } catch (ExecutionException ee) {
+            Throwable t = ee.getCause();
+            if (t instanceof ShutdownSignalException) throw (ShutdownSignalException) t;
+            if (t instanceof AssertionFailedError) throw (AssertionFailedError) t;
+            throw (Exception)t;
+        } catch (TimeoutException _) {
+            fail(testTitle + ": timeout");
+        }
     }
-
 }

test/src/com/rabbitmq/client/test/server/MessageRecovery.java

@@ -16,22 +16,22 @@
 package com.rabbitmq.client.test.server;
 
 import com.rabbitmq.client.MessageProperties;
-import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.client.test.ConfirmBase;
 
-import java.io.IOException;
-
-public class MessageRecovery extends BrokerTestCase
+public class MessageRecovery extends ConfirmBase
 {
 
     private final static String Q = "recovery-test";
 
-    public void testMessageRecovery() throws IOException, InterruptedException {
+    public void testMessageRecovery()
+        throws Exception
+    {
         channel.confirmSelect();
         channel.queueDeclare(Q, true, false, false, null);
         channel.basicPublish("", Q, false, false,
                              MessageProperties.PERSISTENT_BASIC,
                              "nop".getBytes());
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
 
         restart();
         assertDelivered(Q, 1);

test/src/com/rabbitmq/examples/ConfirmDontLoseMessages.java

@@ -18,12 +18,8 @@
 package com.rabbitmq.examples;
 
 import java.io.IOException;
-import java.util.Collections;
-import java.util.SortedSet;
-import java.util.TreeSet;
 
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.ConfirmListener;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.MessageProperties;
@@ -50,9 +46,6 @@ public static void main(String[] args)
     }
 
     static class Publisher implements Runnable {
-        private volatile SortedSet<Long> unconfirmedSet =
-            Collections.synchronizedSortedSet(new TreeSet<Long>());
-
         public void run() {
             try {
                 long startTime = System.currentTimeMillis();
@@ -61,43 +54,16 @@ public void run() {
                 Connection conn = connectionFactory.newConnection();
                 Channel ch = conn.createChannel();
                 ch.queueDeclare(QUEUE_NAME, true, false, false, null);
-                ch.addConfirmListener(new ConfirmListener() {
-                        public void handleAck(long seqNo, boolean multiple) {
-                            if (multiple) {
-                                unconfirmedSet.headSet(seqNo+1).clear();
-                            } else {
-                                unconfirmedSet.remove(seqNo);
-                            }
-                        }
-
-                        public void handleNack(long seqNo, boolean multiple) {
-                            int lost = 0;
-                            if (multiple) {
-                                SortedSet<Long> nackd =
-                                    unconfirmedSet.headSet(seqNo+1);
-                                lost = nackd.size();
-                                nackd.clear();
-                            } else {
-                                lost = 1;
-                                unconfirmedSet.remove(seqNo);
-                            }
-                            System.out.printf("Probably lost %d messages.\n",
-                                              lost);
-                        }
-                    });
                 ch.confirmSelect();
 
                 // Publish
                 for (long i = 0; i < msgCount; ++i) {
-                    unconfirmedSet.add(ch.getNextPublishSeqNo());
                     ch.basicPublish("", QUEUE_NAME,
                                     MessageProperties.PERSISTENT_BASIC,
                                     "nop".getBytes());
                 }
 
-                // Wait
-                while (unconfirmedSet.size() > 0)
-                    Thread.sleep(10);
+                ch.waitForConfirmsOrDie();
 
                 // Cleanup
                 ch.queueDelete(QUEUE_NAME);