timeout waitForConfirms after 10s

Alexandru Scvortov · Alexandru Scvortov · commit 28efe08d96d0 · 2011-07-29T16:21:24.000+01:00
diff --git a/test/src/com/rabbitmq/client/test/functional/Confirm.java b/test/src/com/rabbitmq/client/test/functional/Confirm.java
@@ -26,12 +26,16 @@
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.test.BrokerTestCase;
 
-
 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
 {
@@ -65,48 +69,49 @@ protected void setUp() throws IOException {
     }
 
     public void testTransient()
-        throws IOException, InterruptedException {
+        throws IOException, InterruptedException, TimeoutException
+    {
         confirmTest("", "confirm-test", false, false, false);
     }
 
     public void testPersistentSimple()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         confirmTest("", "confirm-test", true, false, false);
     }
 
     public void testNonDurable()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         confirmTest("", "confirm-test-nondurable", true, false, false);
     }
 
     public void testPersistentImmediate()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         confirmTest("", "confirm-test", true, false, true);
     }
 
     public void testPersistentImmediateNoConsumer()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         confirmTest("", "confirm-test-noconsumer", true, false, true);
     }
 
     public void testPersistentMandatory()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         confirmTest("", "confirm-test", true, true, false);
     }
 
     public void testPersistentMandatoryReturn()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         confirmTest("", "confirm-test-doesnotexist", true, true, false);
     }
 
     public void testMultipleQueues()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         confirmTest("amq.direct", "confirm-multiple-queues",
                     true, false, false);
@@ -118,43 +123,43 @@ public void testMultipleQueues()
      * internal_sync that notifies the clients. */
 
     public void testQueueDelete()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         publishN("","confirm-test-noconsumer", true, false, false);
 
         channel.queueDelete("confirm-test-noconsumer");
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
     }
 
     public void testQueuePurge()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         publishN("", "confirm-test-noconsumer", true, false, false);
 
         channel.queuePurge("confirm-test-noconsumer");
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
     }
 
     public void testBasicReject()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         basicRejectCommon(false);
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
     }
 
     public void testQueueTTL()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         publishN("", "confirm-ttl", true, false, false);
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
     }
 
     public void testBasicRejectRequeue()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         basicRejectCommon(true);
 
@@ -164,11 +169,11 @@ public void testBasicRejectRequeue()
         channel.basicConsume("confirm-test-noconsumer", true,
                              new DefaultConsumer(channel));
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
     }
 
     public void testBasicRecover()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         publishN("", "confirm-test-noconsumer", true, false, false);
 
@@ -186,7 +191,7 @@ public void testBasicRecover()
         channel.basicConsume("confirm-test-noconsumer", true,
                              new DefaultConsumer(channel));
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
     }
 
     public void testSelect()
@@ -212,7 +217,7 @@ public void testSelect()
     }
 
     public void testWaitForConfirms()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         final SortedSet<Long> unconfirmedSet =
             Collections.synchronizedSortedSet(new TreeSet<Long>());
@@ -238,28 +243,28 @@ public void handleNack(long seqNo, boolean multiple) {
             publish("", "confirm-test", true, false, false);
         }
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
         if (!unconfirmedSet.isEmpty()) {
             fail("waitForConfirms returned with unconfirmed messages");
         }
     }
 
     public void testWaitForConfirmsNoOp()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         channel = connection.createChannel();
         // Don't enable Confirm mode
         publish("", "confirm-test", true, false, false);
