abstract timeout waitForConfirms into a separate file

Oh dear, we forgot to remove ConfirmBase before.  We're using it for
something now, at least.

Author: Alexandru Scvortov

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

@@ -11,60 +11,40 @@
 //  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);
+    protected void waitForConfirms()
+        throws InterruptedException, TimeoutException
+    {
+        try {
+            FutureTask<?> waiter = new FutureTask<Object>(new Runnable() {
+                    public void run() {
+                        try {
+                            channel.waitForConfirmsOrDie();
+                        } catch (IOException e) {
+                            throw (ShutdownSignalException)e.getCause();
+                        } catch (InterruptedException e) {
+                            fail("test interrupted");
+                        }
                     }
-                }
-
-                public void handleNack(long seqNo, boolean multiple) {
-                    fail("got a nack");
-                }
-            });
-        channel.confirmSelect();
+                }, null);
+            (Executors.newSingleThreadExecutor()).execute(waiter);
+            waiter.get(10, TimeUnit.SECONDS);
+        } catch (ExecutionException e) {
+            throw (ShutdownSignalException)e.getCause();
+        }
     }
-
-    protected void waitAcks() throws InterruptedException {
-        while (unconfirmedSet.size() > 0)
-            Thread.sleep(10);
-    }
-
-    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());
-    }
-
 }

test/src/com/rabbitmq/client/test/functional/Confirm.java

@@ -24,20 +24,16 @@
 import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.MessageProperties;
 import com.rabbitmq.client.ShutdownSignalException;
-import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.client.test.ConfirmBase;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Map;
 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 Confirm extends BrokerTestCase
+public class Confirm extends ConfirmBase
 {
     private final static int NUM_MESSAGES = 1000;
 
@@ -318,26 +314,4 @@ protected void publish(String exchangeName, String queueName,
                                         : MessageProperties.BASIC,
                              "nop".getBytes());
     }
-
-    protected void waitForConfirms()
-        throws InterruptedException, TimeoutException
-    {
-        try {
-            FutureTask<?> waiter = new FutureTask<Object>(new Runnable() {
-                    public void run() {
-                        try {
-                            channel.waitForConfirmsOrDie();
-                        } catch (IOException e) {
-                            throw (ShutdownSignalException)e.getCause();
-                        } catch (InterruptedException e) {
-                            fail("test interrupted");
-                        }
-                    }
-                }, null);
-            (Executors.newSingleThreadExecutor()).execute(waiter);
-            waiter.get(10, TimeUnit.SECONDS);
-        } catch (ExecutionException e) {
-            throw (ShutdownSignalException)e.getCause();
-        }
-    }
 }

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

@@ -16,22 +16,25 @@
 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;
+import java.util.concurrent.TimeoutException;
 
-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 IOException, InterruptedException, TimeoutException
+    {
         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);