public void close(int closeCode,                      String closeMessage,                      boolean initiatedByApplication,                      Throwable cause,                      int timeout,                      boolean abort)        throws IOException    {        boolean sync = !(Thread.currentThread() == mainLoopThread);        try {            AMQP.Connection.Close reason =                new AMQP.Connection.Close.Builder()                    .replyCode(closeCode)                    .replyText(closeMessage)                .build();            final ShutdownSignalException sse = startShutdown(reason, initiatedByApplication, cause, true);            if(sync){                BlockingRpcContinuation<AMQCommand> k = new BlockingRpcContinuation<AMQCommand>(){                    @Override                    public AMQCommand transformReply(AMQCommand command) {                        AMQConnection.this.finishShutdown(sse);                        return command;                    }};              _channel0.quiescingRpc(reason, k);              k.getReply(timeout);            } else {              _channel0.quiescingTransmit(reason);            }        } catch (TimeoutException tte) {            if (!abort) {                ShutdownSignalException sse = new ShutdownSignalException(true, true, null, this);                sse.initCause(cause);                throw sse;            }        } catch (ShutdownSignalException sse) {            if (!abort)                throw sse;        } catch (IOException ioe) {            if (!abort)                throw ioe;        } finally {            if(sync) {                _frameHandler.close();            }        }    }

public boolean processControlCommand(Command c) throws IOException    {        // Similar trick to ChannelN.processAsync used here, except        // we're interested in whole-connection quiescing.        // See the detailed comments in ChannelN.processAsync.        Method method = c.getMethod();        if (isOpen()) {            if (method instanceof AMQP.Connection.Close) {                handleConnectionClose(c);                return true;            } else if (method instanceof AMQP.Connection.Blocked) {                AMQP.Connection.Blocked blocked = (AMQP.Connection.Blocked) method;                try {                    for (BlockedListener l : this.blockedListeners) {                        l.handleBlocked(blocked.getReason());                    }                } catch (Throwable ex) {                    getExceptionHandler().handleBlockedListenerException(this, ex);                }                return true;            } else if (method instanceof AMQP.Connection.Unblocked) {                try {                    for (BlockedListener l : this.blockedListeners) {                        l.handleUnblocked();                    }                } catch (Throwable ex) {                    getExceptionHandler().handleBlockedListenerException(this, ex);                }                return true;            } else {                return false;            }        } else {            if (method instanceof AMQP.Connection.Close) {                // Already shutting down, so just send back a CloseOk.                try {                    _channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());                } catch (IOException ignored) { } // ignore                return true;            } else if (method instanceof AMQP.Connection.CloseOk) {                // It's our final "RPC". Time to shut down.                _running = false;                // If Close was sent from within the MainLoop we                // will not have a continuation to return to, so                // we treat this as processed in that case.                return !_channel0.isOutstandingRpc();            } else { // Ignore all others.                return true;            }        }    }