merge from default

Author: Alexandru Scvortov

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

@@ -32,18 +32,22 @@
 import java.io.IOException;
 
 public class ConfirmDontLoseMessages {
-    final static int MSG_COUNT = 10000;
+    static int msgCount = 10000;
     final static String QUEUE_NAME = "confirm-test";
     static ConnectionFactory connectionFactory;
 
     public static void main(String[] args)
         throws IOException, InterruptedException
     {
+        if (args.length > 0) {
+                msgCount = Integer.parseInt(args[0]);
+        }
+
         connectionFactory = new ConnectionFactory();
 
-        // Publish MSG_COUNT messages and wait for confirms.
+        // Consume msgCount messages.
         (new Thread(new Consumer())).start();
-        // Consume MSG_COUNT messages.
+        // Publish msgCount messages and wait for confirms.
         (new Thread(new Publisher())).start();
     }
 
@@ -58,8 +62,7 @@ public void run() {
                 // Setup
                 Connection conn = connectionFactory.newConnection();
                 Channel ch = conn.createChannel();
-                ch.queueDeclare(QUEUE_NAME, true, false, true, null);
-                ch.confirmSelect();
+                ch.queueDeclare(QUEUE_NAME, true, false, false, null);
                 ch.setConfirmListener(new ConfirmListener() {
                         public void handleAck(long seqNo, boolean multiple) {
                             if (multiple) {
@@ -84,9 +87,10 @@ public void handleNack(long seqNo, boolean multiple) {
                                               lost);
                         }
                     });
+                ch.confirmSelect();
 
                 // Publish
-                for (long i = 0; i < MSG_COUNT; ++i) {
+                for (long i = 0; i < msgCount; ++i) {
                     unconfirmedSet.add(ch.getNextPublishSeqNo());
                     ch.basicPublish("", QUEUE_NAME,
                                     MessageProperties.PERSISTENT_BASIC,
@@ -98,6 +102,7 @@ public void handleNack(long seqNo, boolean multiple) {
                     Thread.sleep(10);
 
                 // Cleanup
+                ch.queueDelete(QUEUE_NAME);
                 ch.close();
                 conn.close();
 
@@ -117,16 +122,16 @@ public void run() {
                 // Setup
                 Connection conn = connectionFactory.newConnection();
                 Channel ch = conn.createChannel();
-                ch.queueDeclare(QUEUE_NAME, true, false, true, null);
+                ch.queueDeclare(QUEUE_NAME, true, false, false, null);
 
                 // Consume
                 QueueingConsumer qc = new QueueingConsumer(ch);
                 ch.basicConsume(QUEUE_NAME, true, qc);
-                for (int i = 0; i < MSG_COUNT; ++i) {
+                for (int i = 0; i < msgCount; ++i) {
                     qc.nextDelivery();
                 }
 
-                // Consume
+                // Cleanup
                 ch.close();
                 conn.close();
             } catch (Throwable e) {