Check that consumerTag is valid on basicCancel; also freeze consumer it references at call time.

Author: Steve Powell

src/com/rabbitmq/client/Channel.java

@@ -638,7 +638,7 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * Cancel a consumer. Calls the consumer's {@link Consumer#handleCancelOk}
      * method.
      * @param consumerTag a client- or server-generated consumer tag to establish context
-     * @throws java.io.IOException if an error is encountered
+     * @throws IOException if an error is encountered, or if the consumerTag is unknown
      * @see com.rabbitmq.client.AMQP.Basic.Cancel
      * @see com.rabbitmq.client.AMQP.Basic.CancelOk
      */
@@ -649,12 +649,12 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * basic.recover is asynchronous; in 0-9-1 it is synchronous, and
      * the new, deprecated method basic.recover_async is asynchronous.
      * <p/>
-     * Equivalent to calling <code>basicRecover(true)</code>, messages 
-     * will be requeued and possibly delivered to a different consumer. 
+     * Equivalent to calling <code>basicRecover(true)</code>, messages
+     * will be requeued and possibly delivered to a different consumer.
      * @see #basicRecover(boolean)
      */
      Basic.RecoverOk basicRecover() throws IOException;
-  
+
     /**
      * Ask the broker to resend unacknowledged messages.  In 0-8
      * basic.recover is asynchronous; in 0-9-1 it is synchronous, and

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

@@ -233,7 +233,7 @@ private CountDownLatch broadcastShutdownSignal(ShutdownSignalException signal) {
         }
     }
 
-    public CountDownLatch getShutdownLatch() {
+    CountDownLatch getShutdownLatch() {
         return this.finishedShutdownFlag;
     }
 
@@ -387,7 +387,7 @@ private void callReturnListeners(Command command, Basic.Return basicReturn) {
         }
     }
 
-    private void callFlowListeners(Command command, Channel.Flow channelFlow) {
+    private void callFlowListeners(@SuppressWarnings("unused") Command command, Channel.Flow channelFlow) {
         try {
             for (FlowListener l : this.flowListeners) {
                 l.handleFlow(channelFlow.getActive());
@@ -397,7 +397,7 @@ private void callFlowListeners(Command command, Channel.Flow channelFlow) {
         }
     }
 
-    private void callConfirmListeners(Command command, Basic.Ack ack) {
+    private void callConfirmListeners(@SuppressWarnings("unused") Command command, Basic.Ack ack) {
         try {
             for (ConfirmListener l : this.confirmListeners) {
                 l.handleAck(ack.getDeliveryTag(), ack.getMultiple());
@@ -407,7 +407,7 @@ private void callConfirmListeners(Command command, Basic.Ack ack) {
         }
     }
 
-    private void callConfirmListeners(Command command, Basic.Nack nack) {
+    private void callConfirmListeners(@SuppressWarnings("unused") Command command, Basic.Nack nack) {
         try {
             for (ConfirmListener l : this.confirmListeners) {
                 l.handleNack(nack.getDeliveryTag(), nack.getMultiple());
@@ -463,9 +463,16 @@ public void abort(int closeCode, String closeMessage)
     }
 
     /**
+     * SHOULD BE PRIVATE<br/>
      * Protected API - Close channel with code and message, indicating
      * the source of the closure and a causing exception (null if
      * none).
+     * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
+     * @param closeMessage a message indicating the reason for closing the connection
+     * @param initiatedByApplication true if this comes from an API call, false otherwise
+     * @param cause exception triggering close
+     * @param abort true if we should close and ignore errors
+     * @throws IOException if an error is encountered
      */
     public void close(int closeCode,
                       String closeMessage,
@@ -907,15 +914,16 @@ public String transformReply(AMQCommand replyCommand) {
     public void basicCancel(final String consumerTag)
         throws IOException
     {
+        final Consumer originalConsumer = _consumers.get(consumerTag);
+        if (originalConsumer == null)
+            throw new IOException("Unknown consumerTag");
         BlockingRpcContinuation<Consumer> k = new BlockingRpcContinuation<Consumer>() {
             public Consumer transformReply(AMQCommand replyCommand) {
                 Basic.CancelOk dummy = (Basic.CancelOk) replyCommand.getMethod();
                 Utility.use(dummy);
-                Consumer callback = _consumers.remove(consumerTag);
-                // We need to call back inside the connection thread
-                // in order avoid races with 'deliver' commands
-                dispatcher.handleCancelOk(callback, consumerTag);
-                return callback;
+                _consumers.remove(consumerTag); //may already have been removed
+                dispatcher.handleCancelOk(originalConsumer, consumerTag);
+                return originalConsumer;
             }
         };