Prevent waiting for confirms unless selected

Author: Emile Joubert

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

@@ -83,6 +83,8 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
     /** Sequence number of next published message requiring confirmation.*/
     private long nextPublishSeqNo = 0L;
 
+    private volatile boolean confirmSelected = false;
+
     /** The current default consumer, or null if there is none. */
     private volatile Consumer defaultConsumer = null;
 
@@ -175,6 +177,8 @@ public boolean waitForConfirms()
     /** {@inheritDoc} */
     public boolean waitForConfirms(long timeout)
             throws InterruptedException, TimeoutException {
+        if (!confirmSelected)
+            throw new IllegalStateException("Confirms not selected");
         long startTime = System.currentTimeMillis();
         synchronized (unconfirmedSet) {
             while (true) {
@@ -1048,6 +1052,7 @@ public Tx.RollbackOk txRollback()
     public Confirm.SelectOk confirmSelect()
         throws IOException
     {
+        confirmSelected = true;
         if (nextPublishSeqNo == 0) nextPublishSeqNo = 1;
         return (Confirm.SelectOk)
             exnWrappingRpc(new Confirm.Select(false)).getMethod();

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

@@ -229,13 +229,16 @@ public void handleNack(long seqNo, boolean multiple) {
         }
     }
 
-    public void testWaitForConfirmsNoOp()
+    public void testWaitForConfirmsWithoutConfirmSelected()
         throws IOException, InterruptedException
     {
         channel = connection.createChannel();
         // Don't enable Confirm mode
         publish("", "confirm-test", true, false);
-        channel.waitForConfirmsOrDie(); // Nop
+        try {
+            channel.waitForConfirms();
+            fail("waitForConfirms without confirms selected succeeded");
+        } catch (IllegalStateException _) {}
     }
 
     public void testWaitForConfirmsException()