yezijiang
diff --git a/‎src/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java
Lines changed: 64 additions & 16 deletions b/‎src/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java
Lines changed: 64 additions & 16 deletions
@@ -67,7 +67,14 @@ public class AutorecoveringConnection implements Connection, Recoverable, Networ
     private final Map<String, RecordedConsumer> consumers = new ConcurrentHashMap<String, RecordedConsumer>();
     private final List<ConsumerRecoveryListener> consumerRecoveryListeners = new ArrayList<ConsumerRecoveryListener>();
     private final List<QueueRecoveryListener> queueRecoveryListeners = new ArrayList<QueueRecoveryListener>();
-
+	
+	// Used to block connection recovery attempts after close() is invoked.
+	private volatile boolean manuallyClosed = false;
+	
+	// This lock guards the manuallyClosed flag and the delegate connection.  Guarding these two ensures that a new connection can never
+	// be created after application code has initiated shutdown.  
+	private Object recoveryLock = new Object();
+	
     public AutorecoveringConnection(ConnectionParams params, FrameHandlerFactory f, Address[] addrs) {
         this.cf = new RecoveryAwareAMQConnectionFactory(params, f, addrs);
         this.params = params;
@@ -181,48 +188,69 @@ public boolean isOpen() {
      * @see com.rabbitmq.client.Connection#close()
      */
     public void close() throws IOException {
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
         delegate.close();
     }
 
     /**
      * @see Connection#close(int)
      */
     public void close(int timeout) throws IOException {
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
         delegate.close(timeout);
     }
 
     /**
      * @see Connection#close(int, String, int)
      */
     public void close(int closeCode, String closeMessage, int timeout) throws IOException {
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
         delegate.close(closeCode, closeMessage, timeout);
     }
 
     /**
      * @see com.rabbitmq.client.Connection#abort()
      */
     public void abort() {
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
         delegate.abort();
     }
 
     /**
      * @see Connection#abort(int, String, int)
      */
     public void abort(int closeCode, String closeMessage, int timeout) {
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
         delegate.abort(closeCode, closeMessage, timeout);
     }
     /**
      * @see Connection#abort(int, String)
      */
     public void abort(int closeCode, String closeMessage) {
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
         delegate.abort(closeCode, closeMessage);
     }
 
     /**
      * @see Connection#abort(int)
      */
     public void abort(int timeout) {
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
         delegate.abort(timeout);
     }
 
@@ -261,7 +289,10 @@ public void clearBlockedListeners() {
      * @see com.rabbitmq.client.Connection#close(int, String)
      */
     public void close(int closeCode, String closeMessage) throws IOException {
-        delegate.close(closeCode, closeMessage);
+		synchronized(recoveryLock) {
+			this.manuallyClosed = true;
+		}
+		delegate.close(closeCode, closeMessage);
     }
 
     /**
@@ -404,16 +435,18 @@ public void removeConsumerRecoveryListener(ConsumerRecoveryListener listener) {
 
     synchronized private void beginAutomaticRecovery() throws InterruptedException, IOException, TopologyRecoveryException {
         Thread.sleep(this.params.getNetworkRecoveryInterval());
-        this.recoverConnection();
-        this.recoverShutdownListeners();
-        this.recoverBlockedListeners();
-        this.recoverChannels();
-        if(this.params.isTopologyRecoveryEnabled()) {
-            this.recoverEntities();
-            this.recoverConsumers();
-        }
+        if (!this.recoverConnection())
+			return; 
+		
+		this.recoverShutdownListeners();
+		this.recoverBlockedListeners();
+		this.recoverChannels();
+		if(this.params.isTopologyRecoveryEnabled()) {
+			this.recoverEntities();
+			this.recoverConsumers();
+		}
 
-        this.notifyRecoveryListeners();
+		this.notifyRecoveryListeners();
     }
 
     private void recoverShutdownListeners() {
@@ -428,18 +461,33 @@ private void recoverBlockedListeners() {
         }
     }
 
-    private void recoverConnection() throws IOException, InterruptedException {
-        boolean recovering = true;
-        while (recovering) {
+	// Returns true if the connection was recovered, 
+	// false if application initiated shutdown while attempting recovery.  
+    private boolean recoverConnection() throws IOException, InterruptedException {
+        while (!manuallyClosed)
+		{
             try {
-                this.delegate = this.cf.newConnection();
-                recovering = false;
+				RecoveryAwareAMQConnection newConn = this.cf.newConnection();
+				synchronized(recoveryLock) {
+					if (!manuallyClosed) {
+						// This is the standard case.
+						this.delegate = newConn;					
+						return true;
+					}
+				}
+				// This is the once in a blue moon case.  
+				// Application code just called close as the connection
+				// was being re-established.  So we attempt to close the newly created connection.
+				newConn.abort();
+				return false;
             } catch (Exception e) {
                 // TODO: exponential back-off
                 Thread.sleep(this.params.getNetworkRecoveryInterval());
                 this.getExceptionHandler().handleConnectionRecoveryException(this, e);
             }
         }
+		
+		return false;
     }
 
     private void recoverChannels() {