private void dispose(final Throwable cause) {
        assertUnlocked();
        
        lock.lock();

        if (procThreadUnjoined != null) {
            // dispose() has already been called from the procThread, which destroyed
            // everything but the thread itself.  Join the thread, and don't rerun
            // the rest, which has already been done.
            Thread thread;
            try {
                if(procThreadUnjoined == Thread.currentThread())
                    return;

                thread = procThreadUnjoined;
                procThreadUnjoined = null;
            } finally {
                // We must unlock before joining.
                lock.unlock();
            }

            Helpers.joinThreadUninterruptible(thread);
            return;
        }

        Thread thread = null;
        try {
            if (procThread == null) {
                // Already disposed
                return;
            }

            // If we're running from the thread, we can't join.  Stash the thread
            // in procThreadUnjoined, so when the user calls close() the thread is
            // guaranteed to be joined.
            if(procThread == Thread.currentThread()) {
                procThreadUnjoined = procThread;
            } else {
                thread = procThread;
            }
            procThread = null;
        } finally {
            lock.unlock();
        }

        if (cause == null) {
            fireConnectionClosed();
        } else {
            fireConnectionClosedOnError(cause);
        }

        lock.lock();
        try {
            clearEmptyRequest();
            clearTimeout();
            exchanges = null;
            cmParams = null;
            connectionRecoverablyLost = false;
            pendingResponseAcks = null;
            pendingRequestAcks = null;
            notEmpty.signalAll();
            notFull.signalAll();
        } finally {
            lock.unlock();
        }

        // Shut down the sender.  This will cause the thread to exit.
        httpSender.destroy();

        if(thread != null) {
            Helpers.joinThreadUninterruptible(thread);
        }
        
        if(schedExecOwned)
            schedExec.shutdownNow();
    }