adds random routing key parameter to PerfTest

Author: Alvaro Videla

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

@@ -49,6 +49,7 @@ public static void main(String[] args) {
             String exchangeName     = strArg(cmd, 'e', exchangeType);
             String queueName        = strArg(cmd, 'u', "");
             String routingKey       = strArg(cmd, 'k', null);
+            boolean randomRKey      = cmd.hasOption('K');
             int samplingInterval    = intArg(cmd, 'i', 1);
             float producerRateLimit = floatArg(cmd, 'r', 0.0f);
             float consumerRateLimit = floatArg(cmd, 'R', 0.0f);
@@ -107,6 +108,7 @@ public static void main(String[] args) {
             p.setProducerTxSize(   producerTxSize);
             p.setQueueName(        queueName);
             p.setRoutingKey(       routingKey);
+            p.setRandomRKey(       randomRKey);
             p.setProducerRateLimit(producerRateLimit);
             p.setTimeLimit(        timeLimit);
 
@@ -138,6 +140,7 @@ private static Options getOptions() {
         options.addOption(new Option("e", "exchange",      true, "exchange name"));
         options.addOption(new Option("u", "queue",         true, "queue name"));
         options.addOption(new Option("k", "routingKey",    true, "routing key"));
+        options.addOption(new Option("K", "randomRKey",    false,"use random routing key per message"));
         options.addOption(new Option("i", "interval",      true, "sampling interval"));
         options.addOption(new Option("r", "rate",          true, "producer rate limit"));
         options.addOption(new Option("R", "consumerRate",  true, "consumer rate limit"));

test/src/com/rabbitmq/examples/perf/MulticastParams.java

@@ -45,6 +45,7 @@ public class MulticastParams {
     private String exchangeType = "direct";
     private String queueName = "";
     private String routingKey = null;
+    private boolean randomRKey = false;
 
     private List<?> flags = new ArrayList<Object>();
 
@@ -70,6 +71,10 @@ public void setRoutingKey(String routingKey) {
         this.routingKey = routingKey;
     }
 
+    public void setRandomRKey(boolean randomRKey) {
+        this.randomRKey = randomRKey;
+    }
+
     public void setProducerRateLimit(float producerRateLimit) {
         this.producerRateLimit = producerRateLimit;
     }
@@ -163,6 +168,10 @@ public String getRoutingKey() {
         return routingKey;
     }
 
+    public boolean getRandomRKey() {
+        return randomRKey;
+    }
+
     public Producer createProducer(Connection connection, Stats stats, String id) throws IOException {
         Channel channel = connection.createChannel();
         if (producerTxSize > 0) channel.txSelect();
@@ -171,7 +180,7 @@ public Producer createProducer(Connection connection, Stats stats, String id) th
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
         final Producer producer = new Producer(channel, exchangeName, id,
-                                               flags, producerTxSize,
+                                               randomRKey, flags, producerTxSize,
                                                producerRateLimit, producerMsgCount,
                                                minMsgSize, timeLimit,
                                                confirm, stats);

test/src/com/rabbitmq/examples/perf/Producer.java

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.SortedSet;
 import java.util.TreeSet;
+import java.util.UUID;
 import java.util.concurrent.Semaphore;
 
 public class Producer extends ProducerConsumerBase implements Runnable, ReturnListener,
@@ -37,6 +38,7 @@ public class Producer extends ProducerConsumerBase implements Runnable, ReturnLi
     private Channel channel;
     private String  exchangeName;
     private String  id;
+    private boolean randomRKey;
     private boolean mandatory;
     private boolean immediate;
     private boolean persistent;
@@ -52,7 +54,7 @@ public class Producer extends ProducerConsumerBase implements Runnable, ReturnLi
     private volatile SortedSet<Long> unconfirmedSet =
         Collections.synchronizedSortedSet(new TreeSet<Long>());
 
-    public Producer(Channel channel, String exchangeName, String id,
+    public Producer(Channel channel, String exchangeName, String id, boolean randomRKey,
                     List<?> flags, int txSize,
                     float rateLimit, int msgLimit, int minMsgSize, int timeLimit,
                     long confirm, Stats stats)
@@ -61,6 +63,7 @@ public Producer(Channel channel, String exchangeName, String id,
         this.channel      = channel;
         this.exchangeName = exchangeName;
         this.id           = id;
+        this.randomRKey   = randomRKey;
         this.mandatory    = flags.contains("mandatory");
         this.immediate    = flags.contains("immediate");
         this.persistent   = flags.contains("persistent");
@@ -156,7 +159,7 @@ private void publish(byte[] msg)
         throws IOException {
 
         unconfirmedSet.add(channel.getNextPublishSeqNo());
-        channel.basicPublish(exchangeName, id,
+        channel.basicPublish(exchangeName, randomRKey ? UUID.randomUUID().toString() : id,
                              mandatory, immediate,
                              persistent ? MessageProperties.MINIMAL_PERSISTENT_BASIC : MessageProperties.MINIMAL_BASIC,
                              msg);