/*
 * Copyright 2009 Mike Cumings
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.kenai.jbosh;

import com.kenai.jbosh.ComposableBody.Builder;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * BOSH Client session instance.  Each communication session with a remote
 * connection manager is represented and handled by an instance of this
 * class.  This is the main entry point for client-side communications.
 * To create a new session, a client configuration must first be created
 * and then used to create a client instance:
 * <pre>
 * BOSHClientConfig cfg = BOSHClientConfig.Builder.create(
 *         "http://server:1234/httpbind", "jabber.org")
 *     .setFrom("user@jabber.org")
 *     .build();
 * BOSHClient client = BOSHClient.create(cfg);
 * </pre>
 * Additional client configuration options are available.  See the
 * {@code BOSHClientConfig.Builder} class for more information.
 * <p/>
 * Once a {@code BOSHClient} instance has been created, communication with
 * the remote connection manager can begin.  No attempt will be made to
 * establish a connection to the connection manager until the first call
 * is made to the {@code send(ComposableBody)} method.  Note that it is
 * possible to send an empty body to cause an immediate connection attempt
 * to the connection manager.  Sending an empty message would look like
 * the following:
 * <pre>
 * client.send(ComposableBody.builder().build());
 * </pre>
 * For more information on creating body messages with content, see the
 * {@code ComposableBody.Builder} class documentation.
 * <p/>
 * Once a session has been successfully started, the client instance can be
 * used to send arbitrary payload data.  All aspects of the BOSH
 * protocol involving setting and processing attributes in the BOSH
 * namespace will be handled by the client code transparently and behind the
 * scenes.  The user of the client instance can therefore concentrate
 * entirely on the content of the message payload, leaving the semantics of
 * the BOSH protocol to the client implementation.
 * <p/>
 * To be notified of incoming messages from the remote connection manager,
 * a {@code BOSHClientResponseListener} should be added to the client instance.
 * All incoming messages will be published to all response listeners as they
 * arrive and are processed.  As with the transmission of payload data via
 * the {@code send(ComposableBody)} method, there is no need to worry about
 * handling of the BOSH attributes, since this is handled behind the scenes.
 * <p/>
 * If the connection to the remote connection manager is terminated (either
 * explicitly or due to a terminal condition of some sort), all connection
 * listeners will be notified.  After the connection has been closed, the
 * client instance is considered dead and a new one must be created in order
 * to resume communications with the remote server.
 * <p/>
 * Instances of this class are thread-safe.
 *
 * @see BOSHClientConfig.Builder
 * @see BOSHClientResponseListener
 * @see BOSHClientConnListener
 * @see ComposableBody.Builder
 */
public final class BOSHClient {

    /**
     * Logger.
     */
    private static final Logger LOG = Logger.getLogger(
            BOSHClient.class.getName());

    /**
     * Value of the 'type' attribute used for session termination.
     */
    private static final String TERMINATE = "terminate";
    
    /**
     * Value of the 'type' attribute used for recoverable errors.
     */
    private static final String ERROR = "error";

    /**
     * Message to use for interrupted exceptions.
     */
    private static final String INTERRUPTED = "Interrupted";

    /**
     * Message used for unhandled exceptions.
     */
    private static final String UNHANDLED = "Unhandled Exception";

    /**
     * Message used whena null listener is detected.
     */
    private static final String NULL_LISTENER = "Listener may not b enull";

    /**
     * Default empty request delay.
     */
    private static final int DEFAULT_EMPTY_REQUEST_DELAY = 100;

    /**
     * Amount of time to wait before sending an empty request, in
     * milliseconds.
     */
    private static final int EMPTY_REQUEST_DELAY = Integer.getInteger(
            BOSHClient.class.getName() + ".emptyRequestDelay",
            DEFAULT_EMPTY_REQUEST_DELAY);

    /**
     * Default value for the pause margin.
     */
    private static final int DEFAULT_PAUSE_MARGIN = 500;

    /**
     * The amount of time in milliseconds which will be reserved as a
     * safety margin when scheduling empty requests against a maxpause
     * value.   This should give us enough time to build the message
     * and transport it to the remote host.
     */
    private static final int PAUSE_MARGIN = Integer.getInteger(
            BOSHClient.class.getName() + ".pauseMargin",
            DEFAULT_PAUSE_MARGIN);
    
    /**
     * Flag indicating whether or not we want to perform assertions.
     */
    private static final boolean ASSERTIONS;

    /**
     * Connection listeners.
     */
    private final Set<BOSHClientConnListener> connListeners =
            new CopyOnWriteArraySet<BOSHClientConnListener>();

    /**
     * Request listeners.
     */
    private final Set<BOSHClientRequestListener> requestListeners =
            new CopyOnWriteArraySet<BOSHClientRequestListener>();

    /**
     * Response listeners.
     */
    private final Set<BOSHClientResponseListener> responseListeners =
            new CopyOnWriteArraySet<BOSHClientResponseListener>();

    /**
     * Lock instance.
     */
    private final ReentrantLock lock = new ReentrantLock();

