add nack handling to MCM

Author: Alexandru Scvortov

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

@@ -46,6 +46,7 @@
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.NackListener;
 import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.ReturnListener;
 import com.rabbitmq.client.ShutdownSignalException;
@@ -140,6 +141,7 @@ public static void main(String[] args) {
                                                 confirm, confirmMax);
                 channel.setReturnListener(p);
                 channel.setAckListener(p);
+                channel.setNackListener(p);
                 Thread t = new Thread(p);
                 producerThreads[i] = t;
                 t.start();
@@ -214,7 +216,8 @@ private static List lstArg(CommandLine cmd, char opt) {
         return Arrays.asList(vals);
     }
 
-    public static class Producer implements Runnable, ReturnListener, AckListener {
+    public static class Producer implements Runnable, ReturnListener, AckListener,
+                                            NackListener {
 
         private Channel channel;
         private String  exchangeName;
@@ -236,6 +239,7 @@ public static class Producer implements Runnable, ReturnListener, AckListener {
 
         private boolean   confirm;
         private long      confirmCount;
+        private long      nackCount;
         private Semaphore confirmPool;
         private volatile SortedSet<Long> ackSet =
             Collections.synchronizedSortedSet(new TreeSet<Long>());
@@ -274,6 +278,15 @@ public synchronized void handleBasicReturn(int replyCode,
         }
 
         public void handleAck(long seqNo, boolean multiple) {
+            handleAckNack(seqNo, multiple, false);
+        }
+
+        public void handleNack(long seqNo, boolean multiple) {
+            handleAckNack(seqNo, multiple, true);
+        }
+
+        private void handleAckNack(long seqNo, boolean multiple,
+                                   boolean nack) {
             int numConfirms = 0;
             if (multiple) {
                 SortedSet<Long> confirmed = ackSet.headSet(seqNo + 1);
@@ -284,14 +297,19 @@ public void handleAck(long seqNo, boolean multiple) {
                 numConfirms = 1;
             }
             synchronized (this) {
-                confirmCount += numConfirms;
+                if (nack) {
+                    nackCount += numConfirms;
+                } else {
+                    confirmCount += numConfirms;
+                }
             }
 
             if (confirmPool != null) {
                 for (int i = 0; i < numConfirms; ++i) {
                     confirmPool.release();
                 }
             }
+
         }
 
         public void run() {
@@ -354,21 +372,26 @@ private void delay(long now)
                 Thread.sleep(pause);
             }
             if (elapsed > interval) {
-                long sendRate, returnRate, confirmRate;
+                long sendRate, returnRate, confirmRate, nackRate;
                 synchronized(this) {
                     sendRate     = msgCount     * 1000L / elapsed;
                     returnRate   = returnCount  * 1000L / elapsed;
                     confirmRate  = confirmCount * 1000L / elapsed;
+                    nackRate     = nackCount    * 1000L / elapsed;
                     msgCount     = 0;
                     returnCount  = 0;
                     confirmCount = 0;
+                    nackCount    = 0;
                 }
                 System.out.print("sending rate: " + sendRate + " msg/s");
                 if (mandatory || immediate) {
                     System.out.print(", returns: " + returnRate + " ret/s");
                 }
                 if (confirm) {
                     System.out.print(", confirms: " + confirmRate + " c/s");
+                    if (nackRate > 0) {
+                        System.out.print(", nacks: " + nackRate + " n/s");
+                    }
                 }
                 System.out.println();
                 lastStatsTime = now;