Merged bug19492 into default

Author: Emile Joubert

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

@@ -56,6 +56,7 @@
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.MessageProperties;
 import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.ReturnListener;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.AMQP.Queue;
 import com.rabbitmq.client.QueueingConsumer.Delivery;
@@ -126,11 +127,22 @@ public static void main(String[] args) {
                 Channel channel = conn.createChannel();
                 if (producerTxSize > 0) channel.txSelect();
                 channel.exchangeDeclare(exchangeName, exchangeType);
-                Thread t =
-                    new Thread(new Producer(channel, exchangeName, id,
-                                            flags, producerTxSize,
-                                            1000L * samplingInterval,
-                                            rateLimit, minMsgSize, timeLimit));
+                final Producer p = new Producer(channel, exchangeName, id,
+                                          flags, producerTxSize,
+                                          1000L * samplingInterval,
+                                          rateLimit, minMsgSize, timeLimit);
+                channel.setReturnListener(new ReturnListener() {
+                        public void handleBasicReturn(int replyCode,
+                                                      String replyText,
+                                                      String exchange,
+                                                      String routingKey,
+                                                      AMQP.BasicProperties properties,
+                                                      byte[] body)
+                            throws IOException {
+                            p.logBasicReturn();
+                        }
+                    });
+                Thread t = new Thread(p);
                 producerThreads[i] = t;
                 t.start();
             }
@@ -215,6 +227,7 @@ public static class Producer implements Runnable {
         private long    startTime;
         private long    lastStatsTime;
         private int     msgCount;
+        private int     basicReturnCount;
 
         public Producer(Channel channel, String exchangeName, String id,
                         List flags, int txSize,
@@ -234,6 +247,14 @@ public Producer(Channel channel, String exchangeName, String id,
             this.message      = new byte[minMsgSize];
         }
 
+        public synchronized void logBasicReturn() {
+            basicReturnCount++;
+        }
+
+        public synchronized void resetBasicReturns() {
+            basicReturnCount = 0;
+        }
+
         public void run() {
 
             long now;
@@ -292,7 +313,11 @@ private void delay(long now)
             if (elapsed > interval) {
                 System.out.println("sending rate: " +
                                    (msgCount * 1000L / elapsed) +
-                                   " msg/s");
+                                   " msg/s" +
+                                   ", basic returns: " +
+                                   (basicReturnCount * 1000L / elapsed) +
+                                   " ret/s");
+                resetBasicReturns();
                 msgCount = 0;
                 lastStatsTime = now;
             }