Merge pull request #292 from rabbitmq/rabbitmq-java-client-290

michaelklishin · web-flow · commit 7d71a809f706 · 2017-07-08T02:40:35.000+03:00
Add option to ensure RPC reply is for the current request
diff --git a/src/main/java/com/rabbitmq/client/ConnectionFactory.java b/src/main/java/com/rabbitmq/client/ConnectionFactory.java
@@ -127,6 +127,13 @@ public class ConnectionFactory implements Cloneable {
      */
     private int channelRpcTimeout = DEFAULT_CHANNEL_RPC_TIMEOUT;
 
+    /**
+     * Whether or not channels check the reply type of an RPC call.
+     * Default is false.
+     * @since 4.2.0
+     */
+    private boolean channelShouldCheckRpcResponseType = false;
+
     /** @return the default host to use for connections */
     public String getHost() {
         return host;
@@ -958,6 +965,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setShutdownExecutor(shutdownExecutor);
         result.setHeartbeatExecutor(heartbeatExecutor);
         result.setChannelRpcTimeout(channelRpcTimeout);
+        result.setChannelShouldCheckRpcResponseType(channelShouldCheckRpcResponseType);
         return result;
     }
 
@@ -1126,4 +1134,19 @@ public void setChannelRpcTimeout(int channelRpcTimeout) {
     public int getChannelRpcTimeout() {
         return channelRpcTimeout;
     }
+
+    /**
+     * When set to true, channels will check the response type (e.g. queue.declare
+     * expects a queue.declare-ok response) of RPC calls
+     * and ignore those that do not match.
+     * Default is false.
+     * @param channelShouldCheckRpcResponseType
+     */
+    public void setChannelShouldCheckRpcResponseType(boolean channelShouldCheckRpcResponseType) {
+        this.channelShouldCheckRpcResponseType = channelShouldCheckRpcResponseType;
+    }
+
+    public boolean isChannelShouldCheckRpcResponseType() {
+        return channelShouldCheckRpcResponseType;
+    }
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/AMQChannel.java b/src/main/java/com/rabbitmq/client/impl/AMQChannel.java
@@ -20,6 +20,11 @@
 import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.*;
+import com.rabbitmq.client.AMQP.Basic;
+import com.rabbitmq.client.AMQP.Confirm;
+import com.rabbitmq.client.AMQP.Exchange;
+import com.rabbitmq.client.AMQP.Queue;
+import com.rabbitmq.client.AMQP.Tx;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.utility.BlockingValueOrException;
 import org.slf4j.Logger;
@@ -66,6 +71,8 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
     /** Timeout for RPC calls */
     protected final int _rpcTimeout;
 
+    private final boolean _checkRpcResponseType;
+
     /**
      * Construct a channel on the given connection, with the given channel number.
      * @param connection the underlying connection for this channel
@@ -78,6 +85,7 @@ public AMQChannel(AMQConnection connection, int channelNumber) {
             throw new IllegalArgumentException("Continuation timeout on RPC calls cannot be less than 0");
         }
         this._rpcTimeout = connection.getChannelRpcTimeout();
+        this._checkRpcResponseType = connection.willCheckRpcResponseType();
     }
 
     /**
@@ -153,8 +161,19 @@ public void handleCompleteInboundCommand(AMQCommand command) throws IOException
         // waiting RPC continuation.
         if (!processAsync(command)) {
             // The filter decided not to handle/consume the command,
-            // so it must be some reply to an earlier RPC.
-            RpcContinuation nextOutstandingRpc = nextOutstandingRpc();
+            // so it must be a response to an earlier RPC.
+            if (_checkRpcResponseType) {
+                synchronized (_channelMutex) {
+                    // check if this reply command is intended for the current waiting request before calling nextOutstandingRpc()
+                    if (!_activeRpc.canHandleReply(command)) {
+                        // this reply command is not intended for the current waiting request
+                        // most likely a previous request timed out and this command is the reply for that.
+                        // Throw this reply command away so we don't stop the current request from waiting for its reply
+                        return;
+                    }
+                }
+            }
+            final RpcContinuation nextOutstandingRpc = nextOutstandingRpc();
             // the outstanding RPC can be null when calling Channel#asyncRpc
             if(nextOutstandingRpc != null) {
                 nextOutstandingRpc.handleCommand(command);
@@ -229,7 +248,7 @@ public AMQCommand rpc(Method m, int timeout)
     private AMQCommand privateRpc(Method m)
         throws IOException, ShutdownSignalException
     {
-        SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
+        SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation(m);
         rpc(m, k);
         // At this point, the request method has been sent, and we
         // should wait for the reply to arrive.
@@ -266,7 +285,7 @@ protected ChannelContinuationTimeoutException wrapTimeoutException(final Method
 
     private AMQCommand privateRpc(Method m, int timeout)
             throws IOException, ShutdownSignalException, TimeoutException {
-        SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
+        SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation(m);
         rpc(m, k);
 
         try {
@@ -384,13 +403,25 @@ public AMQConnection getConnection() {
 
     public interface RpcContinuation {
         void handleCommand(AMQCommand command);
+        /** @return true if the reply command can be handled for this request */
+        boolean canHandleReply(AMQCommand command);
         void handleShutdownSignal(ShutdownSignalException signal);
     }
 
     public static abstract class BlockingRpcContinuation<T> implements RpcContinuation {
         public final BlockingValueOrException<T, ShutdownSignalException> _blocker =
             new BlockingValueOrException<T, ShutdownSignalException>();
 
+        protected final Method request;
+
+        public BlockingRpcContinuation() {
+            request = null;
+        }
+
+        public BlockingRpcContinuation(final Method request) {
+            this.request = request;
+        }
+
         @Override
         public void handleCommand(AMQCommand command) {
             _blocker.setValue(transformReply(command));
@@ -412,12 +443,79 @@ public T getReply(int timeout)
             return _blocker.uninterruptibleGetValue(timeout);
         }
 
+        @Override
+        public boolean canHandleReply(AMQCommand command) {
+            // make a best effort attempt to ensure the reply was intended for this rpc request
+            // Ideally each rpc request would tag an id on it that could be returned and referenced on its reply.
+            // But because that would be a very large undertaking to add passively this logic at least protects against ClassCastExceptions
+            if (request != null) {
+                final Method reply = command.getMethod();
+                if (request instanceof Basic.Qos) {
+                    return reply instanceof Basic.QosOk;
+                } else if (request instanceof Basic.Get) {
+                    return reply instanceof Basic.GetOk || reply instanceof Basic.GetEmpty;
+                } else if (request instanceof Basic.Consume) {
+                    if (!(reply instanceof Basic.ConsumeOk))
+                        return false;
+                    // can also check the consumer tags match here. handle case where request consumer tag is empty and server-generated.
+                    final String consumerTag = ((Basic.Consume)request).getConsumerTag();
+                    return consumerTag == null || consumerTag.equals("") || consumerTag.equals(((Basic.ConsumeOk)reply).getConsumerTag());
+                } else if (request instanceof Basic.Cancel) {
+                    if (!(reply instanceof Basic.CancelOk))
+                        return false;
+                    // can also check the consumer tags match here
+                    return ((Basic.Cancel)request).getConsumerTag().equals(((Basic.CancelOk)reply).getConsumerTag());
+                } else if (request instanceof Basic.Recover) {
+                    return reply instanceof Basic.RecoverOk;
+                } else if (request instanceof Exchange.Declare) {
+                    return reply instanceof Exchange.DeclareOk;
+                } else if (request instanceof Exchange.Delete) {
+                    return reply instanceof Exchange.DeleteOk;
+                } else if (request instanceof Exchange.Bind) {
+                    return reply instanceof Exchange.BindOk;
+                } else if (request instanceof Exchange.Unbind) {
+                    return reply instanceof Exchange.UnbindOk;
+                } else if (request instanceof Queue.Declare) {
+                    // we cannot check the queue name, as the server can strip some characters
+                    // see QueueLifecycle test and https://github.com/rabbitmq/rabbitmq-server/issues/710
+                    return reply instanceof Queue.DeclareOk;
+                } else if (request instanceof Queue.Delete) {
+                    return reply instanceof Queue.DeleteOk;
+                } else if (request instanceof Queue.Bind) {
+                    return reply instanceof Queue.BindOk;
+                } else if (request instanceof Queue.Unbind) {
+                    return reply instanceof Queue.UnbindOk;
+                } else if (request instanceof Queue.Purge) {
+                    return reply instanceof Queue.PurgeOk;
+                } else if (request instanceof Tx.Select) {
+                    return reply instanceof Tx.SelectOk;
+                } else if (request instanceof Tx.Commit) {
+                    return reply instanceof Tx.CommitOk;
+                } else if (request instanceof Tx.Rollback) {
+                    return reply instanceof Tx.RollbackOk;
+                } else if (request instanceof Confirm.Select) {
+                    return reply instanceof Confirm.SelectOk;
+                }
+            }
+            // for passivity default to true
+            return true;
+        }
+
         public abstract T transformReply(AMQCommand command);
     }
 
     public static class SimpleBlockingRpcContinuation
         extends BlockingRpcContinuation<AMQCommand>
     {
+
+        public SimpleBlockingRpcContinuation() {
+            super();
+        }
+
+        public SimpleBlockingRpcContinuation(final Method method) {
+            super(method);
+        }
+
         @Override
         public AMQCommand transformReply(AMQCommand command) {
             return command;
diff --git a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java
@@ -134,6 +134,7 @@ public static Map<String, Object> defaultClientProperties() {
     private final Collection<BlockedListener> blockedListeners = new CopyOnWriteArrayList<BlockedListener>();
     protected final MetricsCollector metricsCollector;
     private final int channelRpcTimeout;
+    private final boolean channelShouldCheckRpcResponseType;
 
     /* State modified after start - all volatile */
 
@@ -229,6 +230,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler, Metrics
             throw new IllegalArgumentException("Continuation timeout on RPC calls cannot be less than 0");
         }
         this.channelRpcTimeout = params.getChannelRpcTimeout();
+        this.channelShouldCheckRpcResponseType = params.channelShouldCheckRpcResponseType();
 
         this._channel0 = new AMQChannel(this, 0) {
             @Override public boolean processAsync(Command c) throws IOException {
@@ -1056,4 +1058,8 @@ public void setId(String id) {
     public int getChannelRpcTimeout() {
         return channelRpcTimeout;
     }
+
+    public boolean willCheckRpcResponseType() {
+        return channelShouldCheckRpcResponseType;
+    }
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/ChannelN.java b/src/main/java/com/rabbitmq/client/impl/ChannelN.java
@@ -1229,7 +1229,15 @@ public String basicConsume(String queue, final boolean autoAck, String consumerT
                                final Consumer callback)
         throws IOException
     {
-        BlockingRpcContinuation<String> k = new BlockingRpcContinuation<String>() {
+        final Method m = new Basic.Consume.Builder()
+            .queue(queue)
+            .consumerTag(consumerTag)
+            .noLocal(noLocal)
+            .noAck(autoAck)
+            .exclusive(exclusive)
+            .arguments(arguments)
+            .build();
+        BlockingRpcContinuation<String> k = new BlockingRpcContinuation<String>(m) {
             @Override
             public String transformReply(AMQCommand replyCommand) {
                 String actualConsumerTag = ((Basic.ConsumeOk) replyCommand.getMethod()).getConsumerTag();
@@ -1243,14 +1251,7 @@ public String transformReply(AMQCommand replyCommand) {
             }
         };
 
-        final Method m = new Basic.Consume.Builder()
-                .queue(queue)
-                .consumerTag(consumerTag)
-                .noLocal(noLocal)
-                .noAck(autoAck)
-                .exclusive(exclusive)
-                .arguments(arguments)
-               .build();
+
         rpc(m, k);
 
         try {
@@ -1276,7 +1277,9 @@ public void basicCancel(final String consumerTag)
         final Consumer originalConsumer = _consumers.get(consumerTag);
         if (originalConsumer == null)
             throw new IOException("Unknown consumerTag");
-        BlockingRpcContinuation<Consumer> k = new BlockingRpcContinuation<Consumer>() {
+
+        final Method m = new Basic.Cancel(consumerTag, false);
+        BlockingRpcContinuation<Consumer> k = new BlockingRpcContinuation<Consumer>(m) {
             @Override
             public Consumer transformReply(AMQCommand replyCommand) {
                 if (!(replyCommand.getMethod() instanceof Basic.CancelOk))
@@ -1287,7 +1290,7 @@ public Consumer transformReply(AMQCommand replyCommand) {
             }
         };
 
-        final Method m = new Basic.Cancel(consumerTag, false);
+
         rpc(m, k);
         
         try {
diff --git a/src/main/java/com/rabbitmq/client/impl/ConnectionParams.java b/src/main/java/com/rabbitmq/client/impl/ConnectionParams.java
@@ -40,6 +40,7 @@ public class ConnectionParams {
     private long networkRecoveryInterval;
     private boolean topologyRecovery;
     private int channelRpcTimeout;
+    private boolean channelShouldCheckRpcResponseType;
 
     private ExceptionHandler exceptionHandler;
     private ThreadFactory threadFactory;
@@ -114,6 +115,10 @@ public int getChannelRpcTimeout() {
         return channelRpcTimeout;
     }
 
+    public boolean channelShouldCheckRpcResponseType() {
+        return channelShouldCheckRpcResponseType;
+    }
+
     public void setUsername(String username) {
         this.username = username;
     }
@@ -189,4 +194,8 @@ public void setHeartbeatExecutor(ScheduledExecutorService heartbeatExecutor) {
     public void setChannelRpcTimeout(int channelRpcTimeout) {
         this.channelRpcTimeout = channelRpcTimeout;
     }
+
+    public void setChannelShouldCheckRpcResponseType(boolean channelShouldCheckRpcResponseType) {
+        this.channelShouldCheckRpcResponseType = channelShouldCheckRpcResponseType;
+    }
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/nio/HeaderWriteRequest.java b/src/main/java/com/rabbitmq/client/impl/nio/HeaderWriteRequest.java
@@ -19,8 +19,6 @@
 
 import java.io.DataOutputStream;
 import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.channels.WritableByteChannel;
 
 /**
  *
diff --git a/src/test/java/com/rabbitmq/client/test/AMQChannelTest.java b/src/test/java/com/rabbitmq/client/test/AMQChannelTest.java
@@ -106,6 +106,62 @@ public Void call() throws Exception {
         assertThat(rpcResponse.getMethod(), is(response));
     }
 
+    @Test
+    public void testRpcTimeoutReplyComesDuringNexRpc() throws Exception {
+        int rpcTimeout = 100;
+        AMQConnection connection = mock(AMQConnection.class);
+        when(connection.getChannelRpcTimeout()).thenReturn(rpcTimeout);
+        when(connection.willCheckRpcResponseType()).thenReturn(Boolean.TRUE);
+
+        final DummyAmqChannel channel = new DummyAmqChannel(connection, 1);
+        Method method = new AMQImpl.Queue.Declare.Builder()
+            .queue("123")
+            .durable(false)
+            .exclusive(true)
+            .autoDelete(true)
+            .arguments(null)
+            .build();
+
+        try {
+            channel.rpc(method);
+            fail("Should time out and throw an exception");
+        } catch(final ChannelContinuationTimeoutException e) {
+            // OK
+            assertThat((DummyAmqChannel) e.getChannel(), is(channel));
+            assertThat(e.getChannelNumber(), is(channel.getChannelNumber()));
+            assertThat(e.getMethod(), is(method));
+            assertNull("outstanding RPC should have been cleaned", channel.nextOutstandingRpc());
+        }
+
+        // now do a basic.consume request and have the queue.declareok returned instead
+        method = new AMQImpl.Basic.Consume.Builder()
+            .queue("123")
+            .consumerTag("")
+            .arguments(null)
+            .build();
+
+        final Method response1 = new AMQImpl.Queue.DeclareOk.Builder()
+            .queue("123")
+            .consumerCount(0)
+            .messageCount(0).build();
+
+        final Method response2 = new AMQImpl.Basic.ConsumeOk.Builder()
+            .consumerTag("456").build();
+
+        scheduler.schedule(new Callable<Void>() {
+            @Override
+            public Void call() throws Exception {
+                channel.handleCompleteInboundCommand(new AMQCommand(response1));
+                Thread.sleep(10);
+                channel.handleCompleteInboundCommand(new AMQCommand(response2));
+                return null;
+            }
+        }, (long) (rpcTimeout / 2.0), TimeUnit.MILLISECONDS);
+
+        AMQCommand rpcResponse = channel.rpc(method);
+        assertThat(rpcResponse.getMethod(), is(response2));
+    }
+
     static class DummyAmqChannel extends AMQChannel {
 
         public DummyAmqChannel(AMQConnection connection, int channelNumber) {

Original file line number	Diff line number	Diff line change
`@@ -40,6 +40,7 @@ public class ConnectionParams {`
`40`	`40`	`private long networkRecoveryInterval;`
`41`	`41`	`private boolean topologyRecovery;`
`42`	`42`	`private int channelRpcTimeout;`
	`43`	`+ private boolean channelShouldCheckRpcResponseType;`
`43`	`44`
`44`	`45`	`private ExceptionHandler exceptionHandler;`
`45`	`46`	`private ThreadFactory threadFactory;`
`@@ -114,6 +115,10 @@ public int getChannelRpcTimeout() {`
`114`	`115`	`return channelRpcTimeout;`
`115`	`116`	`}`
`116`	`117`
	`118`	`+ public boolean channelShouldCheckRpcResponseType() {`
	`119`	`+ return channelShouldCheckRpcResponseType;`
	`120`	`+ }`
	`121`	`+`
`117`	`122`	`public void setUsername(String username) {`
`118`	`123`	`this.username = username;`
`119`	`124`	`}`
`@@ -189,4 +194,8 @@ public void setHeartbeatExecutor(ScheduledExecutorService heartbeatExecutor) {`
`189`	`194`	`public void setChannelRpcTimeout(int channelRpcTimeout) {`
`190`	`195`	`this.channelRpcTimeout = channelRpcTimeout;`
`191`	`196`	`}`
	`197`	`+`
	`198`	`+ public void setChannelShouldCheckRpcResponseType(boolean channelShouldCheckRpcResponseType) {`
	`199`	`+ this.channelShouldCheckRpcResponseType = channelShouldCheckRpcResponseType;`
	`200`	`+ }`
`192`	`201`	`}`
Original file line number	Diff line number	Diff line change
`@@ -19,8 +19,6 @@`
`19`	`19`
`20`	`20`	`import java.io.DataOutputStream;`
`21`	`21`	`import java.io.IOException;`
`22`		`-import java.nio.ByteBuffer;`
`23`		`-import java.nio.channels.WritableByteChannel;`
`24`	`22`
`25`	`23`	`/**`
`26`	`24`	`*`