merge bug 24538 into default (WaitForConfirms with timeouts)

Author: Alexandru Scvortov

src/com/rabbitmq/client/Channel.java

@@ -639,7 +639,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
      */
@@ -650,12 +650,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/AMQConnection.java

@@ -578,9 +578,7 @@ public boolean processControlCommand(Command c) throws IOException
                 // Already shutting down, so just send back a CloseOk.
                 try {
                     _channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());
-                } catch (IOException ioe) {
-                    Utility.emptyStatement();
-                }
+                } catch (IOException _) { } // ignore
                 return true;
             } else if (method instanceof AMQP.Connection.CloseOk) {
                 // It's our final "RPC". Time to shut down.
@@ -599,9 +597,7 @@ public void handleConnectionClose(Command closeCommand) {
         ShutdownSignalException sse = shutdown(closeCommand, false, null, false);
         try {
             _channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());
-        } catch (IOException ioe) {
-            Utility.emptyStatement();
-        }
+        } catch (IOException _) { } // ignore
         _brokerInitiatedShutdown = true;
         Thread scw = new SocketCloseWait(sse);
         scw.setName("AMQP Connection Closing Monitor " +
@@ -708,9 +704,7 @@ public void abort(int closeCode, String closeMessage, int timeout)
     {
         try {
             close(closeCode, closeMessage, true, null, timeout, true);
-        } catch (IOException e) {
-            Utility.emptyStatement();
-        }
+        } catch (IOException _) { } // ignore
     }
 
     /**

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

@@ -121,10 +121,8 @@ public ChannelN(AMQConnection connection, int channelNumber,
      * @throws IOException if any problem is encountered
      */
     public void open() throws IOException {
-        // wait for the Channel.OpenOk response, then ignore it
-        Channel.OpenOk openOk =
-            (Channel.OpenOk) exnWrappingRpc(new Channel.Open(UNSPECIFIED_OUT_OF_BAND)).getMethod();
-        Utility.use(openOk);
+        // wait for the Channel.OpenOk response, and ignore it
+        exnWrappingRpc(new Channel.Open(UNSPECIFIED_OUT_OF_BAND));
     }
 
     public void addReturnListener(ReturnListener listener) {
@@ -267,7 +265,7 @@ private CountDownLatch broadcastShutdownSignal(ShutdownSignalException signal) {
         }
     }
 
-    public CountDownLatch getShutdownLatch() {
+    CountDownLatch getShutdownLatch() {
         return this.finishedShutdownFlag;
     }
 
@@ -421,7 +419,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());
@@ -431,7 +429,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());
@@ -441,7 +439,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());
@@ -496,10 +494,17 @@ public void abort(int closeCode, String closeMessage)
         close(closeCode, closeMessage, true, null, true);
     }
 
+    // TODO: method should be private
     /**
      * 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,
@@ -941,23 +946,22 @@ 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;
+                replyCommand.getMethod();
+                _consumers.remove(consumerTag); //may already have been removed
+                dispatcher.handleCancelOk(originalConsumer, consumerTag);
+                return originalConsumer;
             }
         };
 
         rpc(new Basic.Cancel(consumerTag, false), k);
 
         try {
-            Consumer callback = k.getReply();
-            Utility.use(callback);
+            k.getReply(); // discard result
         } catch(ShutdownSignalException ex) {
             throw wrap(ex);
         }

src/com/rabbitmq/utility/Utility.java

@@ -33,9 +33,9 @@ public ThrowableCreatedElsewhere(Throwable throwable) {
           this.setStackTrace(throwable.getStackTrace());
         }
 
-        @Override public Throwable fillInStackTrace(){ 
-            return this; 
-        } 
+        @Override public Throwable fillInStackTrace(){
+            return this;
+        }
     }
 
     public static <T extends Throwable & SensibleClone<T>> T fixStackTrace(T throwable) {
@@ -44,19 +44,19 @@ public static <T extends Throwable & SensibleClone<T>> T fixStackTrace(T throwab
       if(throwable.getCause() == null) {
         // We'd like to preserve the original stack trace in the cause.
         // Unfortunately Java doesn't let you set the cause once it's been
-        // set once. This means we have to choose between either 
+        // set once. This means we have to choose between either
         //  - not preserving the type
         //  - sometimes losing the original stack trace
         //  - performing nasty reflective voodoo which may or may not work
         // We only lose the original stack trace when there's a root cause
-        // which will hopefully be enlightening enough on its own that it 
-        // doesn't matter too much. 
+        // which will hopefully be enlightening enough on its own that it
+        // doesn't matter too much.
         try {
           throwable.initCause(new ThrowableCreatedElsewhere(throwable));
         } catch(IllegalStateException e) {
-          // This exception was explicitly initialised with a null cause. 
-          // Alas this means we can't set the cause even though it has none. 
-          // Thanks. 
+          // This exception was explicitly initialised with a null cause.
+          // Alas this means we can't set the cause even though it has none.
+          // Thanks.
         }
       }
 
@@ -70,7 +70,7 @@ public static <T extends Throwable & SensibleClone<T>> T fixStackTrace(T throwab
       return throwable;
     }
 
-  
+
     public static String makeStackTrace(Throwable throwable) {
         ByteArrayOutputStream baOutStream = new ByteArrayOutputStream();
         PrintStream printStream = new PrintStream(baOutStream, false);
@@ -80,22 +80,4 @@ public static String makeStackTrace(Throwable throwable) {
         printStream.close(); // closes baOutStream
         return text;
     }
-
-    /**
-     * Used to indicate that we are not, in fact, using a variable, also to silence compiler warnings.
-     *
-     * @param value the object we're pretending to use
-     */
-    public static void use(Object value) {
-        if (value != null) {
-            return;
-        }
-    }
-
-    /**
-     * Similarly, for situations where an empty statement is required
-     */
-    public static void emptyStatement() {
-        use(null);
-    }
 }