    /**
     * Condition indicating that there are messages to be exchanged.
     */
    private final Condition notEmpty = lock.newCondition();

    /**
     * Condition indicating that there are available slots for sending
     * messages.
     */
    private final Condition notFull = lock.newCondition();

    /**
     * Condition indicating that there are no outstanding connections.
     */
    private final Condition drained = lock.newCondition();

    /**
     * Session configuration.
     */
    private final BOSHClientConfig cfg;

    /**
     * Processor thread runnable instance.
     */
    private final Runnable procRunnable = new Runnable() {
        /**
         * Process incoming messages.
         */
        public void run() {
            processMessages();
        }
    };

    /**
     * Processor thread runnable instance.
     */
    private final Runnable emptyRequestRunnable = new Runnable() {
        /**
         * Process incoming messages.
         */
        public void run() {
            sendEmptyRequest();
        }
    };

    /**
     * HTTPSender instance.
     */
    private final HTTPSender httpSender =
            new ApacheHTTPSender();

    /**
     * Storage for test hook implementation.
     */
    private final AtomicReference<ExchangeInterceptor> exchInterceptor =
            new AtomicReference<ExchangeInterceptor>();

    /**
     * Request ID sequence to use for the session.
     */
    private final RequestIDSequence requestIDSeq = new RequestIDSequence();

    /**
     * ScheduledExcecutor to use for deferred tasks.
     */
    private final ScheduledExecutorService schedExec =
            Executors.newSingleThreadScheduledExecutor();

    /************************************************************
     * The following vars must be accessed via the lock instance.
     */

    /**
     * Thread which is used to process responses from the connection
     * manager.  Becomes null when session is terminated.
     */
    private Thread procThread;

    /**
     * Future for sending a deferred empty request, if needed.
     */
    private ScheduledFuture emptyRequestFuture;

    /**
     * Connection Manager session parameters.  Only available when in a
     * connected state.
     */
    private CMSessionParams cmParams;

    /**
     * List of active/outstanding requests.
     */
    private Queue<HTTPExchange> exchanges = new LinkedList<HTTPExchange>();

    /**
     * Set of RIDs which have been received, for the purpose of sending
     * response acknowledgements.
     */
    private SortedSet<Long> pendingResponseAcks = new TreeSet<Long>();
    
    /**
     * The highest RID that we've already received a response for.  This value
     * is used to implement response acks.
     */
    private Long responseAck = Long.valueOf(-1L);

    /**
     * List of requests which have been made but not yet acknowledged.  This
     * list remains unpopulated if the CM is not acking requests.
     */
    private List<ComposableBody> pendingRequestAcks =
            new ArrayList<ComposableBody>();

    ///////////////////////////////////////////////////////////////////////////
    // Classes:

    /**
     * Class used in testing to dynamically manipulate received exchanges
     * at test runtime.
     */
    abstract static class ExchangeInterceptor {
        /**
         * Limit construction.
         */
        ExchangeInterceptor() {
            // Empty;
        }

        /**
         * Hook to manipulate an HTTPExchange as is is about to be processed.
         *
         * @param exch original exchange that would be processed
         * @return replacement exchange instance, or {@code null} to skip
         *  processing of this exchange
         */
        abstract HTTPExchange interceptExchange(final HTTPExchange exch);
    }

    ///////////////////////////////////////////////////////////////////////////
    // Constructors:

    /**
     * Determine whether or not we should perform assertions.  Assertions
     * can be specified via system property explicitly, or defaulted to
     * the JVM assertions status.
     */
    static {
        final String prop =
                BOSHClient.class.getSimpleName() + ".assertionsEnabled";
        boolean enabled = false;
        if (System.getProperty(prop) == null) {
            assert enabled = true;
        } else {
            enabled = Boolean.getBoolean(prop);
        }
        ASSERTIONS = enabled;
    }

    /**
     * Prevent direct construction.
     */
    private BOSHClient(final BOSHClientConfig sessCfg) {
        cfg = sessCfg;
        init();
    }

    ///////////////////////////////////////////////////////////////////////////
    // Public methods:

    /**
     * Create a new BOSH client session using the client configuration
     * information provided.
     *
     * @param clientCfg session configuration
     * @return BOSH session instance
     */
    public static BOSHClient create(final BOSHClientConfig clientCfg) {
        if (clientCfg == null) {
            throw(new IllegalArgumentException(
                    "Client configuration may not be null"));
        }
        return new BOSHClient(clientCfg);
    }

    /**
     * Get the client configuration that was used to create this client
     * instance.
     *
     * @return client configuration
     */
    public BOSHClientConfig getBOSHClientConfig() {
        return cfg;
    }

    /**
     * Adds a connection listener to the session.
     *
     * @param listener connection listener to add, if not already added
     */
    public void addBOSHClientConnListener(
            final BOSHClientConnListener listener) {
        if (listener == null) {
            throw(new IllegalArgumentException(NULL_LISTENER));
        }
        connListeners.add(listener);
    }

    /**
     * Removes a connection listener from the session.
     *
     * @param listener connection listener to remove, if previously added
     */
    public void removeBOSHClientConnListener(
            final BOSHClientConnListener listener) {
        if (listener == null) {
            throw(new IllegalArgumentException(NULL_LISTENER));
        }
        connListeners.remove(listener);
    }

