Skip to content

Commit 0dc9ea2

Browse files
sergio91ptacogoluegnes
authored andcommitted
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 (cherry picked from commit 7253c94) Conflicts: src/test/java/com/rabbitmq/client/test/ChannelNTest.java
1 parent 129dc6a commit 0dc9ea2

File tree

2 files changed

+40
-6
lines changed

2 files changed

+40
-6
lines changed

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

+10-1
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,9 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
8282
private final SortedSet<Long> unconfirmedSet =
8383
Collections.synchronizedSortedSet(new TreeSet<Long>());
8484

85+
/** Whether the confirm select method has been successfully activated */
86+
private boolean confirmSelectActivated = false;
87+
8588
/** Whether any nacks have been received since the last waitForConfirms(). */
8689
private volatile boolean onlyAcksReceived = true;
8790

@@ -1553,10 +1556,16 @@ public Tx.RollbackOk txRollback()
15531556
public Confirm.SelectOk confirmSelect()
15541557
throws IOException
15551558
{
1559+
if (confirmSelectActivated) {
1560+
return new Confirm.SelectOk();
1561+
}
1562+
15561563
if (nextPublishSeqNo == 0) nextPublishSeqNo = 1;
1557-
return (Confirm.SelectOk)
1564+
Confirm.SelectOk result = (Confirm.SelectOk)
15581565
exnWrappingRpc(new Confirm.Select(false)).getMethod();
15591566

1567+
confirmSelectActivated = true;
1568+
return result;
15601569
}
15611570

15621571
/** Public API - {@inheritDoc} */

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

+30-5
Original file line numberDiff line numberDiff line change
@@ -15,21 +15,22 @@
1515

1616
package com.rabbitmq.client.test;
1717

18+
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Command;
1819
import com.rabbitmq.client.Method;
20+
import com.rabbitmq.client.TrafficListener;
1921
import com.rabbitmq.client.impl.*;
2022
import org.junit.jupiter.api.AfterEach;
2123
import org.junit.jupiter.api.BeforeEach;
2224
import org.junit.jupiter.api.Test;
2325
import org.mockito.Mockito;
2426

2527
import java.io.IOException;
26-
import java.util.concurrent.ExecutorService;
27-
import java.util.concurrent.Executors;
28-
import java.util.concurrent.atomic.AtomicReference;
29-
import java.util.stream.Stream;
28+
import java.util.concurrent.*;
29+
import java.util.concurrent.atomic.AtomicReference;import java.util.stream.Stream;
3030

3131
import static org.assertj.core.api.Assertions.assertThat;
3232
import static org.assertj.core.api.Assertions.assertThatThrownBy;
33+
import static org.junit.jupiter.api.Assertions.assertNotNull;
3334

3435
public class ChannelNTest {
3536

@@ -67,7 +68,7 @@ public void callingBasicCancelForUnknownConsumerThrowsException() throws Excepti
6768
@Test
6869
public void qosShouldBeUnsignedShort() {
6970
AMQConnection connection = Mockito.mock(AMQConnection.class);
70-
AtomicReference<com.rabbitmq.client.AMQP.Basic.Qos> qosMethod = new AtomicReference<>();
71+
AtomicReference<AMQP.Basic.Qos> qosMethod = new AtomicReference<>();
7172
ChannelN channel = new ChannelN(connection, 1, consumerWorkService) {
7273
@Override
7374
public AMQCommand exnWrappingRpc(Method m) {
@@ -106,6 +107,30 @@ public TestConfig(int value, Consumer call, int expected) {
106107
});
107108
}
108109

110+
@Test
111+
public void confirmSelectOnlySendsRPCCallOnce() throws Exception {
112+
AMQConnection connection = Mockito.mock(AMQConnection.class);
113+
TrafficListener trafficListener = Mockito.mock(TrafficListener.class);
114+
115+
Mockito.when(connection.getTrafficListener()).thenReturn(trafficListener);
116+
117+
ChannelN channel = new ChannelN(connection, 1, consumerWorkService);
118+
119+
Future<AMQImpl.Confirm.SelectOk> future = executorService.submit(() -> {
120+
try {
121+
return channel.confirmSelect();
122+
} catch (IOException e) {
123+
throw new RuntimeException(e);
124+
}
125+
});
126+
127+
channel.handleCompleteInboundCommand(new AMQCommand(new AMQImpl.Confirm.SelectOk()));
128+
129+
assertNotNull(future.get(1, TimeUnit.SECONDS));
130+
assertNotNull(channel.confirmSelect());
131+
Mockito.verify(trafficListener, Mockito.times(1)).write(Mockito.any(Command.class));
132+
}
133+
109134
interface Consumer {
110135

111136
void apply(int value) throws Exception;

0 commit comments

Comments
 (0)