Make AMQConnection#doFinalShutdown idempotent

Now recovery can be triggered from write operations,
late connection failure discoveries can re-trigger the shutdown
and emit spurious exception.

References #341

Author: acogoluegnes

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

@@ -32,6 +32,7 @@
 import java.net.SocketTimeoutException;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 final class Copyright {
     final static String COPYRIGHT="Copyright (c) 2007-2017 Pivotal Software, Inc.";
@@ -63,6 +64,8 @@ public class AMQConnection extends ShutdownNotifierComponent implements Connecti
 
     private final ErrorOnWriteListener errorOnWriteListener;
 
+    private final AtomicBoolean finalShutdownStarted = new AtomicBoolean(false);
+
     /**
      * Retrieve a copy of the default table of client properties that
      * will be sent to the server during connection startup. This
@@ -700,14 +703,16 @@ private void handleFailure(Throwable ex)  {
 
     /** private API */
     public void doFinalShutdown() {
-        _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();
+        if (finalShutdownStarted.compareAndSet(false, true)) {
+            _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() {