    /**
     * Adds a request message listener to the session.
     *
     * @param listener request listener to add, if not already added
     */
    public void addBOSHClientRequestListener(
            final BOSHClientRequestListener listener) {
        if (listener == null) {
            throw(new IllegalArgumentException(NULL_LISTENER));
        }
        requestListeners.add(listener);
    }

    /**
     * Removes a request message listener from the session, if previously
     * added.
     *
     * @param listener instance to remove
     */
    public void removeBOSHClientRequestListener(
            final BOSHClientRequestListener listener) {
        if (listener == null) {
            throw(new IllegalArgumentException(NULL_LISTENER));
        }
        requestListeners.remove(listener);
    }

    /**
     * Adds a response message listener to the session.
     *
     * @param listener response listener to add, if not already added
     */
    public void addBOSHClientResponseListener(
            final BOSHClientResponseListener listener) {
        if (listener == null) {
            throw(new IllegalArgumentException(NULL_LISTENER));
        }
        responseListeners.add(listener);
    }

    /**
     * Removes a response message listener from the session, if previously
     * added.
     *
     * @param listener instance to remove
     */
    public void removeBOSHClientResponseListener(
            final BOSHClientResponseListener listener) {
        if (listener == null) {
            throw(new IllegalArgumentException(NULL_LISTENER));
        }
        responseListeners.remove(listener);
    }

    /**
     * Send the provided message data to the remote connection manager.  The
     * provided message body does not need to have any BOSH-specific attribute
     * information set.  It only needs to contain the actual message payload
     * that should be delivered to the remote server.
     * <p/>
     * The first call to this method will result in a connection attempt
     * to the remote connection manager.  Subsequent calls to this method
     * will block until the underlying session state allows for the message
     * to be transmitted.  In certain scenarios - such as when the maximum
     * number of outbound connections has been reached - calls to this method
     * will block for short periods of time.
     *
     * @param body message data to send to remote server
     * @throws BOSHException on message transmission failure
     */
    public void send(final ComposableBody body) throws BOSHException {
        assertUnlocked();
        if (body == null) {
            throw(new IllegalArgumentException(
                    "Message body may not be null"));
        }

        HTTPExchange exch;
        CMSessionParams params;
        lock.lock();
        try {
            blockUntilSendable(body);
            if (!isWorking() && !isTermination(body)) {
                throw(new BOSHException(
                        "Cannot send message when session is closed"));
            }
            
            long rid = requestIDSeq.getNextRID();
            ComposableBody request = body;
            params = cmParams;
            if (params == null && exchanges.isEmpty()) {
                // This is the first message being sent
                request = applySessionCreationRequest(rid, body);
            } else {
                request = applySessionData(rid, body);
                if (cmParams.isAckingRequests()) {
                    pendingRequestAcks.add(request);
                }
            }
            exch = new HTTPExchange(request);
            exchanges.add(exch);
            notEmpty.signalAll();
            clearEmptyRequest();
        } finally {
            lock.unlock();
        }
        AbstractBody finalReq = exch.getRequest();
        HTTPResponse resp = httpSender.send(params, finalReq);
        exch.setHTTPResponse(resp);
        fireRequestSent(finalReq);
    }

    /**
     * Attempt to pause the current session.  When supported by the remote
     * connection manager, pausing the session will result in the connection
     * manager closing out all outstanding requests (including the pause
     * request) and increases the inactivity timeout of the session.  The
     * exact value of the temporary timeout is dependent upon the connection
     * manager.  This method should be used if a client encounters an
     * exceptional temporary situation during which it will be unable to send
     * requests to the connection manager for a period of time greater than
     * the maximum inactivity period.
     *
     * The session will revert back to it's normal, unpaused state when the
     * client sends it's next message.
     *
     * @return {@code true} if the connection manager supports session pausing,
     *  {@code false} if the connection manager does not support session
     *  pausing or if the session has not yet been established
     */
    public boolean pause() {
        assertUnlocked();
        lock.lock();
        AttrMaxPause maxPause = null;
        try {
            if (cmParams == null) {
                return false;
            }

            maxPause = cmParams.getMaxPause();
            if (maxPause == null) {
                return false;
            }
        } finally {
            lock.unlock();
        }
        try {
            send(ComposableBody.builder()
                    .setAttribute(Attributes.PAUSE, maxPause.toString())
                    .build());
        } catch (BOSHException boshx) {
            LOG.log(Level.FINEST, "Could not send pause", boshx);
        }
        return true;
    }

    /**
     * End the BOSH session by disconnecting from the remote BOSH connection
     * manager.
     *
     * @throws BOSHException when termination message cannot be sent
     */
    public void disconnect() throws BOSHException {
        disconnect(ComposableBody.builder().build());
    }

