Schedule recovery probing

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -63,7 +63,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final List<RpcServer> rpcServers = new CopyOnWriteArrayList<>();
   private final TopologyListener topologyListener;
   private volatile EntityRecovery entityRecovery;
-  private final Future<?> recoveryLoop;
+  private final Future<?> recoveryTask;
   private final BlockingQueue<Runnable> recoveryRequestQueue;
   private final AtomicBoolean recoveringConnection = new AtomicBoolean(false);
   private final DefaultConnectionSettings<?> connectionSettings;
@@ -90,23 +90,52 @@ final class AmqpConnection extends ResourceBase implements Connection {
 
     if (recoveryConfiguration.activated()) {
       this.recoveryRequestQueue = new ArrayBlockingQueue<>(10);
-      this.recoveryLoop =
-          this.executorService()
-              .submit(
+      Duration recoveryCheckPeriod = Duration.ofMillis(100L);
+      this.recoveryTask =
+          this.scheduledExecutorService()
+              .scheduleAtFixedRate(
                   () -> {
-                    while (!Thread.currentThread().isInterrupted()) {
-                      try {
-                        Runnable recoveryTask = this.recoveryRequestQueue.take();
-                        LOGGER.debug("Running recovery task for connection '{}'...", this.name());
-                        recoveryTask.run();
-                        LOGGER.debug("Recovery task for connection '{}' has run.", this.name());
-                      } catch (InterruptedException e) {
-                        return;
-                      } catch (Exception e) {
-                        LOGGER.warn("Error during connection recovery for '{}'", this.name(), e);
-                      }
+                    Runnable recoveryTask = this.recoveryRequestQueue.poll();
+                    if (recoveryTask != null) {
+                      this.executorService()
+                          .submit(
+                              () -> {
+                                try {
+                                  LOGGER.debug(
+                                      "Running recovery task for connection '{}'...", this.name());
+                                  recoveryTask.run();
+                                  LOGGER.debug(
+                                      "Recovery task for connection '{}' has run.", this.name());
+                                } catch (Exception e) {
+                                  LOGGER.warn(
+                                      "Error during connection recovery for '{}'", this.name(), e);
+                                }
+                              });
                     }
-                  });
+                  },
+                  recoveryCheckPeriod.toMillis(),
+                  recoveryCheckPeriod.toMillis(),
+                  TimeUnit.MILLISECONDS);
+      //      this.recoveryLoop =
+      //          this.executorService()
+      //              .submit(
+      //                  () -> {
+      //                    while (!Thread.currentThread().isInterrupted()) {
+      //                      try {
+      //                        Runnable recoveryTask = this.recoveryRequestQueue.take();
+      //                        LOGGER.debug("Running recovery task for connection '{}'...",
+      // this.name());
+      //                        recoveryTask.run();
+      //                        LOGGER.debug("Recovery task for connection '{}' has run.",
+      // this.name());
+      //                      } catch (InterruptedException e) {
+      //                        return;
+      //                      } catch (Exception e) {
+      //                        LOGGER.warn("Error during connection recovery for '{}'",
+      // this.name(), e);
+      //                      }
+      //                    }
+      //                  });
       disconnectHandler = recoveryDisconnectHandler(recoveryConfiguration, builder.name());
     } else {
       disconnectHandler =
@@ -116,7 +145,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
             this.close(failureCause);
           };
       this.recoveryRequestQueue = null;
-      this.recoveryLoop = null;
+      this.recoveryTask = null;
     }
     if (this.connectionSettings.affinity().activated()) {
       this.affinity =
@@ -668,8 +697,8 @@ private void close(Throwable cause) {
     if (this.closed.compareAndSet(false, true)) {
       this.state(CLOSING, cause);
       this.environment.removeConnection(this);
-      if (this.recoveryLoop != null) {
-        this.recoveryLoop.cancel(true);
+      if (this.recoveryTask != null) {
+        this.recoveryTask.cancel(true);
       }
       if (this.topologyListener instanceof AutoCloseable) {
         try {