Merge in default

Author: Steve Powell

codegen.py

@@ -264,12 +264,12 @@ def printWritePropertiesTo(c):
                 print "            if (this.%s != null) writer.write%s(this.%s);" % (jfName, jfClass, jfName)
         print "        }"
 
-    def printAppendArgumentDebugStringTo(c):
+    def printAppendPropertyDebugStringTo(c):
         appendList = [ "%s=\")\n               .append(this.%s)\n               .append(\""
                        % (f.name, java_field_name(f.name))
                        for f in c.fields ]
         print
-        print "        public void appendArgumentDebugStringTo(StringBuilder acc) {"
+        print "        public void appendPropertyDebugStringTo(StringBuilder acc) {"
         print "            acc.append(\"(%s)\");" % (", ".join(appendList))
         print "        }"
 
@@ -386,7 +386,7 @@ def printSetter(fieldType, fieldName):
             printSetter(jType, jName)
 
         printWritePropertiesTo(c)
-        printAppendArgumentDebugStringTo(c)
+        printAppendPropertyDebugStringTo(c)
         printPropertiesBuilderClass(c)
 
         print "    }"

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/AMQCommand.java

@@ -146,7 +146,7 @@ private static StringBuilder contentBodyStringBuilder(byte[] body, boolean suppr
             if (suppressBody) {
                 return new StringBuilder().append(body.length).append(" bytes of payload");
             } else {
-                return new StringBuilder().append('\"').append(body).append('\"');
+                return new StringBuilder().append('\"').append(new String(body, "UTF-8")).append('\"');
             }
         } catch (Exception e) {
             return new StringBuilder().append('|').append(body.length).append('|');

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());
@@ -462,10 +462,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,
@@ -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;
             }
         };
 

test/src/com/rabbitmq/examples/ConsumerMain.java

@@ -188,13 +188,13 @@ public void report(boolean writeStats) throws IOException {
                 sumL += v;
             }
 
-            double avgL = sumL / messageCount;
-            System.out.println("CONSUMER -  Message count: " + messageCount);
-            System.out.println("Total time, milliseconds: " + totalDelta);
-            System.out.println("Overall messages-per-second: " + (messageCount / (totalDelta / 1000.0)));
-            System.out.println("Min latency, milliseconds: " + minL);
-            System.out.println("Avg latency, milliseconds: " + avgL);
-            System.out.println("Max latency, milliseconds: " + maxL);
+            System.out.println("CONSUMER -       Overall: "
+                            + String.format("%d messages in %dms, a rate of %.2f msgs/sec", messageCount,
+                                    totalDelta,
+                                    (messageCount / (totalDelta / 1000.0))));
+            System.out.println("Latency  - Min (Avg) Max: "
+                    + String.format("%dms (%.2fms) %dms", minL, sumL
+                            / messageCount, maxL));
 
             if (writeStats) {
                 PrintStream o = new PrintStream(new FileOutputStream("simple-latency-experiment.csv"));