Revert "Use queue name if necessary as routing key in consumers"

This reverts commit 8199bf7.

See #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 String                      routingKey;
+    private final String                routingKey;
     private final List<String>          queueNames;
     private final int                   txSize;
     private final boolean               autoAck;
@@ -92,10 +92,6 @@ public void run() {
         }
     }
 
-    public void setRoutingKey(String routingKey) {
-        this.routingKey = routingKey;
-    }
-
     private class ConsumerImpl extends DefaultConsumer {
         long now;
         int totalMsgCount = 0;
@@ -135,6 +131,7 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
                 }
 
                 now = System.currentTimeMillis();
+
                 stats.handleRecv(routingKey.equals(envelope.getRoutingKey()) ? (nano - msgNano) : 0L);
                 if (rateLimit > 0.0f) {
                     delay(now);

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

@@ -22,8 +22,6 @@
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Random;
@@ -76,7 +74,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<>();
-        Collection<Consumer> consumers = new ArrayList<>();
         Thread[] consumerThreads = new Thread[params.getConsumerThreadCount()];
         Connection[] consumerConnections = new Connection[params.getConsumerCount()];
         for (int i = 0; i < consumerConnections.length; i++) {
@@ -91,7 +88,6 @@ public void run(boolean announceStartup)
                     System.out.println("id: " + testID + ", starting consumer #" + i + ", channel #" + j);
                 }
                 Consumer consumer = params.createConsumer(conn, stats, routingKey);
-                consumers.add(consumer);
                 queueNames.addAll(consumer.getQueueNames());
                 Thread t = new Thread(consumer);
                 consumerThreads[(i * params.getConsumerChannelCount()) + j] = t;
@@ -115,12 +111,6 @@ public void run(boolean announceStartup)
             producersRoutingKey = this.routingKey;
         }
 
-        // consumers need the publishing routing key to match it against
-        // against the received message and update the stats
-        for (Consumer consumer : consumers) {
-            consumer.setRoutingKey(producersRoutingKey);
-        }
-
         Thread[] producerThreads = new Thread[params.getProducerThreadCount()];
         Connection[] producerConnections = new Connection[params.getProducerCount()];
         for (int i = 0; i < producerConnections.length; i++) {