simpler code and update tests

Author: Alexandru Scvortov

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

@@ -189,30 +189,12 @@ public ConfirmListener getConfirmListener() {
     /** {@inheritDoc} */
     public boolean waitForConfirms()
         throws IOException, InterruptedException
-    {
-        return waitForConfirmsInternal(false);
-    }
-
-    /** {@inheritDoc} */
-    public void waitForConfirmsOrDie()
-        throws IOException, InterruptedException
-    {
-        waitForConfirmsInternal(true);
-    }
-
-    protected boolean waitForConfirmsInternal(boolean dieOnNack)
-        throws IOException, InterruptedException
     {
         synchronized (unconfirmedSet) {
             while (true) {
                 if (getCloseReason() != null) {
                     throw Utility.fixStackTrace(getCloseReason());
                 }
-                if (dieOnNack && nacksReceived) {
-                    close(AMQP.REPLY_SUCCESS, "OK", true,
-                          new RuntimeException("received nack"), false);
-                    throw Utility.fixStackTrace(getCloseReason());
-                }
                 if (unconfirmedSet.isEmpty()) {
                     boolean noNacksReceived = !nacksReceived;
                     nacksReceived = false;
@@ -223,6 +205,17 @@ protected boolean waitForConfirmsInternal(boolean dieOnNack)
         }
     }
 
+    /** {@inheritDoc} */
+    public void waitForConfirmsOrDie()
+        throws IOException, InterruptedException
+    {
+        if (!waitForConfirms()) {
+            close(AMQP.REPLY_SUCCESS, "OK", true,
+                  new RuntimeException("received nack"), false);
+            throw Utility.fixStackTrace(getCloseReason());
+        }
+    }
+
     /**
      * Sets the current {@link ConfirmListener}.
      * A null argument is interpreted to mean "do not use a confirm listener".

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

@@ -122,7 +122,7 @@ public void testConfirmQueueDelete()
 
         channel.queueDelete("confirm-test-noconsumer");
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
     }
 
     public void testConfirmQueuePurge()
@@ -132,23 +132,23 @@ public void testConfirmQueuePurge()
 
         channel.queuePurge("confirm-test-noconsumer");
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
     }
 
     public void testConfirmBasicReject()
         throws IOException, InterruptedException
     {
         basicRejectCommon(false);
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
     }
 
     public void testConfirmQueueTTL()
         throws IOException, InterruptedException
     {
         publishN("", "confirm-ttl", true, false, false);
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
     }
 
     public void testConfirmBasicRejectRequeue()
@@ -162,7 +162,7 @@ public void testConfirmBasicRejectRequeue()
         channel.basicConsume("confirm-test-noconsumer", true,
                              new DefaultConsumer(channel));
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
     }
 
     public void testConfirmBasicRecover()
@@ -184,7 +184,7 @@ public void testConfirmBasicRecover()
         channel.basicConsume("confirm-test-noconsumer", true,
                              new DefaultConsumer(channel));
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
     }
 
     public void testConfirmSelect()
@@ -238,7 +238,7 @@ public void handleNack(long seqNo, boolean multiple) {
             publish("", "confirm-test", true, false, false);
         }
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
         if (!unconfirmedSet.isEmpty()) {
             fail("waitForConfirms returned with unconfirmed messages");
         }
@@ -250,7 +250,7 @@ public void testWaitForConfirmsNoOp()
         channel = connection.createChannel();
         // Don't enable Confirm mode
         publish("", "confirm-test", true, false, false);
-        waitAcks(); // Nop
+        channel.waitForConfirmsOrDie(); // Nop
     }
 
     public void testWaitForConfirmsException()
@@ -259,7 +259,7 @@ public void testWaitForConfirmsException()
         publishN("", "confirm-test", true, false, false);
         channel.close();
         try {
-            waitAcks();
+            channel.waitForConfirmsOrDie();
             fail("waitAcks worked on a closed channel");
         } catch (Exception e) {
             //whoosh; everything ok
@@ -274,7 +274,7 @@ public void confirmTest(String exchange, String queueName,
     {
         publishN(exchange, queueName, persistent, mandatory, immediate);
 
-        waitAcks();
+        channel.waitForConfirmsOrDie();
     }
 
     private void publishN(String exchangeName, String queueName,
@@ -309,9 +309,4 @@ protected void publish(String exchangeName, String queueName,
                                         : MessageProperties.BASIC,
                              "nop".getBytes());
     }
-
-    protected void waitAcks() throws IOException, InterruptedException {
-        if (!channel.waitForConfirms())
-            fail("got nacks");
-    }
 }

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

@@ -31,8 +31,7 @@ public void test() throws IOException, InterruptedException {
         channel.basicPublish("", Q, false, false,
                              MessageProperties.PERSISTENT_BASIC,
                              "nop".getBytes());
-        if (!channel.waitForConfirms())
-            fail("received nacks");
+        channel.waitForConfirmsOrDie();
 
         restart();
         assertDelivered(Q, 1);