Do not confirmSelect more than once per channel

In order to avoid unnecessary blocking RPC calls and conform to best
practices, the Channel now checks if it is already activated confirm
mode before sending a confirm.select RPC call.

If confirm mode is already activated, calling confirmSelect() again
returns immediately without sending an RPC call.

Closes #1056

Author: sergio91pt

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

@@ -82,6 +82,9 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
     private final SortedSet<Long> unconfirmedSet =
             Collections.synchronizedSortedSet(new TreeSet<Long>());
 
+    /** Whether the confirm select method has been successfully activated */
+    private boolean confirmSelectActivated = false;
+
     /** Whether any nacks have been received since the last waitForConfirms(). */
     private volatile boolean onlyAcksReceived = true;
 
@@ -1553,10 +1556,16 @@ public Tx.RollbackOk txRollback()
     public Confirm.SelectOk confirmSelect()
         throws IOException
     {
+        if (confirmSelectActivated) {
+            return new Confirm.SelectOk();
+        }
+
         if (nextPublishSeqNo == 0) nextPublishSeqNo = 1;
-        return (Confirm.SelectOk)
+        Confirm.SelectOk result = (Confirm.SelectOk)
             exnWrappingRpc(new Confirm.Select(false)).getMethod();
 
+        confirmSelectActivated = true;
+        return result;
     }
 
     /** Public API - {@inheritDoc} */

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

@@ -15,17 +15,20 @@
 
 package com.rabbitmq.client.test;
 
+import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Method;
+import com.rabbitmq.client.TrafficListener;
 import com.rabbitmq.client.impl.*;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.io.IOException;
+import java.util.concurrent.*;
 import java.util.stream.Stream;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 public class ChannelNTest {
 
@@ -81,6 +84,30 @@ public TestConfig(int value, Consumer call) {
                 .forEach(config -> assertThatThrownBy(() -> config.call.apply(config.value)).isInstanceOf(IllegalArgumentException.class));
     }
 
+    @Test
+    public void confirmSelectOnlySendsRPCCallOnce() throws Exception {
+        AMQConnection connection = Mockito.mock(AMQConnection.class);
+        TrafficListener trafficListener = Mockito.mock(TrafficListener.class);
+
+        Mockito.when(connection.getTrafficListener()).thenReturn(trafficListener);
+
+        ChannelN channel = new ChannelN(connection, 1, consumerWorkService);
+
+        Future<AMQImpl.Confirm.SelectOk> future = executorService.submit(() -> {
+            try {
+                return channel.confirmSelect();
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        });
+
+        channel.handleCompleteInboundCommand(new AMQCommand(new AMQImpl.Confirm.SelectOk()));
+
+        assertNotNull(future.get(1, TimeUnit.SECONDS));
+        assertNotNull(channel.confirmSelect());
+        Mockito.verify(trafficListener, Mockito.times(1)).write(Mockito.any(Command.class));
+    }
+
     interface Consumer {
 
         void apply(int value) throws Exception;