Add internal listener to collect consumer queues in tests

This is more reliable than listing queues before and while the test is
running, as unrelated queues can go away between calls.

Author: acogoluegnes

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

@@ -140,6 +140,7 @@ public class MulticastParams {
 
   private PrintStream out = System.out;
   private boolean netty = false;
+  private java.util.function.Consumer<List<String>> consumerConfiguredQueueListener = qs -> {};
 
   public void setExchangeType(String exchangeType) {
     this.exchangeType = exchangeType;
@@ -639,6 +640,8 @@ public Consumer createConsumer(
     Recovery.RecoveryProcess recoveryProcess =
         setupRecoveryProcess(connection, topologyHandlerResult.topologyRecording);
 
+    this.consumerConfiguredQueueListener.accept(topologyHandlerResult.configuredQueues);
+
     Consumer consumer =
         new Consumer(
             new ConsumerParameters()
@@ -782,6 +785,11 @@ boolean netty() {
     return this.netty;
   }
 
+  public void setConsumerConfiguredQueueListener(
+      java.util.function.Consumer<List<String>> listener) {
+    this.consumerConfiguredQueueListener = listener;
+  }
+
   /**
    * Contract to handle the creation and configuration of resources. E.g. creation of queues,
    * binding exchange to queues.

src/test/java/com/rabbitmq/perf/it/ConnectionRecoveryIT.java

@@ -18,9 +18,11 @@
 import static com.rabbitmq.perf.TestUtils.randomName;
 import static com.rabbitmq.perf.TestUtils.threadFactory;
 import static com.rabbitmq.perf.TestUtils.waitAtMost;
+import static com.rabbitmq.perf.it.Host.listQueues;
 import static com.rabbitmq.perf.it.Utils.latchCompletionHandler;
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.IntStream.range;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -38,6 +40,8 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -66,11 +70,12 @@ public class ConnectionRecoveryIT {
 
   static final String URI = "amqp://localhost";
 
-  static final List<String> URIS = Collections.singletonList(URI);
+  static final List<String> URIS = singletonList(URI);
 
   static final int RATE = 11;
 
   MulticastParams params;
+  Queue<String> consumerQueues = new ConcurrentLinkedQueue<>();
 
   ExecutorService executorService;
 
@@ -219,6 +224,7 @@ public void init(TestInfo info) {
     params.setProducerCount(1);
     params.setConsumerCount(1);
     params.setProducerRateLimit(RATE);
+    params.setConsumerConfiguredQueueListener(consumerQueues::addAll);
     cf = new ConnectionFactory();
     cf.setNetworkRecoveryInterval(2000);
     cf.setTopologyRecoveryEnabled(false);
@@ -410,12 +416,11 @@ public void shouldRecoverWithPreDeclared(boolean polling, TestInfo info) throws
 
   @Test
   void durableServerNamedQueueShouldBeReusedIfStillThere(TestInfo info) throws Exception {
-    params.setFlags(Arrays.asList("persistent"));
+    params.setFlags(singletonList("persistent"));
     params.setAutoDelete(false);
     params.setExclusive(false);
     params.setQueueNames(Collections.emptyList());
 
-    List<String> queuesBeforeTest = Host.listServerNamedQueues();
     MulticastSet.CompletionHandler completionHandler = latchCompletionHandler(1, info);
     String createdQueue = null;
     try {
@@ -424,15 +429,12 @@ void durableServerNamedQueueShouldBeReusedIfStillThere(TestInfo info) throws Exc
           new MulticastSet(performanceMetrics, cf, params, "", URIS, completionHandler);
       run(set);
       waitAtMost(10, () -> msgConsumed.get() >= 3 * producerConsumerCount * RATE);
-      List<String> queuesDuringTest = Host.listServerNamedQueues();
-      assertThat(queuesDuringTest).hasSize(queuesBeforeTest.size() + 1);
-      queuesDuringTest.removeAll(queuesBeforeTest);
-      assertThat(queuesDuringTest).hasSize(1);
-      createdQueue = queuesDuringTest.get(0);
+      assertThat(consumerQueues).hasSize(1);
+      createdQueue = consumerQueues.poll();
       long messageCountBeforeClosing = msgConsumed.get();
       closeAllConnections();
       waitAtMost(10, () -> msgConsumed.get() >= 2 * messageCountBeforeClosing);
-      assertThat(Host.listServerNamedQueues()).hasSize(queuesBeforeTest.size() + 1);
+      assertThat(listQueues()).contains(createdQueue);
       completionHandler.countDown("stopped in test");
       waitAtMost(10, () -> testIsDone.get());
     } finally {
@@ -454,7 +456,7 @@ void recreateBindingEvenOnPreDeclaredDurableQueue(TestInfo info) throws Exceptio
       ch.exchangeDelete("direct");
     }
 
-    params.setQueueNames(Collections.singletonList(queue));
+    params.setQueueNames(singletonList(queue));
     params.setPredeclared(true);
 
     int producerConsumerCount = params.getProducerCount();

src/test/java/com/rabbitmq/perf/it/Host.java

@@ -145,10 +145,6 @@ public static List<ConnectionInfo> listConnections() throws IOException {
     return result;
   }
 
-  public static List<String> listServerNamedQueues() throws IOException {
-    return listQueues().stream().filter(q -> q.startsWith("amq.gen")).collect(Collectors.toList());
-  }
-
   public static List<String> listQueues() throws IOException {
     // "messages" column will help ignore the header line
     String output = capture(rabbitmqctl("list_queues -q name messages").getInputStream());

src/test/java/com/rabbitmq/perf/it/MiscellaneousIT.java

@@ -35,6 +35,8 @@
 import java.io.PrintStream;
 import java.time.Duration;
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -148,23 +150,21 @@ void consumerShouldAckWhenRateLimitationIsEnabled(TestInfo info) throws Exceptio
 
   @Test
   void consumerShouldRecoverWhenServerNamedQueueIsDeleted(TestInfo info) throws Exception {
+    Queue<String> queues = new ConcurrentLinkedQueue<>();
     int rate = 100;
     params.setProducerCount(1);
     params.setConsumerCount(1);
     params.setProducerRateLimit(rate);
-
-    List<String> queuesBeforeTest = Host.listQueues();
+    params.setConsumerConfiguredQueueListener(queues::addAll);
 
     MulticastSet.CompletionHandler completionHandler = latchCompletionHandler(1, info);
     MulticastSet set =
         new MulticastSet(performanceMetrics, cf, params, "", URIS, completionHandler);
     run(set);
 
     waitAtMost(10, () -> msgConsumed.get() >= 3 * rate);
-    List<String> queuesDuringTest = Host.listQueues();
-    queuesDuringTest.removeAll(queuesBeforeTest);
-    assertThat(queuesDuringTest).hasSize(1);
-    String queue = queuesDuringTest.get(0);
+    assertThat(queues).hasSize(1);
+    String queue = queues.poll();
 
     ConnectionFactory connectionFactory = new ConnectionFactory();
     long messageConsumedBeforeDeletion;