Stops Recovery When Client Closes

Adds a boolean flag, manuallyClosed to keep track of whether or not the
AutoRecoveringConnection has been told to close by the application.

If flag is set before recovery attempt, autorecovery fails out.
If flag is set mid recovery attempt, creates the new
connection, then immediately aborts it.

Author: dhakimTRX

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

@@ -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() {