Use queue name if necessary as routing key in consumers

Otherwise stats latency is never updated and received
messages metrics aren't not displayed.

References #45

Author: acogoluegnes

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

@@ -35,7 +35,7 @@ public class Consumer extends ProducerConsumerBase implements Runnable {
 
     private ConsumerImpl                q;
     private final Channel               channel;
-    private final String                routingKey;
+    private String                      routingKey;
     private final List<String>          queueNames;
     private final int                   txSize;
     private final boolean               autoAck;
@@ -94,6 +94,10 @@ public void run() {
         }
     }
 
+    public void setRoutingKey(String routingKey) {
+        this.routingKey = routingKey;
+    }
+
     private class ConsumerImpl extends DefaultConsumer {
         long now;
         int totalMsgCount = 0;
@@ -128,7 +132,6 @@ 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,6 +22,8 @@
 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;
@@ -74,6 +76,7 @@ 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++) {
@@ -88,6 +91,7 @@ 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;
@@ -111,6 +115,12 @@ 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++) {