-        channel.waitForConfirmsOrDie(); // Nop
+        waitForConfirms(); // Nop
     }
 
     public void testWaitForConfirmsException()
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         publishN("", "confirm-test", true, false, false);
         channel.close();
         try {
-            channel.waitForConfirmsOrDie();
+            waitForConfirms();
             fail("waitAcks worked on a closed channel");
         } catch (ShutdownSignalException sse) {
             if (!(sse.getReason() instanceof AMQP.Channel.Close))
@@ -274,11 +279,11 @@ public void testWaitForConfirmsException()
     public void confirmTest(String exchange, String queueName,
                             boolean persistent, boolean mandatory,
                             boolean immediate)
-        throws IOException, InterruptedException
+        throws IOException, InterruptedException, TimeoutException
     {
         publishN(exchange, queueName, persistent, mandatory, immediate);
 
-        channel.waitForConfirmsOrDie();
+        waitForConfirms();
     }
 
     private void publishN(String exchangeName, String queueName,
@@ -313,4 +318,26 @@ 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();
+        }
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -26,12 +26,16 @@`
`26`	`26`	`import com.rabbitmq.client.ShutdownSignalException;`
`27`	`27`	`import com.rabbitmq.client.test.BrokerTestCase;`
`28`	`28`
`29`		`-`
`30`	`29`	`import java.io.IOException;`
`31`	`30`	`import java.util.Collections;`
`32`	`31`	`import java.util.Map;`
`33`	`32`	`import java.util.SortedSet;`
`34`	`33`	`import java.util.TreeSet;`
	`34`	`+import java.util.concurrent.ExecutionException;`
	`35`	`+import java.util.concurrent.Executors;`
	`36`	`+import java.util.concurrent.FutureTask;`
	`37`	`+import java.util.concurrent.TimeoutException;`
	`38`	`+import java.util.concurrent.TimeUnit;`
`35`	`39`
`36`	`40`	`public class Confirm extends BrokerTestCase`
`37`	`41`	`{`
`@@ -65,48 +69,49 @@ protected void setUp() throws IOException {`
`65`	`69`	`}`
`66`	`70`
`67`	`71`	`public void testTransient()`
`68`		`- throws IOException, InterruptedException {`
	`72`	`+ throws IOException, InterruptedException, TimeoutException`
	`73`	`+ {`
`69`	`74`	`confirmTest("", "confirm-test", false, false, false);`
`70`	`75`	`}`
`71`	`76`
`72`	`77`	`public void testPersistentSimple()`
`73`		`- throws IOException, InterruptedException`
	`78`	`+ throws IOException, InterruptedException, TimeoutException`
`74`	`79`	`{`
`75`	`80`	`confirmTest("", "confirm-test", true, false, false);`
`76`	`81`	`}`
`77`	`82`
`78`	`83`	`public void testNonDurable()`
`79`		`- throws IOException, InterruptedException`
	`84`	`+ throws IOException, InterruptedException, TimeoutException`
`80`	`85`	`{`
`81`	`86`	`confirmTest("", "confirm-test-nondurable", true, false, false);`
`82`	`87`	`}`
`83`	`88`
`84`	`89`	`public void testPersistentImmediate()`
`85`		`- throws IOException, InterruptedException`
	`90`	`+ throws IOException, InterruptedException, TimeoutException`
`86`	`91`	`{`
`87`	`92`	`confirmTest("", "confirm-test", true, false, true);`
`88`	`93`	`}`
`89`	`94`
`90`	`95`	`public void testPersistentImmediateNoConsumer()`
`91`		`- throws IOException, InterruptedException`
	`96`	`+ throws IOException, InterruptedException, TimeoutException`
`92`	`97`	`{`
`93`	`98`	`confirmTest("", "confirm-test-noconsumer", true, false, true);`
`94`	`99`	`}`
`95`	`100`
`96`	`101`	`public void testPersistentMandatory()`
`97`		`- throws IOException, InterruptedException`
	`102`	`+ throws IOException, InterruptedException, TimeoutException`
`98`	`103`	`{`
`99`	`104`	`confirmTest("", "confirm-test", true, true, false);`
`100`	`105`	`}`
`101`	`106`
`102`	`107`	`public void testPersistentMandatoryReturn()`
`103`		`- throws IOException, InterruptedException`
	`108`	`+ throws IOException, InterruptedException, TimeoutException`
`104`	`109`	`{`
`105`	`110`	`confirmTest("", "confirm-test-doesnotexist", true, true, false);`
`106`	`111`	`}`
`107`	`112`
`108`	`113`	`public void testMultipleQueues()`
`109`		`- throws IOException, InterruptedException`
	`114`	`+ throws IOException, InterruptedException, TimeoutException`
`110`	`115`	`{`
`111`	`116`	`confirmTest("amq.direct", "confirm-multiple-queues",`
`112`	`117`	`true, false, false);`
`@@ -118,43 +123,43 @@ public void testMultipleQueues()`
`118`	`123`	`* internal_sync that notifies the clients. */`
`119`	`124`
`120`	`125`	`public void testQueueDelete()`
`121`		`- throws IOException, InterruptedException`
	`126`	`+ throws IOException, InterruptedException, TimeoutException`
`122`	`127`	`{`
`123`	`128`	`publishN("","confirm-test-noconsumer", true, false, false);`
`124`	`129`
`125`	`130`	`channel.queueDelete("confirm-test-noconsumer");`
`126`	`131`
`127`		`- channel.waitForConfirmsOrDie();`
	`132`	`+ waitForConfirms();`
`128`	`133`	`}`
`129`	`134`
`130`	`135`	`public void testQueuePurge()`
`131`		`- throws IOException, InterruptedException`
	`136`	`+ throws IOException, InterruptedException, TimeoutException`
`132`	`137`	`{`
`133`	`138`	`publishN("", "confirm-test-noconsumer", true, false, false);`
`134`	`139`
`135`	`140`	`channel.queuePurge("confirm-test-noconsumer");`
`136`	`141`
`137`		`- channel.waitForConfirmsOrDie();`
	`142`	`+ waitForConfirms();`
`138`	`143`	`}`
`139`	`144`
`140`	`145`	`public void testBasicReject()`
`141`		`- throws IOException, InterruptedException`
	`146`	`+ throws IOException, InterruptedException, TimeoutException`
`142`	`147`	`{`
`143`	`148`	`basicRejectCommon(false);`
`144`	`149`
`145`		`- channel.waitForConfirmsOrDie();`
	`150`	`+ waitForConfirms();`
`146`	`151`	`}`
`147`	`152`
`148`	`153`	`public void testQueueTTL()`
`149`		`- throws IOException, InterruptedException`
	`154`	`+ throws IOException, InterruptedException, TimeoutException`
`150`	`155`	`{`
`151`	`156`	`publishN("", "confirm-ttl", true, false, false);`
`152`	`157`
`153`		`- channel.waitForConfirmsOrDie();`
	`158`	`+ waitForConfirms();`
`154`	`159`	`}`
`155`	`160`
`156`	`161`	`public void testBasicRejectRequeue()`
`157`		`- throws IOException, InterruptedException`
	`162`	`+ throws IOException, InterruptedException, TimeoutException`
`158`	`163`	`{`
`159`	`164`	`basicRejectCommon(true);`
`160`	`165`
`@@ -164,11 +169,11 @@ public void testBasicRejectRequeue()`
`164`	`169`	`channel.basicConsume("confirm-test-noconsumer", true,`
`165`	`170`	`new DefaultConsumer(channel));`
`166`	`171`
`167`		`- channel.waitForConfirmsOrDie();`
	`172`	`+ waitForConfirms();`
`168`	`173`	`}`
`169`	`174`
`170`	`175`	`public void testBasicRecover()`
`171`		`- throws IOException, InterruptedException`
	`176`	`+ throws IOException, InterruptedException, TimeoutException`
`172`	`177`	`{`
`173`	`178`	`publishN("", "confirm-test-noconsumer", true, false, false);`
`174`	`179`
`@@ -186,7 +191,7 @@ public void testBasicRecover()`
`186`	`191`	`channel.basicConsume("confirm-test-noconsumer", true,`
`187`	`192`	`new DefaultConsumer(channel));`
`188`	`193`
`189`		`- channel.waitForConfirmsOrDie();`
	`194`	`+ waitForConfirms();`
`190`	`195`	`}`
`191`	`196`
`192`	`197`	`public void testSelect()`
`@@ -212,7 +217,7 @@ public void testSelect()`
`212`	`217`	`}`
`213`	`218`
`214`	`219`	`public void testWaitForConfirms()`
`215`		`- throws IOException, InterruptedException`
	`220`	`+ throws IOException, InterruptedException, TimeoutException`
`216`	`221`	`{`
`217`	`222`	`final SortedSet<Long> unconfirmedSet =`
`218`	`223`	`Collections.synchronizedSortedSet(new TreeSet<Long>());`
`@@ -238,28 +243,28 @@ public void handleNack(long seqNo, boolean multiple) {`
`238`	`243`	`publish("", "confirm-test", true, false, false);`
`239`	`244`	`}`
`240`	`245`
`241`		`- channel.waitForConfirmsOrDie();`
	`246`	`+ waitForConfirms();`
`242`	`247`	`if (!unconfirmedSet.isEmpty()) {`
`243`	`248`	`fail("waitForConfirms returned with unconfirmed messages");`
`244`	`249`	`}`
`245`	`250`	`}`
`246`	`251`
`247`	`252`	`public void testWaitForConfirmsNoOp()`
`248`		`- throws IOException, InterruptedException`
	`253`	`+ throws IOException, InterruptedException, TimeoutException`
`249`	`254`	`{`
`250`	`255`	`channel = connection.createChannel();`
`251`	`256`	`// Don't enable Confirm mode`
`252`	`257`	`publish("", "confirm-test", true, false, false);`
`253`		`- channel.waitForConfirmsOrDie(); // Nop`
	`258`	`+ waitForConfirms(); // Nop`
`254`	`259`	`}`
`255`	`260`
`256`	`261`	`public void testWaitForConfirmsException()`
`257`		`- throws IOException, InterruptedException`
	`262`	`+ throws IOException, InterruptedException, TimeoutException`
`258`	`263`	`{`
`259`	`264`	`publishN("", "confirm-test", true, false, false);`
`260`	`265`	`channel.close();`
`261`	`266`	`try {`
`262`		`- channel.waitForConfirmsOrDie();`
	`267`	`+ waitForConfirms();`
`263`	`268`	`fail("waitAcks worked on a closed channel");`
`264`	`269`	`} catch (ShutdownSignalException sse) {`
`265`	`270`	`if (!(sse.getReason() instanceof AMQP.Channel.Close))`
`@@ -274,11 +279,11 @@ public void testWaitForConfirmsException()`
`274`	`279`	`public void confirmTest(String exchange, String queueName,`
`275`	`280`	`boolean persistent, boolean mandatory,`
`276`	`281`	`boolean immediate)`
`277`		`- throws IOException, InterruptedException`
	`282`	`+ throws IOException, InterruptedException, TimeoutException`
`278`	`283`	`{`
`279`	`284`	`publishN(exchange, queueName, persistent, mandatory, immediate);`
`280`	`285`
`281`		`- channel.waitForConfirmsOrDie();`
	`286`	`+ waitForConfirms();`
`282`	`287`	`}`
`283`	`288`
`284`	`289`	`private void publishN(String exchangeName, String queueName,`
`@@ -313,4 +318,26 @@ protected void publish(String exchangeName, String queueName,`
`313`	`318`	`: MessageProperties.BASIC,`
`314`	`319`	`"nop".getBytes());`
`315`	`320`	`}`
	`321`	`+`
	`322`	`+ protected void waitForConfirms()`
	`323`	`+ throws InterruptedException, TimeoutException`
	`324`	`+ {`
	`325`	`+ try {`
	`326`	`+ FutureTask<?> waiter = new FutureTask<Object>(new Runnable() {`
	`327`	`+ public void run() {`
	`328`	`+ try {`
	`329`	`+ channel.waitForConfirmsOrDie();`
	`330`	`+ } catch (IOException e) {`
	`331`	`+ throw (ShutdownSignalException)e.getCause();`
	`332`	`+ } catch (InterruptedException e) {`
	`333`	`+ fail("test interrupted");`
	`334`	`+ }`
	`335`	`+ }`
	`336`	`+ }, null);`
	`337`	`+ (Executors.newSingleThreadExecutor()).execute(waiter);`
	`338`	`+ waiter.get(10, TimeUnit.SECONDS);`
	`339`	`+ } catch (ExecutionException e) {`
	`340`	`+ throw (ShutdownSignalException)e.getCause();`
	`341`	`+ }`
	`342`	`+ }`
`316`	`343`	`}`