Dispatch heartbeat failure in dedicated thread in NIO mode

Otherwise connection recovery can occur in the same NIO thread, leading
to a deadlock.

Fixes #413

(cherry picked from commit 46d3d2e)
(cherry picked from commit 4f900f0)

Conflicts:
	src/main/java/com/rabbitmq/client/impl/nio/NioLoop.java

Author: acogoluegnes

src/main/java/com/rabbitmq/client/impl/nio/NioLoop.java

@@ -77,7 +77,7 @@ public void run() {
                         long now = System.currentTimeMillis();
                         if ((now - state.getLastActivity()) > state.getConnection().getHeartbeat() * 1000 * 2) {
                             try {
-                                state.getConnection().handleHeartbeatFailure();
+                                handleHeartbeatFailure(state);
                             } catch (Exception e) {
                                 LOGGER.warn("Error after heartbeat failure of connection {}", state.getConnection());
                             } catch (AssertionError e) {
@@ -270,48 +270,68 @@ protected void handleIoError(SocketChannelFrameHandlerState state, Throwable ex)
         }
     }
 
+    protected void handleHeartbeatFailure(final SocketChannelFrameHandlerState state) {
+        if (needToDispatchIoError(state)) {
+            dispatchShutdownToConnection(
+                new Runnable() {
+
+                    @Override
+                    public void run() {
+                        state.getConnection().handleHeartbeatFailure();
+                    }
+                },
+                state.getConnection().toString()
+            );
+        } else {
+            try {
+                state.close();
+            } catch (IOException e) {
+
+            }
+        }
+    }
+
     protected boolean needToDispatchIoError(final SocketChannelFrameHandlerState state) {
         return state.getConnection().isOpen();
     }
 
     protected void dispatchIoErrorToConnection(final SocketChannelFrameHandlerState state, final Throwable ex) {
-        // In case of recovery after the shutdown,
-        // the new connection shouldn't be initialized in
-        // the NIO thread, to avoid a deadlock.
-        Runnable shutdown = new Runnable() {
+        dispatchShutdownToConnection(
+            new Runnable() {
 
-            @Override
-            public void run() {
-                try {
+                @Override
+                public void run() {
                     state.getConnection().handleIoError(ex);
-                } catch(AssertionError e) {
-                    LOGGER.warn("Assertion error during error dispatching to connection: " + e.getMessage());
                 }
-            }
-        };
-        if (executorService() == null) {
-            String name = "rabbitmq-connection-shutdown-" + state.getConnection();
-            Thread shutdownThread = Environment.newThread(threadFactory(), shutdown, name);
-            shutdownThread.start();
-        } else {
-            executorService().submit(shutdown);
-        }
+            },
+            state.getConnection().toString()
+        );
     }
 
     protected void dispatchShutdownToConnection(final SocketChannelFrameHandlerState state) {
-        Runnable shutdown = new Runnable() {
-            @Override
-            public void run() {
-                state.getConnection().doFinalShutdown();
-            }
-        };
+        dispatchShutdownToConnection(
+            new Runnable() {
+
+                @Override
+                public void run() {
+                    state.getConnection().doFinalShutdown();
+                }
+            },
+            state.getConnection().toString()
+        );
+    }
+
+    protected void dispatchShutdownToConnection(Runnable connectionShutdownRunnable, String connectionName) {
+        // In case of recovery after the shutdown,
+        // the new connection shouldn't be initialized in
+        // the NIO thread, to avoid a deadlock.
         if (this.connectionShutdownExecutor != null) {
-            connectionShutdownExecutor.execute(shutdown);
+            connectionShutdownExecutor.execute(connectionShutdownRunnable);
         } else if (executorService() != null) {
-            executorService().execute(shutdown);
+            executorService().execute(connectionShutdownRunnable);
         } else {
-            String name = "rabbitmq-connection-shutdown-" + state.getConnection();
-            Thread shutdownThread = Environment.newThread(threadFactory(), shutdown, name);
+            String name = "rabbitmq-connection-shutdown-" + connectionName;
+            Thread shutdownThread = Environment.newThread(threadFactory(), connectionShutdownRunnable, name);
             shutdownThread.start();
         }
     }