Polish consumer variable latency support

Author: acogoluegnes

src/docs/asciidoc/usage-advanced.adoc

@@ -124,13 +124,15 @@ then back to 1 kB for 30 seconds, and so on.
 
 You can simulate processing time per message with either a fixed or a variable latency value in microseconds.
 
-The `--consumer-latency' ('-L') option sets a fixed consumer latency in microseconds. In the example below a 1ms latency is set.
+The `--consumer-latency` (`-L`) option sets a fixed consumer latency in microseconds. In the example
+below a 1 ms latency is set:
 
   bin/runjava com.rabbitmq.perf.PerfTest --consumer-latency 1000
 
-The `--variable-latency' ('-vl') option sets a variable consumer latency. In the example below it is set to 60 seconds of 1ms followed by 60 seconds of 1 second).
+The `--variable-latency` (`-vl`) option sets a variable consumer latency. In the example below it is
+set to 1 ms for 60 seconds then 1 second for 30 seconds:
 
-  bin/runjava com.rabbitmq.perf.PerfTest --variable-latency 1000:60 --variable-latency 1000000:60
+  bin/runjava com.rabbitmq.perf.PerfTest --variable-latency 1000:60 --variable-latency 1000000:30
 
 == Working With Many Queues
 

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

@@ -391,7 +391,25 @@ private interface ConsumerLatency {
 
     }
 
-    private static class VariableConsumerLatency implements  ConsumerLatency {
+    private static boolean latencySleep(long delay) {
+        try {
+            long ms = delay / 1000;
+            Thread.sleep(ms);
+            return true;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return false;
+        }
+    }
+
+    private static boolean latencyBusyWait(long delay) {
+        delay = delay * 1000;
+        long start = System.nanoTime();
+        while (System.nanoTime() - start < delay) ;
+        return true;
+    }
+
+    private static class VariableConsumerLatency implements ConsumerLatency {
 
         private final ValueIndicator<Long> consumerLatenciesIndicator;
 
@@ -400,35 +418,17 @@ private VariableConsumerLatency(ValueIndicator<Long> consumerLatenciesIndicator)
         }
 
         @Override
-        public boolean simulateLatency()
-        {
+        public boolean simulateLatency() {
             long consumerLatencyInMicroSeconds = consumerLatenciesIndicator.getValue();
             if (consumerLatencyInMicroSeconds <= 0) {
                 return true;
             } else if (consumerLatencyInMicroSeconds >= 1000) {
-                return sleep();
+                return latencySleep(consumerLatencyInMicroSeconds);
             } else {
-                return busyWait();
+                return latencyBusyWait(consumerLatencyInMicroSeconds);
             }
         }
 
-        private boolean sleep() {
-            try {
-                long ms = consumerLatenciesIndicator.getValue() / 1000;
-                Thread.sleep(ms);
-                return true;
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                return false;
-            }
-        }
-
-        private boolean busyWait() {
-            long delay = consumerLatenciesIndicator.getValue() * 1000;
-            long start = System.nanoTime();
-            while(System.nanoTime() - start < delay);
-            return true;
-        }
     }
 
     private static class NoWaitConsumerLatency implements ConsumerLatency {
@@ -450,14 +450,7 @@ private ThreadSleepConsumerLatency(ValueIndicator<Long> consumerLatenciesIndicat
 
         @Override
         public boolean simulateLatency() {
-            try {
-                long ms = consumerLatenciesIndicator.getValue() / 1000;
-                Thread.sleep(ms);
-                return true;
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                return false;
-            }
+            return latencySleep(consumerLatenciesIndicator.getValue());
         }
     }
 
@@ -472,10 +465,7 @@ private BusyWaitConsumerLatency(ValueIndicator<Long> consumerLatenciesIndicator)
 
         @Override
         public boolean simulateLatency() {
-            long delay = consumerLatenciesIndicator.getValue() * 1000;
-            long start = System.nanoTime();
-            while(System.nanoTime() - start < delay);
-            return true;
+            return latencyBusyWait(consumerLatenciesIndicator.getValue());
         }
     }
 

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

@@ -332,7 +332,7 @@ private void createConsumers(boolean announceStartup,
         }
     }
 
-    private void createProducers(boolean announceStartup, AgentState[] producerStates, Connection[] producerConnections) throws URISyntaxException, NoSuchAlgorithmException, KeyManagementException, IOException, TimeoutException {
+    private void createProducers(boolean announceStartup, AgentState[] producerStates, Connection[] producerConnections) throws IOException, TimeoutException {
         for (int i = 0; i < producerConnections.length; i++) {
             if (announceStartup) {
                 System.out.println("id: " + testID + ", starting producer #" + i);