Add --time option to performance tool

Fixes #198

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -394,6 +394,13 @@ public void setMaxSegmentSize(ByteCapacity in) {
       description = "AMQP URI to use to create super stream topology")
   private String amqpUri;
 
+  @CommandLine.Option(
+      names = {"--time", "-z"},
+      description = "run duration in seconds, unlimited by default",
+      defaultValue = "0",
+      converter = Utils.GreaterThanOrEqualToZeroIntegerTypeConverter.class)
+  private int time;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
   private List<Monitoring> monitorings;
@@ -940,7 +947,11 @@ public Integer call() throws Exception {
       Thread shutdownHook = new Thread(() -> latch.countDown());
       Runtime.getRuntime().addShutdownHook(shutdownHook);
       try {
-        latch.await();
+        if (this.time > 0) {
+          latch.await(this.time, TimeUnit.SECONDS);
+        } else {
+          latch.await();
+        }
         Runtime.getRuntime().removeShutdownHook(shutdownHook);
       } catch (InterruptedException e) {
         // moving on to the closing sequence

src/main/java/com/rabbitmq/stream/perf/Utils.java

@@ -487,6 +487,23 @@ public Integer convert(String input) {
     }
   }
 
+  static class GreaterThanOrEqualToZeroIntegerTypeConverter
+      implements CommandLine.ITypeConverter<Integer> {
+
+    @Override
+    public Integer convert(String input) {
+      try {
+        Integer value = Integer.valueOf(input);
+        if (value < 0) {
+          throw new IllegalArgumentException();
+        }
+        return value;
+      } catch (Exception e) {
+        throw new CommandLine.TypeConversionException(input + " is not greater than or equal to 0");
+      }
+    }
+  }
+
   static class CompressionTypeConverter implements CommandLine.ITypeConverter<Compression> {
 
     @Override

src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java

@@ -37,6 +37,7 @@
 import java.net.HttpURLConnection;
 import java.net.ServerSocket;
 import java.net.URL;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
@@ -409,6 +410,17 @@ void confirmLatencyShouldBeIncluded() throws Exception {
         .doesNotContain("confirm latency");
   }
 
+  @Test
+  void shouldStopWhenTimeIsSet() throws Exception {
+    int time = 3;
+    long start = System.nanoTime();
+    run(builder().time(time));
+    waitUntilStreamExists(s);
+    waitRunEnds();
+    assertThat(Duration.ofNanos(System.nanoTime() - start))
+        .isGreaterThanOrEqualTo(Duration.ofSeconds(3));
+  }
+
   void singleActiveConsumersOnSuperStream() throws Exception {
     String consumerName = "app-1";
     Future<?> run =
@@ -639,6 +651,11 @@ ArgumentsBuilder singleActiveConsumer() {
       return this;
     }
 
+    ArgumentsBuilder time(int time) {
+      arguments.put("time", String.valueOf(time));
+      return this;
+    }
+
     String build() {
       return this.arguments.entrySet().stream()
           .map(e -> "--" + e.getKey() + (e.getValue().isEmpty() ? "" : (" " + e.getValue())))