    /**
     * End the BOSH session by disconnecting from the remote BOSH connection
     * manager, sending the provided content in the final connection
     * termination message.
     *
     * @param msg final message to send
     * @throws BOSHException when termination message cannot be sent
     */
    public void disconnect(final ComposableBody msg) throws BOSHException {
        if (msg == null) {
            throw(new IllegalArgumentException(
                    "Message body may not be null"));
        }

        Builder builder = msg.rebuild();
        builder.setAttribute(Attributes.TYPE, TERMINATE);
        send(builder.build());
    }

    /**
     * Forcibly close this client session instance.  The preferred mechanism
     * to close the connection is to send a disconnect message and wait for
     * organic termination.  Calling this method simply shuts down the local
     * session without sending a termination message, releasing all resources
     * associated with the session.
     */
    public void close() {
        dispose(new BOSHException("Session explicitly closed by caller"));
    }

    ///////////////////////////////////////////////////////////////////////////
    // Package-private methods:

    /**
     * Get the current CM session params.
     *
     * @return current session params, or {@code null}
     */
    CMSessionParams getCMSessionParams() {
        lock.lock();
        try {
            return cmParams;
        } finally {
            lock.unlock();
        }
    }

    /**
     * Wait until no more messages are waiting to be processed.
     */
    void drain() {
        lock.lock();
        try {
            LOG.finest("Waiting while draining...");
            while (isWorking()
                    && (emptyRequestFuture == null
                    || emptyRequestFuture.isDone())) {
                try {
                    drained.await();
                } catch (InterruptedException intx) {
                    LOG.log(Level.FINEST, INTERRUPTED, intx);
                }
            }
            LOG.finest("Drained");
        } finally {
            lock.unlock();
        }
    }

    /**
     * Test method used to forcibly discard next exchange.
     *
     * @param interceptor exchange interceptor
     */
    void setExchangeInterceptor(final ExchangeInterceptor interceptor) {
        exchInterceptor.set(interceptor);
    }


    ///////////////////////////////////////////////////////////////////////////
    // Private methods:

