Merge bug24409 into default

Author: Steve Powell

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

@@ -18,12 +18,8 @@
 package com.rabbitmq.examples;
 
 import java.io.IOException;
-import java.util.Collections;
-import java.util.SortedSet;
-import java.util.TreeSet;
 
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.ConfirmListener;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.MessageProperties;
@@ -50,9 +46,6 @@ public static void main(String[] args)
     }
 
     static class Publisher implements Runnable {
-        private volatile SortedSet<Long> unconfirmedSet =
-            Collections.synchronizedSortedSet(new TreeSet<Long>());
-
         public void run() {
             try {
                 long startTime = System.currentTimeMillis();
@@ -61,43 +54,16 @@ public void run() {
                 Connection conn = connectionFactory.newConnection();
                 Channel ch = conn.createChannel();
                 ch.queueDeclare(QUEUE_NAME, true, false, false, null);
-                ch.addConfirmListener(new ConfirmListener() {
-                        public void handleAck(long seqNo, boolean multiple) {
-                            if (multiple) {
-                                unconfirmedSet.headSet(seqNo+1).clear();
-                            } else {
-                                unconfirmedSet.remove(seqNo);
-                            }
-                        }
-
-                        public void handleNack(long seqNo, boolean multiple) {
-                            int lost = 0;
-                            if (multiple) {
-                                SortedSet<Long> nackd =
-                                    unconfirmedSet.headSet(seqNo+1);
-                                lost = nackd.size();
-                                nackd.clear();
-                            } else {
-                                lost = 1;
-                                unconfirmedSet.remove(seqNo);
-                            }
-                            System.out.printf("Probably lost %d messages.\n",
-                                              lost);
-                        }
-                    });
                 ch.confirmSelect();
 
                 // Publish
                 for (long i = 0; i < msgCount; ++i) {
-                    unconfirmedSet.add(ch.getNextPublishSeqNo());
                     ch.basicPublish("", QUEUE_NAME,
                                     MessageProperties.PERSISTENT_BASIC,
                                     "nop".getBytes());
                 }
 
-                // Wait
-                while (unconfirmedSet.size() > 0)
-                    Thread.sleep(10);
+                ch.waitForConfirmsOrDie();
 
                 // Cleanup
                 ch.queueDelete(QUEUE_NAME);