Introduce a way for AMQConnection to notify its recovery aware wrapper

Fixes rabbitmq#135.

Author: michaelklishin

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -21,10 +21,7 @@
 import java.net.InetAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.*;
 
 import com.rabbitmq.client.AMQP;
@@ -45,6 +42,7 @@
 import com.rabbitmq.client.SaslMechanism;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.impl.AMQChannel.BlockingRpcContinuation;
+import com.rabbitmq.client.impl.recovery.RecoveryCanBeginListener;
 import com.rabbitmq.utility.BlockingCell;
 
 final class Copyright {
@@ -65,6 +63,9 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
  private Thread mainLoopThread;
  private ThreadFactory threadFactory = Executors.defaultThreadFactory();
 
+ private final List<RecoveryCanBeginListener> recoveryCanBeginListeners =
+ new ArrayList<RecoveryCanBeginListener>();
+
  /**
  * Retrieve a copy of the default table of client properties that
  * will be sent to the server during connection startup. This
@@ -576,10 +577,31 @@ public void run() {
  _frameHandler.close();
  _appContinuation.set(null);
  notifyListeners();
+ // assuming that shutdown listeners do not do anything
+ // asynchronously, e.g. start new threads, this effectively
+ // guarantees that we only begin recovery when all shutdown
+ // listeners have executed
+ notifyRecoveryCanBeginListeners();
  }
  }
  }
 
+ private void notifyRecoveryCanBeginListeners() {
+ ShutdownSignalException sse = this.getCloseReason();
+ for(RecoveryCanBeginListener fn : this.recoveryCanBeginListeners) {
+ fn.recoveryCanBegin(sse);
+ }
+ }
+
+ public void addRecoveryCanBeginListener(RecoveryCanBeginListener fn) {
+ this.recoveryCanBeginListeners.add(fn);
+ }
+
+ @SuppressWarnings(value = "unused")
+ public void removeRecoveryCanBeginListener(RecoveryCanBeginListener fn) {
+ this.recoveryCanBeginListeners.remove(fn);
+ }
+
  /**
  * Called when a frame-read operation times out
  * @throws MissedHeartbeatException if heart-beats have been missed

src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java

@@ -376,8 +376,10 @@ public int getLocalPort() {
 
  private void addAutomaticRecoveryListener() {
  final AutorecoveringConnection c = this;
- ShutdownListener automaticRecoveryListener = new ShutdownListener() {
- public void shutdownCompleted(ShutdownSignalException cause) {
+ // this listener will run after shutdown listeners,
+ // see https://github.com/rabbitmq/rabbitmq-java-client/issues/135
+ RecoveryCanBeginListener starter = new RecoveryCanBeginListener() {
+ public void recoveryCanBegin(ShutdownSignalException cause) {
  try {
  if (shouldTriggerConnectionRecovery(cause)) {
  c.beginAutomaticRecovery();
@@ -388,10 +390,7 @@ public void shutdownCompleted(ShutdownSignalException cause) {
  }
  };
  synchronized (this) {
- if(!this.shutdownHooks.contains(automaticRecoveryListener)) {
- this.shutdownHooks.add(automaticRecoveryListener);
- }
- this.delegate.addShutdownListener(automaticRecoveryListener);
+ this.delegate.addRecoveryCanBeginListener(starter);
  }
  }
 
@@ -441,18 +440,20 @@ public void removeConsumerRecoveryListener(ConsumerRecoveryListener listener) {
 
  synchronized private void beginAutomaticRecovery() throws InterruptedException, IOException, TopologyRecoveryException {
  Thread.sleep(this.params.getNetworkRecoveryInterval());
- if (!this.recoverConnection())
-return; 
-
-this.recoverShutdownListeners();
-this.recoverBlockedListeners();
-this.recoverChannels();
-if(this.params.isTopologyRecoveryEnabled()) {
-this.recoverEntities();
-this.recoverConsumers();
-}
+ if (!this.recoverConnection()) {
+ return;
+ }
 
-this.notifyRecoveryListeners();
+ this.addAutomaticRecoveryListener();
+ this.recoverShutdownListeners();
+ this.recoverBlockedListeners();
+ this.recoverChannels();
+ if(this.params.isTopologyRecoveryEnabled()) {
+ this.recoverEntities();
+ this.recoverConsumers();
+ }
+
+ this.notifyRecoveryListeners();
  }
 
  private void recoverShutdownListeners() {

src/main/java/com/rabbitmq/client/impl/recovery/RecoveryCanBeginListener.java

@@ -0,0 +1,12 @@
+package com.rabbitmq.client.impl.recovery;
+
+import com.rabbitmq.client.ShutdownSignalException;
+
+/**
+ * Used internally to indicate when connection recovery can
+ * begin. See {@link https://github.com/rabbitmq/rabbitmq-java-client/issues/135}.
+ * This is package-local by design.
+ */
+public interface RecoveryCanBeginListener {
+ void recoveryCanBegin(ShutdownSignalException cause);
+}