Revert "Use default exchange by default"

This reverts commit 2ae0c79.

Closes #54.

References #45.

Author: michaelklishin

src/main/java/com/rabbitmq/perf/Consumer.java

@@ -33,7 +33,7 @@ public class Consumer extends ProducerConsumerBase implements Runnable {
 
     private ConsumerImpl                q;
     private final Channel               channel;
-    private final String                routingKey;
+    private final String                id;
     private final List<String>          queueNames;
     private final int                   txSize;
     private final boolean               autoAck;
@@ -45,14 +45,14 @@ public class Consumer extends ProducerConsumerBase implements Runnable {
     private final Map<String, String>   consumerTagBranchMap = Collections.synchronizedMap(new HashMap<String, String>());
     private final ConsumerLatency       consumerLatency;
 
-    public Consumer(Channel channel, String routingKey,
+    public Consumer(Channel channel, String id,
                     List<String> queueNames, int txSize, boolean autoAck,
                     int multiAckEvery, Stats stats, float rateLimit, int msgLimit, int timeLimit,
                     int consumerLatencyInMicroSeconds) {
 
         this.channel       = channel;
-        this.routingKey    = routingKey;
-        this.queueNames    = Collections.unmodifiableList(queueNames);
+        this.id            = id;
+        this.queueNames    = queueNames;
         this.rateLimit     = rateLimit;
         this.txSize        = txSize;
         this.autoAck       = autoAck;
@@ -132,7 +132,7 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
 
                 now = System.currentTimeMillis();
 
-                stats.handleRecv(routingKey.equals(envelope.getRoutingKey()) ? (nano - msgNano) : 0L);
+                stats.handleRecv(id.equals(envelope.getRoutingKey()) ? (nano - msgNano) : 0L);
                 if (rateLimit > 0.0f) {
                     delay(now);
                 }
@@ -161,10 +161,6 @@ public void handleCancel(String consumerTag) throws IOException {
         }
     }
 
-    public List<String> getQueueNames() {
-        return this.queueNames;
-    }
-
     private interface ConsumerLatency {
 
         void simulateLatency();

src/main/java/com/rabbitmq/perf/MulticastParams.java

@@ -43,19 +43,19 @@ public class MulticastParams {
     private int producerMsgCount = 0;
     private int consumerMsgCount = 0;
 
-    private String exchangeName = "";
+    private String exchangeName = "direct";
     private String exchangeType = "direct";
-    private List<String> queueNames = new ArrayList<>();
+    private List<String> queueNames = new ArrayList<String>();
     private String routingKey = null;
     private boolean randomRoutingKey = false;
 
-    private List<?> flags = new ArrayList<>();
+    private List<?> flags = new ArrayList<Object>();
 
     private int multiAckEvery = 0;
     private boolean autoAck = true;
     private boolean autoDelete = false;
 
-    private List<String> bodyFiles = new ArrayList<>();
+    private List<String> bodyFiles = new ArrayList<String>();
     private String bodyContentType = null;
 
     private boolean predeclared;
@@ -74,9 +74,9 @@ public void setExchangeName(String exchangeName) {
 
     public void setQueueNames(List<String> queueNames) {
         if(queueNames == null) {
-            this.queueNames = new ArrayList<>();
+            this.queueNames = new ArrayList<String>();
         } else {
-            this.queueNames = new ArrayList<>(queueNames);
+            this.queueNames = new ArrayList<String>(queueNames);
         }
     }
 
@@ -219,21 +219,21 @@ public boolean getRandomRoutingKey() {
 
     public void setBodyFiles(List<String> bodyFiles) {
         if (bodyFiles == null) {
-            this.bodyFiles = new ArrayList<>();
+            this.bodyFiles = new ArrayList<String>();
         } else {
-            this.bodyFiles = new ArrayList<>(bodyFiles);
+            this.bodyFiles = new ArrayList<String>(bodyFiles);
         }
     }
 
     public void setBodyContentType(String bodyContentType) {
         this.bodyContentType = bodyContentType;
     }
 
-    public Producer createProducer(Connection connection, Stats stats, String routingKey) throws IOException {
+    public Producer createProducer(Connection connection, Stats stats, String id) throws IOException {
         Channel channel = connection.createChannel();
         if (producerTxSize > 0) channel.txSelect();
         if (confirm >= 0) channel.confirmSelect();
-        if (!predeclared && !exchangeExists(connection, exchangeName)) {
+        if (!predeclared || !exchangeExists(connection, exchangeName)) {
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
         MessageBodySource messageBodySource = null;
@@ -242,7 +242,7 @@ public Producer createProducer(Connection connection, Stats stats, String routin
         } else {
             messageBodySource = new TimeSequenceMessageBodySource(minMsgSize);
         }
-        final Producer producer = new Producer(channel, exchangeName, routingKey,
+        final Producer producer = new Producer(channel, exchangeName, id,
                                                randomRoutingKey, flags, producerTxSize,
                                                producerRateLimit, producerMsgCount,
                                                timeLimit,
@@ -252,24 +252,26 @@ public Producer createProducer(Connection connection, Stats stats, String routin
         return producer;
     }
 
-    public Consumer createConsumer(Connection connection, Stats stats, String routingKey) throws IOException {
+    public Consumer createConsumer(Connection connection, Stats stats, String id) throws IOException {
         Channel channel = connection.createChannel();
         if (consumerTxSize > 0) channel.txSelect();
-        List<String> generatedQueueNames = configureQueues(connection, routingKey);
+        List<String> generatedQueueNames = configureQueues(connection, id);
         if (consumerPrefetch > 0) channel.basicQos(consumerPrefetch);
         if (channelPrefetch > 0) channel.basicQos(channelPrefetch, true);
-        return new Consumer(channel, routingKey, generatedQueueNames,
+        return new Consumer(channel, id, generatedQueueNames,
                                          consumerTxSize, autoAck, multiAckEvery,
                                          stats, consumerRateLimit, consumerMsgCount, timeLimit, consumerLatencyInMicroseconds);
     }
 
     public boolean shouldConfigureQueues() {
+        // don't declare any queues when --predeclared is passed,
+        // otherwise unwanted server-named queues without consumers will pile up. MK.
         return consumerCount == 0 && !(queueNames.size() == 0);
     }
 
-    public List<String> configureQueues(Connection connection, String routingKey) throws IOException {
+    public List<String> configureQueues(Connection connection, String id) throws IOException {
         Channel channel = connection.createChannel();
-        if (!predeclared && !exchangeExists(connection, exchangeName)) {
+        if (!predeclared || !exchangeExists(connection, exchangeName)) {
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
         // To ensure we get at-least 1 default queue:
@@ -293,7 +295,7 @@ public List<String> configureQueues(Connection connection, String routingKey) th
             // skipping binding to default exchange,
             // as it's not possible to explicitly bind to it.
             if (!"".equals(exchangeName) && !"amq.default".equals(exchangeName)) {
-                channel.queueBind(qName, exchangeName, routingKey);
+                channel.queueBind(qName, exchangeName, id);
             }
         }
         channel.abort();
@@ -306,12 +308,20 @@ private static boolean exchangeExists(Connection connection, final String exchan
             // NB: default exchange always exists
             return true;
         } else {
-            return exists(connection, ch -> ch.exchangeDeclarePassive(exchangeName));
+            return exists(connection, new Checker() {
+                public void check(Channel ch) throws IOException {
+                    ch.exchangeDeclarePassive(exchangeName);
+                }
+            });
         }
     }
 
     private static boolean queueExists(Connection connection, final String queueName) throws IOException {
-        return queueName != null && exists(connection, ch -> ch.queueDeclarePassive(queueName));
+        return queueName != null && exists(connection, new Checker() {
+            public void check(Channel ch) throws IOException {
+                ch.queueDeclarePassive(queueName);
+            }
+        });
     }
 
     private interface Checker {

src/main/java/com/rabbitmq/perf/MulticastSet.java

@@ -22,15 +22,13 @@
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Random;
-import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeoutException;
 
 public class MulticastSet {
-    private final String routingKey;
+    private final String id;
     private final Stats stats;
     private final ConnectionFactory factory;
     private final MulticastParams params;
@@ -42,9 +40,9 @@ public class MulticastSet {
     public MulticastSet(Stats stats, ConnectionFactory factory,
         MulticastParams params, List<String> uris) {
         if (params.getRoutingKey() == null) {
-            this.routingKey = UUID.randomUUID().toString();
+            this.id = UUID.randomUUID().toString();
         } else {
-            this.routingKey = params.getRoutingKey();
+            this.id = params.getRoutingKey();
         }
         this.stats = stats;
         this.factory = factory;
@@ -56,9 +54,9 @@ public MulticastSet(Stats stats, ConnectionFactory factory,
     public MulticastSet(Stats stats, ConnectionFactory factory,
         MulticastParams params, String testID, List<String> uris) {
         if (params.getRoutingKey() == null) {
-            this.routingKey = UUID.randomUUID().toString();
+            this.id = UUID.randomUUID().toString();
         } else {
-            this.routingKey = params.getRoutingKey();
+            this.id = params.getRoutingKey();
         }
         this.stats = stats;
         this.factory = factory;
@@ -73,7 +71,6 @@ public void run() throws IOException, InterruptedException, TimeoutException, No
 
     public void run(boolean announceStartup)
         throws IOException, InterruptedException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException {
-        Set<String> queueNames = new HashSet<>();
         Thread[] consumerThreads = new Thread[params.getConsumerThreadCount()];
         Connection[] consumerConnections = new Connection[params.getConsumerCount()];
         for (int i = 0; i < consumerConnections.length; i++) {
@@ -87,30 +84,18 @@ public void run(boolean announceStartup)
                 if (announceStartup) {
                     System.out.println("id: " + testID + ", starting consumer #" + i + ", channel #" + j);
                 }
-                Consumer consumer = params.createConsumer(conn, stats, routingKey);
-                queueNames.addAll(consumer.getQueueNames());
-                Thread t = new Thread(consumer);
+                Thread t = new Thread(params.createConsumer(conn, stats, id));
                 consumerThreads[(i * params.getConsumerChannelCount()) + j] = t;
             }
         }
 
         if (params.shouldConfigureQueues()) {
             setUri();
             Connection conn = factory.newConnection();
-            List<String> configuredQueues = params.configureQueues(conn, routingKey);
-            queueNames.addAll(configuredQueues);
+            params.configureQueues(conn, id);
             conn.close();
         }
 
-        String producersRoutingKey;
-        // if one queue, and no routing key, the routing key
-        // must be the queue name and we should use the default exchange
-        if (queueNames.size() == 1 && params.getRoutingKey() == null) {
-            producersRoutingKey = queueNames.iterator().next();
-        } else {
-            producersRoutingKey = this.routingKey;
-        }
-
         Thread[] producerThreads = new Thread[params.getProducerThreadCount()];
         Connection[] producerConnections = new Connection[params.getProducerCount()];
         for (int i = 0; i < producerConnections.length; i++) {
@@ -124,7 +109,7 @@ public void run(boolean announceStartup)
                 if (announceStartup) {
                     System.out.println("id: " + testID + ", starting producer #" + i + ", channel #" + j);
                 }
-                Thread t = new Thread(params.createProducer(conn, stats, producersRoutingKey));
+                Thread t = new Thread(params.createProducer(conn, stats, id));
                 producerThreads[(i * params.getProducerChannelCount()) + j] = t;
             }
         }

src/main/java/com/rabbitmq/perf/PerfTest.java

@@ -59,7 +59,7 @@ public static void main(String[] args) {
                 getInstance().getTime());
             testID                   = strArg(cmd, 'd', "test-"+testID);
             String exchangeType      = strArg(cmd, 't', "direct");
-            String exchangeName      = getExchangeName(cmd, "");
+            String exchangeName      = getExchangeName(cmd, exchangeType);
             String queueNames        = strArg(cmd, 'u', null);
             String routingKey        = strArg(cmd, 'k', null);
             boolean randomRoutingKey = cmd.hasOption('K');