Merge pull request #57 from dmcintyre-pivotal/master

Support for testing sharded queues

Author: michaelklishin

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

@@ -42,12 +42,14 @@ public class MulticastParams {
     private float consumerRateLimit = 0;
     private int producerMsgCount = 0;
     private int consumerMsgCount = 0;
+    private boolean consumerSlowStart = false;
 
     private String exchangeName = "direct";
     private String exchangeType = "direct";
     private List<String> queueNames = new ArrayList<String>();
     private String routingKey = null;
     private boolean randomRoutingKey = false;
+    private boolean skipBindingQueues = false;
 
     private List<?> flags = new ArrayList<Object>();
 
@@ -87,6 +89,10 @@ public void setRoutingKey(String routingKey) {
     public void setRandomRoutingKey(boolean randomRoutingKey) {
         this.randomRoutingKey = randomRoutingKey;
     }
+    
+    public void setSkipBindingQueues(boolean skipBindingQueues) {
+    	this.skipBindingQueues = skipBindingQueues;
+    }
 
     public void setProducerRateLimit(float producerRateLimit) {
         this.producerRateLimit = producerRateLimit;
@@ -111,6 +117,10 @@ public void setConsumerCount(int consumerCount) {
     public void setConsumerChannelCount(int consumerChannelCount) {
         this.consumerChannelCount = consumerChannelCount;
     }
+    
+    public void setConsumerSlowStart(boolean slowStart) {
+        this.consumerSlowStart = slowStart;
+    }
 
     public void setProducerTxSize(int producerTxSize) {
         this.producerTxSize = producerTxSize;
@@ -188,6 +198,10 @@ public int getConsumerCount() {
     public int getConsumerChannelCount() {
         return consumerChannelCount;
     }
+    
+    public boolean getConsumerSlowStart() {
+        return consumerSlowStart;
+    }
 
     public int getConsumerThreadCount() {
         return consumerCount * consumerChannelCount;
@@ -216,6 +230,10 @@ public String getRoutingKey() {
     public boolean getRandomRoutingKey() {
         return randomRoutingKey;
     }
+    
+    public boolean getSkipBindingQueues() {
+    	return skipBindingQueues;
+    }
 
     public void setBodyFiles(List<String> bodyFiles) {
         if (bodyFiles == null) {
@@ -294,7 +312,7 @@ public List<String> configureQueues(Connection connection, String id) throws IOE
             generatedQueueNames.add(qName);
             // skipping binding to default exchange,
             // as it's not possible to explicitly bind to it.
-            if (!"".equals(exchangeName) && !"amq.default".equals(exchangeName)) {
+            if (!"".equals(exchangeName) && !"amq.default".equals(exchangeName) && !skipBindingQueues) {
                 channel.queueBind(qName, exchangeName, id);
             }
         }

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

@@ -116,6 +116,10 @@ public void run(boolean announceStartup)
 
         for (Thread consumerThread : consumerThreads) {
             consumerThread.start();
+            if(params.getConsumerSlowStart()) {
+            	System.out.println("Delaying start by 1 second because -S/--slow-start was requested");
+            	Thread.sleep(1000);
+            }
         }
 
         for (Thread producerThread : producerThreads) {

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

@@ -63,6 +63,7 @@ public static void main(String[] args) {
             String queueNames        = strArg(cmd, 'u', null);
             String routingKey        = strArg(cmd, 'k', null);
             boolean randomRoutingKey = cmd.hasOption('K');
+            boolean skipBindingQueues= cmd.hasOption("sb");
             int samplingInterval     = intArg(cmd, 'i', 1);
             float producerRateLimit  = floatArg(cmd, 'r', 0.0f);
             float consumerRateLimit  = floatArg(cmd, 'R', 0.0f);
@@ -78,6 +79,7 @@ public static void main(String[] args) {
             int channelPrefetch      = intArg(cmd, 'Q', 0);
             int consumerPrefetch     = intArg(cmd, 'q', 0);
             int minMsgSize           = intArg(cmd, 's', 0);
+            boolean slowStart        = cmd.hasOption('S');
             int timeLimit            = intArg(cmd, 'z', 0);
             int producerMsgCount     = intArg(cmd, 'C', 0);
             int consumerMsgCount     = intArg(cmd, 'D', 0);
@@ -154,6 +156,7 @@ public void run() {
             p.setConsumerMsgCount(      consumerMsgCount);
             p.setConsumerRateLimit(     consumerRateLimit);
             p.setConsumerTxSize(        consumerTxSize);
+            p.setConsumerSlowStart(     slowStart);
             p.setExchangeName(          exchangeName);
             p.setExchangeType(          exchangeType);
             p.setFlags(                 flags);
@@ -168,6 +171,7 @@ public void run() {
             p.setProducerTxSize(        producerTxSize);
             p.setQueueNames(            queueNames == null ? null : asList(queueNames.split(",")));
             p.setRoutingKey(            routingKey);
+            p.setSkipBindingQueues(     skipBindingQueues);
             p.setRandomRoutingKey(      randomRoutingKey);
             p.setProducerRateLimit(     producerRateLimit);
             p.setTimeLimit(             timeLimit);
@@ -233,11 +237,13 @@ private static Options getOptions() {
         options.addOption(new Option("u", "queue",                  true, "queue name"));
         options.addOption(new Option("k", "routing-key",            true, "routing key"));
         options.addOption(new Option("K", "random-routing-key",     false,"use random routing key per message"));
+        options.addOption(new Option("sb", "skip-binding-queues",   false,"don't bind queues to the exchange"));
         options.addOption(new Option("i", "interval",               true, "sampling interval in seconds"));
         options.addOption(new Option("r", "rate",                   true, "producer rate limit"));
         options.addOption(new Option("R", "consumer-rate",          true, "consumer rate limit"));
         options.addOption(new Option("x", "producers",              true, "producer count"));
         options.addOption(new Option("y", "consumers",              true, "consumer count"));
+        options.addOption(new Option("S", "slow-start",             false,"start consumers slowly (1 sec delay between each)"));
         options.addOption(new Option("X", "producer-channel-count", true, "channels per producer"));
         options.addOption(new Option("Y", "consumer-channel-count", true, "channels per consumer"));
         options.addOption(new Option("m", "ptxsize",                true, "producer tx size"));