Add consumer latency option

To simulate some processing on the consumer side.

Fixes #40

Author: acogoluegnes

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

@@ -45,10 +45,12 @@ public class Consumer extends ProducerConsumerBase implements Runnable {
     private final long                  timeLimit;
     private final CountDownLatch        latch = new CountDownLatch(1);
     private final Map<String, String>   consumerTagBranchMap = Collections.synchronizedMap(new HashMap<String, String>());
+    private final ConsumerLatency       consumerLatency;
 
     public Consumer(Channel channel, String id,
                     List<String> queueNames, int txSize, boolean autoAck,
-                    int multiAckEvery, Stats stats, float rateLimit, int msgLimit, int timeLimit) {
+                    int multiAckEvery, Stats stats, float rateLimit, int msgLimit, int timeLimit,
+                    int consumerLatencyInMicroSeconds) {
 
         this.channel       = channel;
         this.id            = id;
@@ -60,6 +62,13 @@ public Consumer(Channel channel, String id,
         this.stats         = stats;
         this.msgLimit      = msgLimit;
         this.timeLimit     = 1000L * timeLimit;
+        if (consumerLatencyInMicroSeconds <= 0) {
+            this.consumerLatency = new NoWaitConsumerLatency();
+        } else if (consumerLatencyInMicroSeconds >= 1000) {
+            this.consumerLatency = new ThreadSleepConsumerLatency(consumerLatencyInMicroSeconds / 1000);
+        } else {
+            this.consumerLatency = new BusyWaitConsumerLatency(consumerLatencyInMicroSeconds * 1000);
+        }
     }
 
     public void run() {
@@ -121,7 +130,10 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
                 now = System.currentTimeMillis();
 
                 stats.handleRecv(id.equals(envelope.getRoutingKey()) ? (nano - msgNano) : 0L);
-                delay(now);
+                if (rateLimit > 0.0f) {
+                    delay(now);
+                }
+                consumerLatency.simulateLatency();
             }
             if (msgLimit != 0 && msgCount >= msgLimit) { // NB: not quite the inverse of above
                 latch.countDown();
@@ -145,4 +157,53 @@ public void handleCancel(String consumerTag) throws IOException {
             }
         }
     }
+
+    private interface ConsumerLatency {
+
+        void simulateLatency();
+
+    }
+
+    private static class NoWaitConsumerLatency implements ConsumerLatency {
+
+        @Override
+        public void simulateLatency() {
+            // NO OP
+        }
+
+    }
+
+    private static class ThreadSleepConsumerLatency implements ConsumerLatency {
+
+        private final int waitTime;
+
+        private ThreadSleepConsumerLatency(int waitTime) {
+            this.waitTime = waitTime;
+        }
+
+        @Override
+        public void simulateLatency() {
+            try {
+                Thread.sleep(waitTime);
+            } catch (InterruptedException e) {
+                throw new RuntimeException("Exception while simulating latency", e);
+            }
+        }
+    }
+
+    // from https://stackoverflow.com/a/11499351
+    private static class BusyWaitConsumerLatency implements ConsumerLatency {
+
+        private final long delay;
+
+        private BusyWaitConsumerLatency(long delay) {
+            this.delay = delay;
+        }
+
+        @Override
+        public void simulateLatency() {
+            long start = System.nanoTime();
+            while(System.nanoTime() - start < delay);
+        }
+    }
 }

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

@@ -62,6 +62,8 @@ public class MulticastParams {
 
     private Map<String, Object> queueArguments;
 
+    private int consumerLatencyInMicroseconds;
+
     public void setExchangeType(String exchangeType) {
         this.exchangeType = exchangeType;
     }
@@ -175,6 +177,10 @@ public void setQueueArguments(Map<String, Object> queueArguments) {
         this.queueArguments = queueArguments;
     }
 
+    public void setConsumerLatencyInMicroseconds(int consumerLatencyInMicroseconds) {
+        this.consumerLatencyInMicroseconds = consumerLatencyInMicroseconds;
+    }
+
     public int getConsumerCount() {
         return consumerCount;
     }
@@ -254,7 +260,7 @@ public Consumer createConsumer(Connection connection, Stats stats, String id) th
         if (channelPrefetch > 0) channel.basicQos(channelPrefetch, true);
         return new Consumer(channel, id, generatedQueueNames,
                                          consumerTxSize, autoAck, multiAckEvery,
-                                         stats, consumerRateLimit, consumerMsgCount, timeLimit);
+                                         stats, consumerRateLimit, consumerMsgCount, timeLimit, consumerLatencyInMicroseconds);
     }
 
     public boolean shouldConfigureQueues() {

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

@@ -88,6 +88,7 @@ public static void main(String[] args) {
             boolean predeclared      = cmd.hasOption('p');
             boolean autoDelete       = boolArg(cmd, "ad", true);
             String queueArgs         = strArg(cmd, "qa", null);
+            int consumerLatencyInMicroseconds = intArg(cmd, 'L', 0);
 
             String uri               = strArg(cmd, 'h', "amqp://localhost");
             String urisParameter     = strArg(cmd, 'H', null);
@@ -151,6 +152,7 @@ public static void main(String[] args) {
             p.setBodyFiles(             bodyFiles == null ? null : asList(bodyFiles.split(",")));
             p.setBodyContentType(       bodyContentType);
             p.setQueueArguments(queueArguments(queueArgs));
+            p.setConsumerLatencyInMicroseconds(consumerLatencyInMicroseconds);
 
             MulticastSet set = new MulticastSet(stats, factory, p, testID, uris);
             set.run(true);
@@ -231,6 +233,7 @@ private static Options getOptions() {
         options.addOption(new Option("T", "bodyContenType",         true, "body content-type"));
         options.addOption(new Option("ad", "autoDelete", true, "should the queue be auto-deleted, default is true"));
         options.addOption(new Option("qa", "queueArgs", true, "queue arguments as key/pair values, separated by commas"));
+        options.addOption(new Option("L", "consumerLatency", true, "consumer latency in microseconds"));
         options.addOption(new Option("useDefaultSslContext", "useDefaultSslContext",       false,"use JVM default SSL context"));
         return options;
     }