    /**
     * Initialize the session.  This initializes the underlying HTTP
     * transport implementation and starts the receive thread.
     */
    private void init() {
        assertUnlocked();
        
        lock.lock();
        try {
            httpSender.init(cfg);
            procThread = new Thread(procRunnable);
            procThread.setDaemon(true);
            procThread.setName(BOSHClient.class.getSimpleName()
                    + "[" + System.identityHashCode(this)
                    + "]: Receive thread");
            procThread.start();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Destroy this session.
     *
     * @param cause the reason for the session termination, or {@code null}
     *  for normal termination
     */
    private void dispose(final Throwable cause) {
        assertUnlocked();
        
        lock.lock();
        try {
            if (procThread == null) {
                // Already disposed
                return;
            }
            procThread = null;
        } finally {
            lock.unlock();
        }

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

        lock.lock();
        try {
            clearEmptyRequest();
            exchanges = null;
            cmParams = null;
            pendingResponseAcks = null;
            pendingRequestAcks = null;
            notEmpty.signalAll();
            notFull.signalAll();
            drained.signalAll();
        } finally {
            lock.unlock();
        }
        
        httpSender.destroy();
        schedExec.shutdownNow();
    }

    /**
     * Determines if the message body specified indicates a request to
     * pause the session.
     *
     * @param msg message to evaluate
     * @return {@code true} if the message is a pause request, {@code false}
     *  otherwise
     */
    private static boolean isPause(final AbstractBody msg) {
        return msg.getAttribute(Attributes.PAUSE) != null;
    }
    
    /**
     * Determines if the message body specified indicates a termination of
     * the session.
     *
     * @param msg message to evaluate
     * @return {@code true} if the message is a session termination,
     *  {@code false} otherwise
     */
    private static boolean isTermination(final AbstractBody msg) {
        return TERMINATE.equals(msg.getAttribute(Attributes.TYPE));
    }

    /**
     * Evaluates the HTTP response code and response message and returns the
     * terminal binding condition that it describes, if any.
     *
     * @param respCode HTTP response code
     * @param respBody response body
     * @return terminal binding condition, or {@code null} if not a terminal
     *  binding condition message
     */
    private TerminalBindingCondition getTerminalBindingCondition(
            final int respCode,
            final AbstractBody respBody) {
        assertLocked();

        if (isTermination(respBody)) {
            String str = respBody.getAttribute(Attributes.CONDITION);
            return TerminalBindingCondition.forString(str);
        }
        // Check for deprecated HTTP Error Conditions
        if (cmParams != null && cmParams.getVersion() == null) {
            return TerminalBindingCondition.forHTTPResponseCode(respCode);
        }
        return null;
    }

    /**
     * Determines if the message specified is immediately sendable or if it
     * needs to block until the session state changes.
     *
     * @param msg message to evaluate
     * @return {@code true} if the message can be immediately sent,
     *  {@code false} otherwise
     */
    private boolean isImmediatelySendable(final AbstractBody msg) {
        assertLocked();

        if (cmParams == null) {
            // block if we're waiting for a response to our first request
            return exchanges.isEmpty();
        }

        AttrRequests requests = cmParams.getRequests();
        if (requests == null) {
            return true;
        }
        int maxRequests = requests.intValue();
        if (exchanges.size() < maxRequests) {
            return true;
        }
        if (exchanges.size() == maxRequests
                && (isTermination(msg) || isPause(msg))) {
            // One additional terminate or pause message is allowed
            return true;
        }
        return false;
    }

    /**
     * Determines whether or not the session is still active.
     *
     * @return {@code true} if it is, {@code false} otherwise
     */
    private boolean isWorking() {
        assertLocked();

        return procThread != null;
    }

    /**
     * Blocks until either the message provided becomes immediately
     * sendable or until the session is terminated.
     *
     * @param msg message to evaluate
     */
    private void blockUntilSendable(final AbstractBody msg) {
        assertLocked();

        while (isWorking() && !isImmediatelySendable(msg)) {
            try {
                notFull.await();
            } catch (InterruptedException intx) {
                LOG.log(Level.FINEST, INTERRUPTED, intx);
            }
        }
    }

    /**
     * Modifies the specified body message such that it becomes a new
     * BOSH session creation request.
     *
     * @param rid request ID to use
     * @param orig original body to modify
     * @return modified message which acts as a session creation request
     */
    private ComposableBody applySessionCreationRequest(
            final long rid, final ComposableBody orig) throws BOSHException {
        assertLocked();
        
        Builder builder = orig.rebuild();
        builder.setAttribute(Attributes.TO, cfg.getTo());
        builder.setAttribute(Attributes.XML_LANG, cfg.getLang());
        builder.setAttribute(Attributes.VER,
                AttrVersion.getSupportedVersion().toString());
        builder.setAttribute(Attributes.WAIT, "60");
        builder.setAttribute(Attributes.HOLD, "1");
        builder.setAttribute(Attributes.RID, Long.toString(rid));
        applyRoute(builder);
        applyFrom(builder);
        builder.setAttribute(Attributes.ACK, "1");

        // Make sure the following are NOT present (i.e., during retries)
        builder.setAttribute(Attributes.SID, null);
        return builder.build();
    }

    /**
     * Applies routing information to the request message who's builder has
     * been provided.
     *
     * @param builder builder instance to add routing information to
     */
    private void applyRoute(final Builder builder) {
        assertLocked();
        
        String route = cfg.getRoute();
        if (route != null) {
            builder.setAttribute(Attributes.ROUTE, route);
        }
    }

    /**
     * Applies the local station ID information to the request message who's
     * builder has been provided.
     *
     * @param builder builder instance to add station ID information to
     */
    private void applyFrom(final Builder builder) {
        assertLocked();

        String from = cfg.getFrom();
        if (from != null) {
            builder.setAttribute(Attributes.FROM, from);
        }
    }

    /**
     * Applies existing session data to the outbound request, returning the
     * modified request.
     *
     * This method assumes the lock is currently held.
     *
     * @param rid request ID to use
     * @param orig original/raw request
     * @return modified request with session information applied
     */
    private ComposableBody applySessionData(
            final long rid,
            final ComposableBody orig) throws BOSHException {
        assertLocked();

        Builder builder = orig.rebuild();
        builder.setAttribute(Attributes.SID,
                cmParams.getSessionID().toString());
        builder.setAttribute(Attributes.RID, Long.toString(rid));
        applyResponseAcknowledgement(builder, rid);
        return builder.build();
    }

    /**
     * Sets the 'ack' attribute of the request to the value of the highest
     * 'rid' of a request for which it has already received a response in the
     * case where it has also received all responses associated with lower
     * 'rid' values.  The only exception is that, after its session creation
     * request, the client SHOULD NOT include an 'ack' attribute in any request
     * if it has received responses to all its previous requests.
     *
     * @param builder message builder
     * @param rid current request RID
     */
    private void applyResponseAcknowledgement(
            final Builder builder,
            final long rid) {
        assertLocked();

        if (responseAck.equals(Long.valueOf(-1L))) {
            // We have not received any responses yet
            return;
        }

        Long prevRID = Long.valueOf(rid - 1L);
        if (responseAck.equals(prevRID)) {
            // Implicit ack
            return;
        }
        
        builder.setAttribute(Attributes.ACK, responseAck.toString());
    }

    /**
     * While we are "connected", process received responses.
     *
     * This method is run in the processing thread.
     */
    private void processMessages() {
        LOG.log(Level.FINEST, "Processing thread starting");
        try {
            HTTPExchange exch;
            do {
                exch = nextExchange();
                if (exch == null) {
                    break;
                }

                // Test hook to manipulate what the client sees:
                ExchangeInterceptor interceptor = exchInterceptor.get();
                if (interceptor != null) {
                    HTTPExchange newExch = interceptor.interceptExchange(exch);
                    if (newExch == null) {
                        LOG.log(Level.FINE, "Discarding exchange on request "
                                + "of test hook: RID="
                                + exch.getRequest().getAttribute(
                                    Attributes.RID));
                        lock.lock();
                        try {
                            exchanges.remove(exch);
                        } finally {
                            lock.unlock();
                        }
                        continue;
                    }
                    exch = newExch;
                }

                processExchange(exch);
            } while (true);
        } finally {
            LOG.log(Level.FINEST, "Processing thread exiting");
        }

    }

    /**
     * Get the next message exchange to process, blocking until one becomes
     * available if nothing is already waiting for processing.
     *
     * @return next available exchange to process, or {@code null} if no
     *  exchanges are immediately available
     */
    private HTTPExchange nextExchange() {
        assertUnlocked();

        final Thread thread = Thread.currentThread();
        HTTPExchange exch = null;
        lock.lock();
        try {
            do {
                if (!thread.equals(procThread)) {
                    break;
                }
                exch = exchanges.peek();
                if (exch == null) {
                    try {
                        notEmpty.await();
                    } catch (InterruptedException intx) {
                        LOG.log(Level.FINEST, INTERRUPTED, intx);
                    }
                }
            } while (exch == null);
        } finally {
            lock.unlock();
        }
        return exch;
    }

    /**
     * Process the next, provided exchange.  This is the main processing
     * method of the receive thread.
     *
     * @param exch message exchange to process
     */
    private void processExchange(final HTTPExchange exch) {
        assertUnlocked();

        HTTPResponse resp;
        AbstractBody body;
        int respCode;
        try {
            resp = exch.getHTTPResponse();
            body = resp.getBody();
            respCode = resp.getHTTPStatus();
        } catch (BOSHException boshx) {
            LOG.log(Level.FINEST, "Could not obtain response", boshx);
            dispose(boshx);
            return;
        } catch (InterruptedException intx) {
            LOG.log(Level.FINEST, INTERRUPTED, intx);
            dispose(intx);
            return;
        }
        fireResponseReceived(body);

        // Process the message with the current session state
        AbstractBody req = exch.getRequest();
        CMSessionParams params;
        List<HTTPExchange> toResend = null;
        lock.lock();
        try {
            // Check for session creation response info, if needed
            if (cmParams == null) {
                cmParams = CMSessionParams.fromSessionInit(req, body);

                // The following call handles the lock. It's not an escape.
                fireConnectionEstablished();
            }
            params = cmParams;

            checkForTerminalBindingConditions(body, respCode);
            if (isTermination(body)) {
                // Explicit termination
                lock.unlock();
                dispose(null);
                return;
            }
            
            if (isRecoverableBindingCondition(body)) {
                // Retransmit outstanding requests
                if (toResend == null) {
                    toResend = new ArrayList<HTTPExchange>(exchanges.size());
                }
                for (HTTPExchange exchange : exchanges) {
                    HTTPExchange resendExch =
                            new HTTPExchange(exchange.getRequest());
                    toResend.add(resendExch);
                }
                for (HTTPExchange exchange : toResend) {
                    exchanges.add(exchange);
                }
            } else {
                // Process message as normal
                processRequestAcknowledgements(req, body);
                processResponseAcknowledgementData(req);
                HTTPExchange resendExch =
                        processResponseAcknowledgementReport(body);
                if (resendExch != null && toResend == null) {
                    toResend = new ArrayList<HTTPExchange>(1);
                    toResend.add(resendExch);
                    exchanges.add(resendExch);
                }
            }
        } catch (BOSHException boshx) {
            LOG.log(Level.FINEST, "Could not process response", boshx);
            lock.unlock();
            dispose(boshx);
            return;
        } finally {
            if (lock.isHeldByCurrentThread()) {
                try {
                    exchanges.remove(exch);
                    if (exchanges.isEmpty()) {
                        scheduleEmptyRequest(processPauseRequest(req));
                    }
                    notFull.signalAll();
                } finally {
                    lock.unlock();
                }
            }
        }

        if (toResend != null) {
            for (HTTPExchange resend : toResend) {
                HTTPResponse response =
                        httpSender.send(params, resend.getRequest());
                resend.setHTTPResponse(response);
                fireRequestSent(resend.getRequest());
            }
        }
    }
    
    /**
     * Clears any scheduled empty requests.
     */
    private void clearEmptyRequest() {
        assertLocked();

        if (emptyRequestFuture != null) {
            emptyRequestFuture.cancel(false);
            emptyRequestFuture = null;
        }
    }

    /**
     * Calculates the default empty request delay/interval to use for the
     * active session.
     *
     * @return delay in milliseconds
     */
    private long getDefaultEmptyRequestDelay() {
        assertLocked();
        
        // Figure out how long we should wait before sending an empty request
        AttrPolling polling = cmParams.getPollingInterval();
        long delay;
        if (polling == null) {
            delay = EMPTY_REQUEST_DELAY;
        } else {
            delay = polling.getInMilliseconds();
        }
        return delay;
    }

    /**
     * Schedule an empty request to be sent if no other requests are
     * sent in a reasonable amount of time.
     */
    private void scheduleEmptyRequest(long delay) {
        assertLocked();
        if (delay < 0L) {
            throw(new IllegalArgumentException(
                    "Empty request delay must be >= 0 (was: " + delay + ")"));
        }

        clearEmptyRequest();
        if (!isWorking()) {
            return;
        }
        
        // Schedule the transmission
        if (LOG.isLoggable(Level.FINER)) {
            LOG.finer("Scheduling empty request in " + delay + "ms");
        }
        try {
            emptyRequestFuture = schedExec.schedule(emptyRequestRunnable,
                    delay, TimeUnit.MILLISECONDS);
        } catch (RejectedExecutionException rex) {
            LOG.log(Level.FINEST, "Could not schedule empty request", rex);
        }
        drained.signalAll();
    }

    /**
     * Sends an empty request to maintain session requirements.  If a request
     * is sent within a reasonable time window, the empty request transmission
     * will be cancelled.
     */
    private void sendEmptyRequest() {
        assertUnlocked();
        // Send an empty request
        LOG.finest("Sending empty request");
        try {
            send(ComposableBody.builder().build());
        } catch (BOSHException boshx) {
            dispose(boshx);
        }
    }

    /**
     * Assert that the internal lock is held.
     */
    private void assertLocked() {
        if (ASSERTIONS) {
            if (!lock.isHeldByCurrentThread()) {
                throw(new AssertionError("Lock is not held by current thread"));
            }
            return;
        }
    }

    /**
     * Assert that the internal lock is *not* held.
     */
    private void assertUnlocked() {
        if (ASSERTIONS) {
            if (lock.isHeldByCurrentThread()) {
                throw(new AssertionError("Lock is held by current thread"));
            }
            return;
        }
    }

    /**
     * Checks to see if the response indicates a terminal binding condition
     * (as per XEP-0124 section 17).  If it does, an exception is thrown.
     *
     * @param body response body to evaluate
     * @param code HTTP response code
     * @throws BOSHException if a terminal binding condition is detected
     */
    private void checkForTerminalBindingConditions(
            final AbstractBody body,
            final int code)
            throws BOSHException {
        TerminalBindingCondition cond =
                getTerminalBindingCondition(code, body);
        if (cond != null) {
            throw(new BOSHException(
                    "Terminal binding condition encountered: "
                    + cond.getCondition() + "  ("
                    + cond.getMessage() + ")"));
        }
    }

    /**
     * Determines whether or not the response indicates a recoverable
     * binding condition (as per XEP-0124 section 17).
     *
     * @param resp response body
     * @return {@code true} if it does, {@code false} otherwise
     */
    private static boolean isRecoverableBindingCondition(
            final AbstractBody resp) {
        return ERROR.equals(resp.getAttribute(Attributes.TYPE));
    }

    /**
     * Process the request to determine if the empty request delay
     * can be determined by looking to see if the request is a pause
     * request.  If it can, the request's delay is returned, otherwise
     * the default delay is returned.
     * 
     * @return delay in milliseconds that should elapse prior to an
     *  empty message being sent
     */
    private long processPauseRequest(
            final AbstractBody req) {
        assertLocked();

        if (cmParams != null && cmParams.getMaxPause() != null) {
            try {
                AttrPause pause = AttrPause.createFromString(
                        req.getAttribute(Attributes.PAUSE));
                if (pause != null) {
                    long delay = pause.getInMilliseconds() - PAUSE_MARGIN;
                    if (delay < 0) {
                        delay = EMPTY_REQUEST_DELAY;
                    }
                    return delay;
                }
            } catch (BOSHException boshx) {
                LOG.log(Level.FINEST, "Could not extract", boshx);
            }
        }

        return getDefaultEmptyRequestDelay();
    }

    /**
     * Check the response for request acknowledgements and take appropriate
     * action.
     *
     * This method assumes the lock is currently held.
     *
     * @param req request
     * @param resp response
     */
    private void processRequestAcknowledgements(
            final AbstractBody req, final AbstractBody resp) {
        assertLocked();
        
        if (!cmParams.isAckingRequests()) {
            return;
        }

        // If a report or time attribute is set, we aren't acking anything
        if (resp.getAttribute(Attributes.REPORT) != null) {
            return;
        }

        // Figure out what the highest acked RID is
        String acked = resp.getAttribute(Attributes.ACK);
        Long ackUpTo;
        if (acked == null) {
            // Implicit ack of all prior requests up until RID
            ackUpTo = Long.parseLong(req.getAttribute(Attributes.RID));
        } else {
            ackUpTo = Long.parseLong(acked);
        }

        // Remove the acked requests from the list
        if (LOG.isLoggable(Level.FINEST)) {
            LOG.finest("Removing pending acks up to: " + ackUpTo);
        }
        Iterator<ComposableBody> iter = pendingRequestAcks.iterator();
        while (iter.hasNext()) {
            AbstractBody pending = iter.next();
            Long pendingRID = Long.parseLong(
                    pending.getAttribute(Attributes.RID));
            if (pendingRID.compareTo(ackUpTo) <= 0) {
                iter.remove();
            }
        }
    }

    /**
     * Process the response in order to update the response acknowlegement
     * data.
     *
     * This method assumes the lock is currently held.
     *
     * @param req request
     */
    private void processResponseAcknowledgementData(
            final AbstractBody req) {
        assertLocked();
        
        Long rid = Long.parseLong(req.getAttribute(Attributes.RID));
        if (responseAck.equals(Long.valueOf(-1L))) {
            // This is the first request
            responseAck = rid;
        } else {
            pendingResponseAcks.add(rid);
            // Remove up until the first missing response (or end of queue)
            Long whileVal = responseAck;
            while (whileVal.equals(pendingResponseAcks.first())) {
                responseAck = whileVal;
                pendingResponseAcks.remove(whileVal);
                whileVal = Long.valueOf(whileVal.longValue() + 1);
            }
        }
    }

    /**
     * Process the response in order to check for and respond to any potential
     * ack reports.
     *
     * This method assumes the lock is currently held.
     *
     * @param resp response
     * @return exchange to transmit if a resend is to be performed, or
     *  {@code null} if no resend is necessary
     * @throws BOSHException when a a retry is needed but cannot be performed
     */
    private HTTPExchange processResponseAcknowledgementReport(
            final AbstractBody resp)
            throws BOSHException {
        assertLocked();
        
        String reportStr = resp.getAttribute(Attributes.REPORT);
        if (reportStr == null) {
            // No report on this message
            return null;
        }
        
        Long report = Long.parseLong(reportStr);
        Long time = Long.parseLong(resp.getAttribute(Attributes.TIME));
        if (LOG.isLoggable(Level.FINE)) {
            LOG.fine("Received report of missing request (RID="
                    + report + ", time=" + time + "ms)");
        }

        // Find the missing request
        Iterator<ComposableBody> iter = pendingRequestAcks.iterator();
        AbstractBody req = null;
        while (iter.hasNext() && req == null) {
            AbstractBody pending = iter.next();
            Long pendingRID = Long.parseLong(
                    pending.getAttribute(Attributes.RID));
            if (report.equals(pendingRID)) {
                req = pending;
            }
        }

        if (req == null) {
            throw(new BOSHException("Report of missing message with RID '"
                    + reportStr
                    + "' but local copy of that request was not found"));
        }

        // Resend the missing request
        HTTPExchange exch = new HTTPExchange(req);
        exchanges.add(exch);
        notEmpty.signalAll();
        return exch;
    }

    /**
     * Notifies all request listeners that the specified request is being
     * sent.
     *
     * @param request request being sent
     */
    private void fireRequestSent(final AbstractBody request) {
        assertUnlocked();

        BOSHMessageEvent event = null;
        for (BOSHClientRequestListener listener : requestListeners) {
            if (event == null) {
                event = BOSHMessageEvent.createRequestSentEvent(this, request);
            }
            try {
                listener.requestSent(event);
            } catch (Exception ex) {
                LOG.log(Level.WARNING, UNHANDLED, ex);
            }
        }
    }

    /**
     * Notifies all response listeners that the specified response has been
     * received.
     *
     * @param response response received
     */
    private void fireResponseReceived(final AbstractBody response) {
        assertUnlocked();

        BOSHMessageEvent event = null;
        for (BOSHClientResponseListener listener : responseListeners) {
            if (event == null) {
                event = BOSHMessageEvent.createResponseReceivedEvent(
                        this, response);
            }
            try {
                listener.responseReceived(event);
            } catch (Exception ex) {
                LOG.log(Level.WARNING, UNHANDLED, ex);
            }
        }
    }

    /**
     * Notifies all connection listeners that the session has been successfully
     * established.
     */
    private void fireConnectionEstablished() {
        final boolean hadLock = lock.isHeldByCurrentThread();
        if (hadLock) {
            lock.unlock();
        }
        try {
            BOSHClientConnEvent event = null;
            for (BOSHClientConnListener listener : connListeners) {
                if (event == null) {
                    event = BOSHClientConnEvent
                            .createConnectionEstablishedEvent(this);
                }
                try {
                    listener.connectionEvent(event);
                } catch (Exception ex) {
                    LOG.log(Level.WARNING, UNHANDLED, ex);
                }
            }
        } finally {
            if (hadLock) {
                lock.lock();
            }
        }
    }

    /**
     * Notifies all connection listeners that the session has been
     * terminated normally.
     */
    private void fireConnectionClosed() {
        assertUnlocked();

        BOSHClientConnEvent event = null;
        for (BOSHClientConnListener listener : connListeners) {
            if (event == null) {
                event = BOSHClientConnEvent.createConnectionClosedEvent(this);
            }
            try {
                listener.connectionEvent(event);
            } catch (Exception ex) {
                LOG.log(Level.WARNING, UNHANDLED, ex);
            }
        }
    }

    /**
     * Notifies all connection listeners that the session has been
     * terminated due to the exceptional condition provided.
     *
     * @param cause cause of the termination
     */
    private void fireConnectionClosedOnError(
            final Throwable cause) {
        assertUnlocked();

        BOSHClientConnEvent event = null;
        for (BOSHClientConnListener listener : connListeners) {
            if (event == null) {
                event = BOSHClientConnEvent
                        .createConnectionClosedOnErrorEvent(
                        this, pendingRequestAcks, cause);
            }
            try {
                listener.connectionEvent(event);
            } catch (Exception ex) {
                LOG.log(Level.WARNING, UNHANDLED, ex);
            }
        }
    }

}