basicCancel and basicConsume honor rpc timeout.

Also make sure to remove other e2e bindings for autodeleted exchanges

Author: vikinghawk

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

@@ -39,7 +39,7 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AMQChannel.class);
 
-    private static final int NO_RPC_TIMEOUT = 0;
+    protected static final int NO_RPC_TIMEOUT = 0;
 
     /**
      * Protected; used instead of synchronizing on the channel itself,
@@ -64,7 +64,7 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
     public volatile boolean _blockContent = false;
 
     /** Timeout for RPC calls */
-    private final int _rpcTimeout;
+    protected final int _rpcTimeout;
 
     /**
      * Construct a channel on the given connection, with the given channel number.
@@ -243,24 +243,38 @@ private AMQCommand privateRpc(Method m)
             try {
                 return k.getReply(_rpcTimeout);
             } catch (TimeoutException e) {
-                try {
-                    // clean RPC channel state
-                    nextOutstandingRpc();
-                    markRpcFinished();
-                } catch(Exception ex) {
-                    LOGGER.warn("Error while cleaning timed out channel RPC: {}", ex.getMessage());
-                }
-                throw new ChannelContinuationTimeoutException(e, this, this._channelNumber, m);
+                throw wrapTimeoutException(m, e);
             }
         }
     }
+    
+    private void cleanRpcChannelState() {
+        try {
+            // clean RPC channel state
+            nextOutstandingRpc();
+            markRpcFinished();
+        } catch (Exception ex) {
+            LOGGER.warn("Error while cleaning timed out channel RPC: {}", ex.getMessage());
+        }
+    }
+    
+    /** Cleans RPC channel state after a timeout and wraps the TimeoutException in a ChannelContinuationTimeoutException */
+    protected ChannelContinuationTimeoutException wrapTimeoutException(final Method m, final TimeoutException e)  {
+        cleanRpcChannelState();
+        return new ChannelContinuationTimeoutException(e, this, this._channelNumber, m);
+    }
 
     private AMQCommand privateRpc(Method m, int timeout)
             throws IOException, ShutdownSignalException, TimeoutException {
         SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation();
         rpc(m, k);
 
-        return k.getReply(timeout);
+        try {
+            return k.getReply(timeout);
+        } catch (TimeoutException e) {
+            cleanRpcChannelState();
+            throw e;
+        }
     }
 
     public void rpc(Method m, RpcContinuation k)

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

@@ -1239,18 +1239,26 @@ public String transformReply(AMQCommand replyCommand) {
             }
         };
 
-        rpc(new Basic.Consume.Builder()
-             .queue(queue)
-             .consumerTag(consumerTag)
-             .noLocal(noLocal)
-             .noAck(autoAck)
-             .exclusive(exclusive)
-             .arguments(arguments)
-            .build(),
-            k);
+        final Method m = new Basic.Consume.Builder()
+                .queue(queue)
+                .consumerTag(consumerTag)
+                .noLocal(noLocal)
+                .noAck(autoAck)
+                .exclusive(exclusive)
+                .arguments(arguments)
+               .build();
+        rpc(m, k);
 
         try {
-            return k.getReply();
+            if(_rpcTimeout == NO_RPC_TIMEOUT) {
+                return k.getReply();
+            } else {
+                try {
+                    return k.getReply(_rpcTimeout);
+                } catch (TimeoutException e) {
+                    throw wrapTimeoutException(m, e);
+                }
+            }
         } catch(ShutdownSignalException ex) {
             throw wrap(ex);
         }
@@ -1267,17 +1275,26 @@ public void basicCancel(final String consumerTag)
         BlockingRpcContinuation<Consumer> k = new BlockingRpcContinuation<Consumer>() {
             @Override
             public Consumer transformReply(AMQCommand replyCommand) {
-                replyCommand.getMethod();
+                ((Basic.CancelOk) replyCommand.getMethod()).getConsumerTag(); // just to make sure its the method expected
                 _consumers.remove(consumerTag); //may already have been removed
                 dispatcher.handleCancelOk(originalConsumer, consumerTag);
                 return originalConsumer;
             }
         };
 
-        rpc(new Basic.Cancel(consumerTag, false), k);
-
+        final Method m = new Basic.Cancel(consumerTag, false);
+        rpc(m, k);
+        
         try {
-            k.getReply(); // discard result
+            if(_rpcTimeout == NO_RPC_TIMEOUT) {
+                k.getReply(); // discard result
+            } else {
+                try {
+                    k.getReply(_rpcTimeout);
+                } catch (TimeoutException e) {
+                    throw wrapTimeoutException(m, e);
+                }
+            }
         } catch(ShutdownSignalException ex) {
             throw wrap(ex);
         }

src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java

@@ -805,7 +805,7 @@ void maybeDeleteRecordedAutoDeleteExchange(String exchange) {
                     // last binding where this exchange is the source is gone, remove recorded exchange
                     // if it is auto-deleted. See bug 26364.
                     if((x != null) && x.isAutoDelete()) {
-                        this.recordedExchanges.remove(exchange);
+                        deleteRecordedExchange(exchange);
                     }
                 }
             }