Revert "Start metrics service before consumers and publishers"

This reverts commit feb39e2.

Author: acogoluegnes

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

@@ -236,8 +236,6 @@ public void run(boolean announceStartup)
 
             createProducers(announceStartup, producerStates, producerConnections);
 
-            this.performanceMetrics.start();
-
             startConsumers(consumerRunnables);
             startProducers(producerStates);
 
@@ -302,6 +300,7 @@ public void run(boolean announceStartup)
                 shutdownSequence = () -> { };
             }
 
+            this.performanceMetrics.start();
             this.completionHandler.waitForCompletion();
 
             try {
@@ -433,9 +432,7 @@ private void startConsumers(Runnable[] consumerRunnables) throws InterruptedExce
         if (this.params.getConsumerStartDelay().getSeconds() <= 0) {
             this.consumerLatencyIndicator.start();
             for (Runnable runnable : consumerRunnables) {
-                LOGGER.debug("Starting consumer runnable...");
                 runnable.run();
-                LOGGER.debug("Consumer runnable started");
                 if (params.getConsumerSlowStart()) {
                     System.out.println("Delaying start by 1 second because -S/--slow-start was requested");
                     Thread.sleep(1000);

src/main/java/com/rabbitmq/perf/metrics/DefaultPerformanceMetrics.java

@@ -73,7 +73,7 @@ public final class DefaultPerformanceMetrics implements PerformanceMetrics, Auto
   private final Duration interval;
   private final TimeUnit latencyCollectionTimeUnit;
   private final AtomicBoolean firstReport = new AtomicBoolean(false);
-  private final AtomicBoolean started = new AtomicBoolean(false);
+  private final AtomicBoolean closed = new AtomicBoolean(false);
   private final AtomicReference<Histogram> consumedLatency, confirmedLatency;
   private final MetricsFormatter formatter;
 
@@ -162,29 +162,27 @@ private static Runnable wrapInCatch(Runnable runnable) {
 
   @Override
   public void start() {
-    if (this.started.compareAndSet(false, true)) {
-      startTime.set(System.nanoTime());
-      lastTick.set(startTime.get());
-      startTimeForTotal.set(startTime.get());
-
-      scheduledExecutorService.scheduleAtFixedRate(wrapInCatch(() -> {
-        if (!this.started.get()) {
-          return;
-        }
-        if (noActivity()) {
-          this.publishedRate.accumulate(0);
-          this.confirmedRate.accumulate(0);
-          this.nackedRate.accumulate(0);
-          this.returnedRate.accumulate(0);
-          this.receivedRate.accumulate(0);
-          this.confirmedLatency.set(histogram());
-          this.consumedLatency.set(histogram());
-        } else {
-          metrics(System.nanoTime());
-        }
-
-      }), interval.getSeconds(), interval.getSeconds(), TimeUnit.SECONDS);
-    }
+    startTime.set(System.nanoTime());
+    lastTick.set(startTime.get());
+    startTimeForTotal.set(startTime.get());
+
+    scheduledExecutorService.scheduleAtFixedRate(wrapInCatch(() -> {
+      if (this.closed.get()) {
+        return;
+      }
+      if (noActivity()) {
+        this.publishedRate.accumulate(0);
+        this.confirmedRate.accumulate(0);
+        this.nackedRate.accumulate(0);
+        this.returnedRate.accumulate(0);
+        this.receivedRate.accumulate(0);
+        this.confirmedLatency.set(histogram());
+        this.consumedLatency.set(histogram());
+      } else {
+        metrics(System.nanoTime());
+      }
+
+    }), interval.getSeconds(), interval.getSeconds(), TimeUnit.SECONDS);
   }
 
   void metrics(long currentTime) {
@@ -213,7 +211,7 @@ void metrics(long currentTime) {
     this.confirmedLatency.set(histogram());
     this.consumedLatency.set(histogram());
 
-    if (this.started.get()) {
+    if (!this.closed.get()) {
       if (this.firstReport.compareAndSet(false, true)) {
         this.formatter.header();
       }
@@ -322,15 +320,9 @@ public void resetGlobals() {
 
   @Override
   public void close() {
-    if (this.started.compareAndSet(false, true)) {
+    if (this.closed.compareAndSet(false, true)) {
       this.scheduledExecutorService.shutdownNow();
       printFinal();
     }
   }
-
-  // for testing
-  void started(boolean value) {
-    this.started.set(value);
-  }
-
 }

src/test/java/com/rabbitmq/perf/metrics/DefaultPerformanceMetricsTest.java

@@ -166,7 +166,6 @@ public void stats(TestConfiguration testConfiguration) {
 
   void execute(Configurator configurator) {
     DefaultPerformanceMetrics metrics = metrics(configurator);
-    metrics.started(true);
     metrics.metrics(System.nanoTime());
     this.defaultOutput = defaultConsoleOut.toString();
     this.compactOutput = compactConsoleOut.toString();