Make publishing in performance tool able to recover

Author: acogoluegnes

src/test/java/com/rabbitmq/client/amqp/perf/AmqpPerfTest.java

@@ -35,9 +35,13 @@
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class AmqpPerfTest {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(AmqpPerfTest.class);
+
   /*
   ./mvnw -q clean test-compile exec:java \
     -Dexec.mainClass=com.rabbitmq.client.amqp.perf.AmqpPerfTest \
@@ -86,6 +90,12 @@ public static void main(String[] args) throws IOException {
 
       connection
           .consumerBuilder()
+          .listeners(
+              context -> {
+                if (context.currentState() == Resource.State.RECOVERING) {
+                  LOGGER.info("Consumer is recovering...");
+                }
+              })
           .queue(q)
           .initialCredits(1000)
           .messageHandler(
@@ -103,7 +113,24 @@ public static void main(String[] args) throws IOException {
 
       executorService.submit(
           () -> {
-            Publisher publisher = connection.publisherBuilder().exchange(e).key(rk).build();
+            AtomicBoolean shouldPublish = new AtomicBoolean(false);
+            Publisher publisher =
+                connection
+                    .publisherBuilder()
+                    .exchange(e)
+                    .key(rk)
+                    .listeners(
+                        context -> {
+                          if (context.currentState() == Resource.State.OPEN) {
+                            shouldPublish.set(true);
+                          } else {
+                            if (context.currentState() == Resource.State.RECOVERING) {
+                              LOGGER.info("Publisher is recovering...");
+                            }
+                            shouldPublish.set(false);
+                          }
+                        })
+                    .build();
             Publisher.Callback callback =
                 context -> {
                   try {
@@ -116,11 +143,23 @@ public static void main(String[] args) throws IOException {
                 };
             int msgSize = 10;
             while (!Thread.currentThread().isInterrupted()) {
-              long creationTime = System.currentTimeMillis();
-              byte[] payload = new byte[msgSize];
-              writeLong(payload, creationTime);
-              Message message = publisher.message(payload);
-              publisher.publish(message, callback);
+              if (shouldPublish.get()) {
+                long creationTime = System.currentTimeMillis();
+                byte[] payload = new byte[msgSize];
+                writeLong(payload, creationTime);
+                try {
+                  Message message = publisher.message(payload);
+                  publisher.publish(message, callback);
+                } catch (Exception ex) {
+                  LOGGER.info("Error while trying to publish: {}", ex.getMessage());
+                }
+              } else {
+                try {
+                  Thread.sleep(1000L);
+                } catch (InterruptedException ex) {
+                  Thread.interrupted();
+                }
+              }
             }
           });
       out.println("Prometheus endpoint started on http://localhost:" + monitoringPort + "/metrics");

src/test/resources/logback-test.xml

@@ -12,6 +12,7 @@
     <logger name="com.rabbitmq.client.amqp.impl.AmqpConsumer" level="warn" />
     <logger name="com.rabbitmq.client.amqp.impl.AmqpManagement" level="warn" />
     <logger name="org.apache.qpid" level="warn" />
+    <logger name="com.rabbitmq.client.amqp.perf" level="info" />
 
     <root level="warn">
         <appender-ref ref="STDOUT" />