Merging bug19662 into bug21673

Author: Matthew Sackman

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

@@ -117,8 +117,13 @@ private void runIt() throws IOException {
         String queueName = "test queue";
         channel.queueDeclare(queueName, true, false, false, null);
 
+        String testExchangeName = "test exchange";
+        channel.exchangeDeclare(testExchangeName, "fanout");
+
+        channel.queueBind(queueName, testExchangeName, "");
+
         String exchangeName = "test completion";
-        channel.exchangeDeclare(exchangeName, "fanout", false, false, null);
+        channel.exchangeDeclare(exchangeName, "fanout");
 
         String completionQueue = channel.queueDeclare().getQueue();
         channel.queueBind(completionQueue, exchangeName, "");
@@ -135,6 +140,9 @@ private void runIt() throws IOException {
 
         System.out.println("Deleting completion queue.");
         channel.queueDelete(completionQueue);
+        
+        System.out.println("Deleting test exchange.");
+        channel.exchangeDelete(testExchangeName);
 
         System.out.println("Closing the channel.");
         channel.close();

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

@@ -40,6 +40,7 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.AMQP.BasicProperties;
 
 public class ProducerMain implements Runnable {
     public static final int SUMMARY_EVERY_MS = 1000;
@@ -150,15 +151,19 @@ private void runIt() throws IOException {
         _channel = _connection.createChannel();
 
         String queueName = "test queue";
+        String exchangeName = "test exchange";
+        
         _channel.queueDeclare(queueName, true, false, false, null);
+        _channel.exchangeDeclare(exchangeName, "fanout");
+        _channel.queueBind(queueName, exchangeName, "");
 
         if (shouldCommit()) {
             _channel.txSelect();
         }
-        sendBatch(queueName);
+        sendBatch(exchangeName, queueName);
 
         if (_sendCompletion) {
-            String exchangeName = "test completion";
+            exchangeName = "test completion";
             _channel.exchangeDeclarePassive(exchangeName);
             _channel.basicPublish(exchangeName, "", MessageProperties.BASIC, new byte[0]);
             if (shouldCommit())
@@ -181,7 +186,7 @@ public void primeServer(String queueName) throws IOException {
         System.out.println("...starting.");
     }
 
-    public void sendBatch(String queueName) throws IOException {
+    public void sendBatch(String exchangeName, String queueName) throws IOException {
         //primeServer(queueName);
 
         long startTime = System.currentTimeMillis();
@@ -191,6 +196,9 @@ public void sendBatch(String queueName) throws IOException {
 
         long nextSummaryTime = startTime + SUMMARY_EVERY_MS;
         byte[] message = new byte[256];
+        BasicProperties props = shouldPersist() ?
+                MessageProperties.MINIMAL_PERSISTENT_BASIC :
+                    MessageProperties.MINIMAL_BASIC;
         for (int i = 0; i < _messageCount; i++) {
             ByteArrayOutputStream acc = new ByteArrayOutputStream();
             DataOutputStream d = new DataOutputStream(acc);
@@ -205,7 +213,7 @@ public void sendBatch(String queueName) throws IOException {
             acc.flush();
             byte[] message0 = acc.toByteArray();
             System.arraycopy(message0, 0, message, 0, message0.length);
-            _channel.basicPublish("", queueName, shouldPersist() ? MessageProperties.MINIMAL_PERSISTENT_BASIC : MessageProperties.MINIMAL_BASIC,
+            _channel.basicPublish(exchangeName, queueName, props,
                     message);
             sent++;
             if (shouldCommit()) {