/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 flex.messaging.client;

import flex.messaging.ConnectionAwareSession;
import flex.messaging.FlexContext;
import flex.messaging.FlexSession;
import flex.messaging.FlexSessionListener;
import flex.messaging.HttpFlexSession;
import flex.messaging.MessageClient;
import flex.messaging.MessageClientListener;
import flex.messaging.MessageException;
import flex.messaging.endpoints.Endpoint;
import flex.messaging.log.Log;
import flex.messaging.log.LogCategories;
import flex.messaging.messages.CommandMessage;
import flex.messaging.messages.Message;
import flex.messaging.services.MessageService;
import flex.messaging.util.StringUtils;
import flex.messaging.util.TimeoutAbstractObject;

import java.security.Principal;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * Represents a Flex client application instance on the server.
 */
public class FlexClient extends TimeoutAbstractObject implements FlexSessionListener, MessageClientListener {
    //--------------------------------------------------------------------------
    //
    // Public Static Constants
    //
    //--------------------------------------------------------------------------

    /**
     * Log category for FlexClient related messages.
     */
    public static final String FLEX_CLIENT_LOG_CATEGORY = LogCategories.CLIENT_FLEXCLIENT;

    /**
     * This value is passed to the server in an initial client connect to
     * indicate that the client needs a server-assigned FlexClient Id.
     */
    public static final String NULL_FLEXCLIENT_ID = "nil";

    //--------------------------------------------------------------------------
    //
    // Private Static Constants
    //
    //--------------------------------------------------------------------------

    /**
     * Error string constants.
     */
    private static final int FLEX_CLIENT_INVALIDATED = 10027;
    private static final int ENDPOINT_PUSH_HANDLER_ALREADY_REGISTERED = 10033;

    private static final String POLL_WAIT_THREAD_NAME_EXTENSION = "-in-poll-wait";

    //--------------------------------------------------------------------------
    //
    // Private Static Variables
    //
    //--------------------------------------------------------------------------

    /**
     * List of registered FlexClient created listeners.
     */
    private static final CopyOnWriteArrayList<FlexClientListener> createdListeners = new CopyOnWriteArrayList<FlexClientListener>();

    //--------------------------------------------------------------------------
    //
    // Public Static Methods
    //
    //--------------------------------------------------------------------------

    /**
     * Adds a create listener that will be notified when new FlexClients
     * are created.
     *
     * @param listener The listener to add.
     * @see flex.messaging.client.FlexClientListener
     */
    public static void addClientCreatedListener(FlexClientListener listener) {
        if (listener != null)
            createdListeners.addIfAbsent(listener);
    }

    /**
     * Removes a FlexClient created listener.
     *
     * @param listener The listener to remove.
     * @see flex.messaging.client.FlexClientListener
     */
    public static void removeClientCreatedListener(FlexClientListener listener) {
        if (listener != null)
            createdListeners.remove(listener);
    }

    //--------------------------------------------------------------------------
    //
    // Constructor
    //
    //--------------------------------------------------------------------------

    /**
     * Constructs a new FlexClient instance.
     *
     * @param manager The FlexClientManager managing this instance.
     */
    public FlexClient(FlexClientManager manager) {
        this(manager, FlexContext.getMessageBroker().createUUID());
    }

    /**
     * Constructs a new FlexClient instance having the specified Id.
     *
     * @param manager The FlexClientManager managing this instance.
     * @param id      The Id for this instance.
     */
    public FlexClient(FlexClientManager manager, String id) {
        this.id = id;
        flexClientManager = manager;
        updateLastUse();
        valid = true;

        if (Log.isDebug())
            Log.getLogger(FLEX_CLIENT_LOG_CATEGORY).debug("FlexClient created with id '" + this.id + "'.");
    }

    //--------------------------------------------------------------------------
    //
    // Variables
    //
    //--------------------------------------------------------------------------

    /**
     * Storage for custom attributes.
     */
    private volatile Map<String, Object> attributes;

    /**
     * List of registered FlexClient attribute listeners.
     */
    private volatile CopyOnWriteArrayList<FlexClientAttributeListener> attributeListeners;

    /**
     * List of registered FlexClient destroyed listeners.
     */
    private volatile CopyOnWriteArrayList<FlexClientListener> destroyedListeners;

    /**
     * The manager for the FlexClient.
     */
    final FlexClientManager flexClientManager;

    /**
     * The unique Id for the instance.
     */
    private final String id;

    /**
     * Flag used to break cycles during invalidation.
     */
    /* package visibility for FlexClientManager */ volatile boolean invalidating;

    /**
     * Instance level lock to sync for state changes.
     */
    final Object lock = new Object();

    /**
     * MessageClient subscriptions for this MessageClient.
     */
    private volatile CopyOnWriteArrayList<MessageClient> messageClients;

    /**
     * Queues of outbound messages to push to the client keyed by endpoint id.
     * Map(String endpointId, EndpointQueue queue).
     */
    private final Map<String, EndpointQueue> outboundQueues = new ConcurrentHashMap<String, EndpointQueue>(1);

    /**
     * EndpointPushHandlers keyed by endpointId that the FlexClient
     * can use to push messages to remote clients.
     * NOTE: these can't be added to the EndpointQueue data type because the existence of queues depends
     * upon client subscription state whereas endpoints that support push will generally set up their push
     * handling before any subscriptions have been created.
     */
    private Map<String, EndpointPushHandler> endpointPushHandlers;

    /**
     * Associated FlexSessions that represent the connections the FlexClient makes to the server.
     */
    private final CopyOnWriteArrayList<FlexSession> sessions = new CopyOnWriteArrayList<FlexSession>(); // We always have at least one session.

    /**
     * Flag indicating whether the instance is valid; once invalidated this flag is
     * set to false.
     */
    boolean valid;

    /**
     * The principal associated with this client.  Only used when perClientAuthentication
     * is being used.
     */
    private Principal userPrincipal;

    //--------------------------------------------------------------------------
    //
    // Public Methods
    //
    //--------------------------------------------------------------------------

    /**
     * Adds a FlexClient attribute listener that will be notified when an
     * attribute is added, removed or changed. If the attribute implements
     * FlexClientBindingListener, it will be notified before any
     * FlexClientAttributeListeners are notified.
     *
     * @param listener The listener to add.
     */
    public void addClientAttributeListener(FlexClientAttributeListener listener) {
        if (listener != null) {
            checkValid();

            synchronized (lock) {
                if (attributeListeners == null)
                    attributeListeners = new CopyOnWriteArrayList<FlexClientAttributeListener>();
            }

            attributeListeners.addIfAbsent(listener);
        }
    }

    /**
     * Adds a destroy listener that will be notified when the FlexClient
     * is destroyed. Listeners are notified after all attributes
     * have been unbound from the FlexClient and any FlexClientBindingListeners
     * and FlexClientAttributeListeners have been notified.
     *
     * @param listener The listener to add.
     * @see flex.messaging.client.FlexClientListener
     */
    public void addClientDestroyedListener(FlexClientListener listener) {
        if (listener != null) {
            checkValid();

            synchronized (lock) {
                if (destroyedListeners == null)
                    destroyedListeners = new CopyOnWriteArrayList<FlexClientListener>();
            }

            destroyedListeners.addIfAbsent(listener);
        }
    }

    /**
     * Returns the attribute bound to the specified name for the FlexClient, or null
     * if no attribute is bound under the name.
     *
     * @param name The name the attribute is bound to.
     * @return The attribute bound to the specified name.
     */
    public Object getAttribute(String name) {
        synchronized (lock) {
            checkValid();

            updateLastUse();

            return (attributes == null) ? null : attributes.get(name);
        }
    }

    /**
     * Returns a snapshot of the names of all attributes bound to the FlexClient.
     *
     * @return A snapshot of the names of all attributes bound to the FlexClient.
     */
    public Enumeration<String> getAttributeNames() {
        synchronized (lock) {
            checkValid();

            updateLastUse();

            if (attributes == null)
                return Collections.enumeration(Collections.<String>emptyList());

            // Return a copy so we do not run into concurrent modification problems if
            // someone adds to the attributes while iterating through the returned enumeration.
            return Collections.enumeration(new ArrayList<String>(attributes.keySet()));
        }
    }

    /**
     * Returns the push handler registered with the FlexClient with the supplied
     * endpoint id, or null if no push handler was registered with the FlexClient
     * for that endpoint.
     *
     * @return The push handler registered with the FlexClient with the supplied
     * endpoint id, or null if no push handler was registered with the FlexClient
     * for that endpoint.
     */
    public EndpointPushHandler getEndpointPushHandler(String endpointId) {
        synchronized (lock) {
            if (endpointPushHandlers != null && endpointPushHandlers.containsKey(endpointId))
                return endpointPushHandlers.get(endpointId);
            return null;
        }
    }

    /**
     * Returns the queue processor registered with the FlexClient with the supplied
     * endpoint id, or null if no queue processor was registered with the FlexClient
     * for that endpoint.
     *
     * @param endpointId The endpoint id.
     * @return The queue processor registered with the FlexClient.
     */
    public FlexClientOutboundQueueProcessor getOutboundQueueProcessor(String endpointId) {
        EndpointQueue queue = outboundQueues.get(endpointId);
        return (queue != null) ? queue.processor : null;
    }

    /**
     * Returns the endpoint queue registered with the FlexClient with the supplied
     * endpoint id, or null if no endpoint queue was registered with the FlexClient
     * for that endpoint.
     *
     * @param endpointId The endpoint id.
     * @return The endpoint queue registered with the FlexClient.
     */
    public EndpointQueue getEndpointQueue(String endpointId) {
        return outboundQueues.get(endpointId);
    }

    /**
     * Override {@link flex.messaging.util.TimeoutAbstractObject#getLastUse()} to make timeout
     * dependent upon FlexClient inactivity but also upon the presence of an active push-enabled session,
     * async or waited poll, or registered endpoint push handler (all of which indicate that a client has
     * an active, open connection to the server).
     *
     * @return The 'last use' timestamp for the FlexClient, which may be the current system time if the FlexClient
     * has been idle but an open connection from the client to the server exists.
     */
    @Override
    public long getLastUse() {
        synchronized (lock) {
            long currentLastUse = super.getLastUse();
            long idleTime = System.currentTimeMillis() - currentLastUse;
            if (idleTime < flexClientManager.getFlexClientTimeoutMillis())
                return currentLastUse; // Not timed out; this will trigger the timeout to be rescheduled.

            // Check for async long-polls or endpoint streaming connections, if found, keep alive.
            if (!outboundQueues.isEmpty()) {
                for (EndpointQueue queue : outboundQueues.values()) {
                    if (queue.asyncPoll != null)
                        return System.currentTimeMillis();

                    if (endpointPushHandlers != null && endpointPushHandlers.containsKey(queue.endpointId))
                        return System.currentTimeMillis();
                }
            }

            // Check for connected sessions, or a session holding a (non-async) long poll and if found, keep alive.
            for (FlexSession session : sessions) {
                if (session instanceof ConnectionAwareSession) {
                    if (((ConnectionAwareSession) session).isConnected())
                        return System.currentTimeMillis();
                }
                // Otherwise, check for a long-poll.
                if (session.waitMonitor != null) {
                    for (EndpointQueue queue : session.waitMonitor.values()) {
                        if (queue.flexClient.equals(this))
                            return System.currentTimeMillis();
                    }
                }
            }
            return currentLastUse; // Allow the FlexClient to timeout.
        }
    }

    /**
     * Returns the attribute bound to the specified name for the current FlexSession
     * associated with the FlexClient. If the attribute does not exist in the current
     * FlexSession, this method iterates through all the other FlexSessions associated with
     * the FlexClient and either returns the attribute bound, or null if no attribute is bound
     * under the name.
     *
     * @param name The name the attribute is bound to.
     * @return The attribute bound to the specified name.
     */
    public Object getSessionAttribute(String name) {
        Object attributeValue = getSessionAttributeInCurrentSession(name);
        if (attributeValue != null)
            return attributeValue;

        return getSessionAttributeInOtherSessions(name);
    }

    /**
     * Returns a snapshot of the names of all attributes bound to all the FlexSessions
     * associated with the FlexClient.
     *
     * @return A snapshot of the names of all attributes bound to all the FlexSessions
     * associated with the FlexClient.
     */
    public Enumeration<String> getSessionAttributeNames() {
        Set<String> attributeNames = new HashSet<String>();
        for (FlexSession session : sessions)
            attributeNames.addAll(getSessionAttributeNames(session));
        return Collections.enumeration(attributeNames);
    }

    /**
     * Returns the principal associated with this client.  If the client has not
     * authenticated the principal will be null.  Should only be called from FlexContext
     * and only if perClientAuthentication is used.  Not available to users.
     *
     * @return The principal associated with the session.
     */
    public Principal getUserPrincipal() {
        synchronized (lock) {
            checkValid();
            return userPrincipal;
        }
    }

    /**
     * Should only be called from FlexContext and only if perClientAuthentication is used.
     * Not available to users.
     *
     * @param userPrincipal The principal to associate with the session.
     */
    public void setUserPrincipal(Principal userPrincipal) {
        synchronized (lock) {
            checkValid();
            this.userPrincipal = userPrincipal;
        }
    }

    /**
     * Invalidates the FlexClient.
     */
    public void invalidate() {
        synchronized (lock) {
            if (!valid || invalidating)
                return; // Already shutting down.

            invalidating = true; // This thread gets to shut the FlexClient down.
            flexClientManager.removeFlexClient(this);
            cancelTimeout();
        }

        // Unregister from all FlexSessions.
        if (!sessions.isEmpty()) {
            for (FlexSession session : sessions)
                unregisterFlexSession(session);
        }

        // Invalidate associated MessageClient subscriptions.
        if (messageClients != null && !messageClients.isEmpty()) {
            for (MessageClient messageClient : messageClients) {
                messageClient.removeMessageClientDestroyedListener(this);
                messageClient.invalidate();
            }
            messageClients.clear();
        }

        // Notify destroy listeners that we're shutting the FlexClient down.
        if (destroyedListeners != null && !destroyedListeners.isEmpty()) {
            for (FlexClientListener destroyListener : destroyedListeners) {
                destroyListener.clientDestroyed(this);
            }
            destroyedListeners.clear();
        }

        // Unbind all attributes.
        if (attributes != null && !attributes.isEmpty()) {
            Set<String> keySet = attributes.keySet();
            String[] keys = keySet.toArray(new String[keySet.size()]);
            for (String key : keys)
                removeAttribute(key);
        }

        // Close any registered push handlers.
        if (endpointPushHandlers != null && !endpointPushHandlers.isEmpty()) {
            for (EndpointPushHandler handler : endpointPushHandlers.values()) {
                handler.close(true /* notify Channel of disconnect */);
            }
            endpointPushHandlers = null;
        }

        synchronized (lock) {
            valid = false;
            invalidating = false;
        }

        if (Log.isDebug())
            Log.getLogger(FLEX_CLIENT_LOG_CATEGORY).debug("FlexClient with id '" + this.id + "' has been invalidated.");
    }

    /**
     * Returns true if the FlexClient is valid; false if it has been invalidated.
     *
     * @return true if the FlexClient is valid; otherwise false.
     */
    public boolean isValid() {
        synchronized (lock) {
            return valid;
        }
    }

    /**
     * Returns a snapshot of the FlexSessions associated with the FlexClient
     * when this method is invoked.
     * This list is not guaranteed to remain consistent with the actual list
     * of active FlexSessions associated with the FlexClient over time.
     *
     * @return A snapshot of the current list of FlexSessions associated with the FlexClient.
     */
    public List<FlexSession> getFlexSessions() {
        List<FlexSession> currentSessions;
        synchronized (lock) {
            checkValid();

            updateLastUse();

            currentSessions = new ArrayList<FlexSession>(sessions); // Make a copy of the current list to return.
        }
        return currentSessions;
    }

    /**
     * Return the session count.
     *
     * @return The number of sessions associated with this FlexClient.
     */
    public int getSessionCount() {
        int sessionCount;
        synchronized (lock) {
            sessionCount = (sessions != null) ? sessions.size() : 0; // Make a copy of the current list to return.
        }
        return sessionCount;
    }

    /**
     * Return the subscription count.
     *
     * @return The number of subscriptions associated with this FlexClient.
     */
    public int getSubscriptionCount() {
        int count = 0;
        synchronized (lock) {

            if (messageClients != null && !messageClients.isEmpty()) {
                for (MessageClient messageClient : messageClients)
                    count += messageClient.getSubscriptionCount();
            }

        }
        return count;
    }

    /**
     * Returns the message client registered with the FlexClient with the supplied
     * client id, or null if no message client was registered with the FlexClient
     * with that client id.
     *
     * @param clientId The client id.
     * @return The message client registered with the FlexClient.
     */
    public MessageClient getMessageClient(String clientId) {
        synchronized (lock) {
            if (messageClients != null && !messageClients.isEmpty()) {
                for (MessageClient messageClient : messageClients) {
                    if (messageClient.getClientId().equals(clientId))
                        return messageClient;
                }
            }
        }
        return null;
    }

    /**
     * Returns a snapshot of the MessageClients (subscriptions) associated with the FlexClient
     * when this method is invoked.
     * This list is not guaranteed to remain consistent with the actual list
     * of active MessageClients associated with the FlexClient over time.
     *
     * @return A snapshot of the current list of MessageClients associated with the FlexClient.
     */
    public List<MessageClient> getMessageClients() {
        List<MessageClient> currentMessageClients;
        synchronized (lock) {
            checkValid();

            updateLastUse();

            currentMessageClients = (messageClients != null) ? new ArrayList<MessageClient>(messageClients) // Make a copy of the current list to return.
                    : Collections.<MessageClient>emptyList(); // Return an empty list.
        }
        return currentMessageClients;
    }

    /**
     * Returns the unique Id for the FlexClient.
     *
     * @return The unique Id for the FlexClient.
     */
    public String getId() {
        return id;
    }

    /**
     * Implements TimeoutCapable.
     * Determine the time, in milliseconds, that this object is allowed to idle
     * before having its timeout method invoked.
     */
    @Override
    public long getTimeoutPeriod() {
        return flexClientManager.getFlexClientTimeoutMillis();
    }

    /**
     * Implements MessageClientListener.
     * Handling created events is a no-op.
     *
     * @param messageClient The new MessageClient.
     */
    public void messageClientCreated(MessageClient messageClient) {
    }

    /**
     * Implements MessageClientListener.
     * Notification that an associated FlexSession was destroyed.
     *
     * @param messageClient The MessageClient that was destroyed.
     */
    public void messageClientDestroyed(MessageClient messageClient) {
        unregisterMessageClient(messageClient);
    }

    /**
     * Poll for outbound messages for the FlexClient.
     * This method is only invoked by internal code while processing a client poll request; it
     * is not intended for general public use.
     * Poll requests that trigger this method come from client-side polling channels and the request
     * is not specific to a single Consumer/MessageClient instance so process any queued messages for
     * the specified endpoint across all subscriptions.
     *
     * @param endpointId The Id of the endpoint that received the poll request.
     * @return The flush result including messages to return in the poll response and
     * an optional wait time for the next poll/flush.
     */
    public FlushResult poll(String endpointId) {
        EndpointQueue queue = null;
        synchronized (lock) {
            checkValid();

            queue = outboundQueues.get(endpointId);

            if (queue != null)
                return internalPoll(queue);
        }

        if (queue == null) {
            // Otherwise, the client is not subscribed.
            throwNotSubscribedException(endpointId);
        }

        return null;
    }

    /**
     * Poll for outbound messages for the FlexClient and if no messages are available
     * immediately, store a reference to the passed async handler and call back when messages arrive.
     *
     * @param endpointId         The Id of the endpoint that received the poll request.
     * @param handler            The handler to callback when messages arrive.
     * @param waitIntervalMillis The wait interval in milliseconds for the poll to wait for data to arrive
     *                           before returning an empty poll response.
     * @return A <tt>TimeoutAbstractObject</tt> representing the asynchronous poll, or <code>null</code>
     * if the poll request was handled immediately because data was available to return.
     */
    public TimeoutAbstractObject pollAsync(String endpointId, AsyncPollHandler handler, long waitIntervalMillis) {
        EndpointQueue queue;
        TimeoutAbstractObject asyncPollTask = null;

        synchronized (lock) {
            checkValid();

            queue = outboundQueues.get(endpointId);

            // If the queue exists and is not empty, flush immediately.
            if (queue != null) {
                if (!queue.messages.isEmpty()) {
                    handler.asyncPollComplete(internalFlush(queue));
                } else // Set up an async long-poll.
                {
                    // Avoid monopolizing user agent connections.
                    FlexSession session = FlexContext.getFlexSession();
                    synchronized (session) {
                        if (session.asyncPollMap != null) {
                            AsyncPollWithTimeout parkedPoll = session.asyncPollMap.get(endpointId);
                            if (parkedPoll != null) {
                                // If the poll is from the same client for this endpoint, treat it as a no-op.
                                if (parkedPoll.getFlexClient().equals(this)) {
                                    PollFlushResult result = new PollFlushResult();
                                    result.setClientProcessingSuppressed(true);
                                    handler.asyncPollComplete(result);
                                } else // If the poll is for a different client on the same session, swap their waits.
                                {
                                    PollFlushResult result = new PollFlushResult();
                                    result.setAvoidBusyPolling(true);
                                    completeAsyncPoll(parkedPoll, result);
                                }
                            }
                        }
                        AsyncPollWithTimeout asyncPoll = new AsyncPollWithTimeout(this, session, queue, handler, waitIntervalMillis, endpointId);
                        synchronized (session) {
                            if (session.asyncPollMap == null)
                                session.asyncPollMap = new HashMap<String, AsyncPollWithTimeout>();
                            session.asyncPollMap.put(endpointId, asyncPoll);
                        }
                        queue.asyncPoll = asyncPoll;
                        asyncPollTask = asyncPoll;
                    }
                }
            }
        }
        if (queue == null) {
            // The queue was null; let the client know that there are no active subscriptions.
            throwNotSubscribedException(endpointId);
        }
        return asyncPollTask;
    }

    /**
     * Poll for outbound messages for the FlexClient and if no messages are available
     * immediately, put processing into a wait state until messages arrive.
     * This method is only invoked by internal code while processing a client poll request; it
     * is not intended for general public use.
     * Poll requests that trigger this method come from client-side polling channels and the request
     * is not specific to a single Consumer/MessageClient instance so process any queued messages for
     * the specified endpoint across all subscriptions.
     *
     * @param endpointId         The Id of the endpoint that received the poll request.
     * @param session            The FlexSession associated with this waitable poll request.
     * @param listener           The listener to notify before a wait begins and as soon as one completes.
     * @param waitIntervalMillis The maximum amount of time to wait for messages in milliseconds.
     * @return The flush result including messages to return in the poll response and
     * an optional wait time for the next poll/flush.
     */
    public FlushResult pollWithWait(String endpointId, FlexSession session, PollWaitListener listener, long waitIntervalMillis) {
        EndpointQueue queue;
        synchronized (lock) {
            checkValid();

            queue = outboundQueues.get(endpointId);

            // If the queue exists and is not empty there's no reason to wait; flush immediately.
            if (queue != null) {
                FlushResult flushResult = internalPoll(queue);
                if (flushResult != null)
                    return flushResult;
            }
        }

        // The queue exists but it was empty; we can try to wait for messages.
        if (queue != null) {
            synchronized (session) {
                // Set up the waitMonitor on the session; this is a reference to the queue that the
                // current poll request targets and we use it as a wait/notify monitor.
                // This also lets us prevent busy polling cycles from a single client. If we already have a waited
                // poll request a subsequent poll request is treated as a no-op.
                if (session.waitMonitor != null) {
                    final EndpointQueue waitingQueue = session.waitMonitor.get(endpointId);
                    // If the poll is from the same client swf, and the same endpoint, treat it as a no-op poll.
                    if (waitingQueue != null && waitingQueue.flexClient.equals(this)) {
                        PollFlushResult result = new PollFlushResult();
                        result.setClientProcessingSuppressed(true);
                        return result;
                    }
                } else {
                    session.waitMonitor = new HashMap<String, EndpointQueue>();
                }

                // Set the waitMonitor for the session to the queue
                // for this poll request before releasing the lock.
                session.waitMonitor.put(endpointId, queue);
            }

            // Now that the session references the wait monitor this thread will use to wait we can enter
            // the wait state.
            // -1 wait-interval actually means wait until notified.
            waitIntervalMillis = (waitIntervalMillis == -1) ? 0 : waitIntervalMillis;
            String threadName = Thread.currentThread().getName();
            try {
                boolean didWait = false;
                boolean avoidBusyPolling = false;
                synchronized (queue) {
                    // If the message queue is still empty, wait for a message to be added before invoking flush.
                    if (queue.messages.isEmpty()) {
                        reportStatusIfDebug("waiting for new messages to arrive");

                        didWait = true;

                        // Tag thread name during the wait.
                        Thread currentThread = Thread.currentThread();
                        currentThread.setName(threadName + POLL_WAIT_THREAD_NAME_EXTENSION);

                        if (listener != null)
                            listener.waitStart(queue);

                        queue.waitPoll = true; // Mark the queue as waiting.

                        queue.wait(waitIntervalMillis);

                        queue.waitPoll = false; // Unmark the queue as waiting.

                        // Reset thread name now that the wait is over.
                        currentThread.setName(threadName);

                        if (listener != null)
                            listener.waitEnd(queue);

                        if (queue.avoidBusyPolling) {
                            avoidBusyPolling = true;
                            queue.avoidBusyPolling = false;
                        }
                    }
                }

                synchronized (session) {
                    if (session.waitMonitor != null) {
                        session.waitMonitor.remove(endpointId);
                    }
                }

                if (Log.isDebug()) {
                    if (didWait)
                        reportStatusIfDebug("done waiting for new messages to arrive and is flushing the outbound queue");
                    else
                        reportStatusIfDebug("didn't need to wait and is flushing the outbound queue");
                }

                // We need to hold the FlexClient lock to invoke flush.
                FlushResult result;
                synchronized (lock) {
                    result = internalFlush(queue);
                }
                if (avoidBusyPolling) {
                    PollFlushResult swappedPollResult = new PollFlushResult();
                    if (result != null) {
                        swappedPollResult.setMessages(result.getMessages());
                        swappedPollResult.setNextFlushWaitTimeMillis(result.getNextFlushWaitTimeMillis());
                    }
                    swappedPollResult.setAvoidBusyPolling(true);
                    result = swappedPollResult;
                }
                return result;
            } catch (InterruptedException e) {
                if (Log.isWarn())
                    Log.getLogger(FLEX_CLIENT_LOG_CATEGORY).warn("Poll wait thread '" + threadName + "' for FlexClient with id '" + this.id +
                            "' could not finish waiting for new messages to arrive " +
                            "because it was interrupted: " + e.toString());
            }
        } else {
            // The queue was null; let the client know that there are no active subscriptions.
            throwNotSubscribedException(endpointId);
        }
        return null;
    }

    private void reportStatusIfDebug(String message) {
        String threadName = Thread.currentThread().getName();
        if (Log.isDebug())
            Log.getLogger(FLEX_CLIENT_LOG_CATEGORY).debug("Poll wait thread '" + threadName + "' for FlexClient with id '" + this.id + "' is " + message);
    }

    /**
     * Poll for outbound messages for a specific MessageClient/Consumer.
     * This overload of poll() is only invoked when handling a Consumer.receive() request.
     *
     * @param client The specific MessageClient instance to poll for messages for.
     * @return The flush result including messages to return in the poll response.
     * The nextFlushWaitTimeMillis value is always forced to a value of 0 because
     * Consumer.receive() calls are driven by client code and this setting has no meaning.
     */
    public FlushResult poll(MessageClient client) {
        FlushResult flushResult = null;
        String endpointId = client.getEndpointId();
        EndpointQueue queue = null;
        synchronized (lock) {
            checkValid();

            queue = outboundQueues.get(endpointId);
            if (queue != null) {
                try {
                    flushResult = internalFlush(queue, client);
                } catch (RuntimeException e) {
                    if (Log.isError())
                        Log.getLogger(FLEX_CLIENT_LOG_CATEGORY).error("Failed to flush an outbound queue for MessageClient '" + client.getClientId() + "' for FlexClient '" + getId() + "'.", e);
                    throw e;
                }
                if (flushResult != null)
                    flushResult.setNextFlushWaitTimeMillis(0); // Force to 0.
            }
        }
        if (queue == null) {
            throwNotSubscribedException(endpointId);
        }
        return flushResult;
    }

    /**
     * Push a message to the FlexClient.
     * The message is added to the outbound queue of messages for the client and
     * will be pushed if possible or retrieved via a client poll request.
     *
     * @param message       The Message to push.
     * @param messageClient The MessageClient subscription that this message targets.
     */
    public void push(Message message, MessageClient messageClient) {
        // We should check the message client is valid or not
        if (!isValid())
            return;

        // Route this message to the proper per-endpoint outbound queue.
        EndpointQueue queue = outboundQueues.get(messageClient.getEndpointId());

        // This queue may be null if all corresponding subscriptions have been invalidated.
        if (queue == null)
            return;

        boolean empty;

        // We need to obtain the lock here
        // Maintain the pattern of using the FlexClient.lock and ensure that order of locks should always start with the FlexClient.lock
        // This is critical to prevent deadlock cases, see Watson bug 2724938 
        synchronized (lock) {
            synchronized (queue) // To protect the list during the add and allow for notification.
            {
                // Let the processor add the message to the queue.
                try {
                    queue.processor.add(queue.messages, message);
                    empty = queue.messages.isEmpty();

                    if (Log.isDebug())
                        Log.getLogger(LogCategories.MESSAGE_GENERAL).debug(
                                "Queuing message: " + message.getMessageId() +
                                        StringUtils.NEWLINE +
                                        "  to send to MessageClient: " + messageClient.getClientId() +
                                        StringUtils.NEWLINE +
                                        "  for FlexClient: " + messageClient.getFlexClient().getId() +
                                        StringUtils.NEWLINE +
                                        "  via endpoint: " + queue.endpointId +
                                        StringUtils.NEWLINE +
                                        "  client outbound queue size: " + queue.messages.size());
                } catch (RuntimeException e) {
                    if (Log.isError())
                        Log.getLogger(FLEX_CLIENT_LOG_CATEGORY).error("Failed to add a message to an outbound queue for FlexClient '" + getId() + "'.", e);
                    throw e;
                }
                // And notify any threads that may be in a poll wait state.
                if (!empty && queue.waitPoll) {
                    // TODO This updateLastUse call is added here because there used to be a call 
                    // at the beginning of the push method but not convinced that it is needed. 
                    updateLastUse();
                    queue.notifyAll();
                }
            }

            if (!empty) {
                if (queue.asyncPoll != null) {
                    completeAsyncPoll(queue.asyncPoll, internalFlush(queue));
                } else if (!empty && queue.flushTask == null &&
                        (queue.pushSession != null || (endpointPushHandlers != null && endpointPushHandlers.containsKey(queue.endpointId)))) {
                    // If a delayed flush is not scheduled and we have a push-enabled session associated with the queue
                    // or a push-enabled endpoint, try a direct push to the client.
                    // Once again we should acquire the lock for queue, otherwise a potential dead lock could happen, see Watson bug 2724936
                    // By acquiring the queue lock again, we break the cycle by acquiring the queue before holding FlexClient.lock object 
                    synchronized (queue) {
                        directFlush(queue);
                    }
                }
            }
        }
    }

    /**
     * Registers an <tt>EndpointPushHandler</tt> for the specified endpoint to handle pushing messages
     * to remote clients.
     *
     * @param handler    The <tt>EndpointPushHandler</tt> to register.
     * @param endpointId The endpoint to register for.
     */
    public void registerEndpointPushHandler(EndpointPushHandler handler, String endpointId) {
        synchronized (lock) {
            if (endpointPushHandlers == null)
                endpointPushHandlers = new HashMap<String, EndpointPushHandler>(1);

            if (endpointPushHandlers.containsKey(endpointId)) {
                MessageException me = new MessageException();
                me.setMessage(ENDPOINT_PUSH_HANDLER_ALREADY_REGISTERED, new Object[]{getId(), endpointId});
                throw me;
            }

            endpointPushHandlers.put(endpointId, handler);
        }
    }

    /**
     * Used internally to associate a FlexSession with this FlexClient.
     *
     * @param session The FlexSession to associate with this FlexClient.
     */
    public void registerFlexSession(FlexSession session) {
        if (sessions.addIfAbsent(session)) {
            session.addSessionDestroyedListener(this);
            session.registerFlexClient(this);
        }
    }

    /**
     * Used internally to associate a MessageClient with this FlexClient.
     *
     * @param messageClient The MessageClient to associate with this FlexClient.
     */
    public void registerMessageClient(MessageClient messageClient) {
        synchronized (lock) {
            if (messageClients == null)
                messageClients = new CopyOnWriteArrayList<MessageClient>();
        }

        if (messageClients.addIfAbsent(messageClient)) {
            messageClient.addMessageClientDestroyedListener(this);
            String endpointId = messageClient.getEndpointId();
            // Manage the outbound queue this MessageClient's subscription(s) will use
            // and associate the MessageClient with an EndpointPushHandler if one exists for the
            // endpoint the subscription was made over; this allows the shut-down of a
            // push connection to invalidate any subscriptions that are using it.
            synchronized (lock) {
                getOrCreateEndpointQueueAndRegisterSubscription(messageClient, endpointId);
                if (endpointPushHandlers != null) {
                    EndpointPushHandler handler = endpointPushHandlers.get(endpointId);
                    if (handler != null)
                        handler.registerMessageClient(messageClient);
                }
            }
        }
    }

    /**
     * Removes the attribute bound to the specified name for the FlexClient.
     *
     * @param name The name of the attribute to remove.
     */
    public void removeAttribute(String name) {
        Object value; // Used for event dispatch after the attribute is removed.

        synchronized (lock) {
            checkValid();

            updateLastUse();

            value = (attributes != null) ? attributes.remove(name) : null;
        }

        // If no value was bound under this name it's a no-op.
        if (value == null)
            return;

        notifyAttributeUnbound(name, value);
        notifyAttributeRemoved(name, value);
    }

    /**
     * Removes a FlexClient attribute listener.
     *
     * @param listener The listener to remove.
     */
    public void removeClientAttributeListener(FlexClientAttributeListener listener) {
        // No need to check validity; removing a listener is always ok.
        if (listener != null && attributeListeners != null)
            attributeListeners.remove(listener);
    }

    /**
     * Removes a FlexClient destroyed listener.
     *
     * @param listener The listener to remove.
     * @see flex.messaging.client.FlexClientListener
     */
    public void removeClientDestroyedListener(FlexClientListener listener) {
        // No need to check validity; removing a listener is always ok.
        if (listener != null && destroyedListeners != null)
            destroyedListeners.remove(listener);
    }

    /**
     * Removes the attribute bound to the specified name for all the FlexSessions
     * associated with the FlexClient.
     *
     * @param name The name of the attribute to remove.
     */
    public void removeSessionAttribute(String name) {
        for (FlexSession session : sessions)
            session.removeAttribute(name);
    }

    /**
     * Implements FlexSessionListener interface.
     * Notification that a FlexSession was created.
     * This is a no-op because the FlexClient is never added as a static FlexSession created listener
     * but this method is required by the interface. We only listen for the destroyed event from
     * associated FlexSessions.
     *
     * @param session The FlexSession that was created.
     */
    public void sessionCreated(FlexSession session) {
    }

    /**
     * Implements FlexSessionListener interface.
     * Notification that an associated FlexSession was destroyed.
     *
     * @param session The FlexSession that was destroyed.
     */
    public void sessionDestroyed(FlexSession session) {
        unregisterFlexSession(session);
    }

    /**
     * Binds an attribute value for the FlexClient under the specified name.
     *
     * @param name  The name to bind the attribute under.
     * @param value The value of the attribute.
     */
    public void setAttribute(String name, Object value) {
        // Null value set is the same as removeAttribute().
        if (value == null) {
            removeAttribute(name);
            return;
        }

        Object oldValue; // Used to determine which events to dispatch after the set is performed.

        // Only synchronize for the attribute mutation; event dispatch doesn't require it.
        synchronized (lock) {
            checkValid();

            updateLastUse();

            if (attributes == null)
                attributes = new HashMap<String, Object>();

            oldValue = attributes.put(name, value);
        }

        if (oldValue == null) {
            notifyAttributeBound(name, value);
            notifyAttributeAdded(name, value);
        } else {
            notifyAttributeUnbound(name, oldValue);
            notifyAttributeReplaced(name, oldValue);
            notifyAttributeBound(name, value);
        }
    }

    /**
     * Binds an attribute value for the current FlexSession associated with the
     * FlexClient under the specified name. If the current FlexSession is NIO-based
     * (NIOHTTPFlexSession or RTMPFlexSession), and if the FlexClient is associated
     * with a Servlet-based session (HttpFlexSession) as well, the attribute is bound
     * on the Servlet-based session too to make it available to the underlying J2EE HttpSession.
     *
     * @param name  The name to bind the attribute under.
     * @param value The value of the attribute.
     */
    public void setSessionAttribute(String name, Object value) {
        setSessionAttributeInCurrentSession(name, value);
        if (!isCurrentSessionServletBased())
            setSessionAttributeInServletBasedSession(name, value);
    }

    /**
     * Implements TimeoutCapable.
     * Inform the object that it has timed out.
     */
    public void timeout() {
        invalidate();
    }

    /**
     * Unregisters an <tt>EndpointPushHandler</tt> from the specified endpoint.
     *
     * @param handler    The <tt>EndpointPushHandler</tt> to unregister.
     * @param endpointId The endpoint to unregister from.
     */
    public void unregisterEndpointPushHandler(EndpointPushHandler handler, String endpointId) {
        synchronized (lock) {
            if (endpointPushHandlers == null)
                return; // No-op.

            if (endpointPushHandlers.get(endpointId).equals(handler))
                endpointPushHandlers.remove(endpointId);
        }
    }

    /**
     * Used internally to disassociate a FlexSession from this FlexClient.
     *
     * @param session The FlexSession to disassociate from this FlexClient.
     */
    public void unregisterFlexSession(FlexSession session) {
        if (sessions.remove(session)) {
            session.removeSessionDestroyedListener(this);
            session.unregisterFlexClient(this);
            // Once all client sessions/connections terminate; shut down.
            if (sessions.isEmpty())
                invalidate();
        }
    }

    /**
     * Used internally to disassociate a MessageClient (subscription) from a FlexClient.
     *
     * @param messageClient The MessageClient to disassociate from the FlexClient.
     */
    public void unregisterMessageClient(MessageClient messageClient) {
        if (messageClients != null && messageClients.remove(messageClient)) {
            messageClient.removeMessageClientDestroyedListener(this);
            String endpointId = messageClient.getEndpointId();
            // Manage the outbound queue that this subscription uses.
            synchronized (lock) {
                EndpointQueue queue = outboundQueues.get(endpointId);
                if (queue != null) {
                    // Decrement the ref count of MessageClients using this queue.
                    queue.messageClientRefCount--;

                    // Unregister the message client from the outbound throttle
                    // manager (if one exists).
                    OutboundQueueThrottleManager tm = queue.processor.getOutboundQueueThrottleManager();
                    if (tm != null)
                        tm.unregisterAllSubscriptions(messageClient.getDestinationId());

                    // If we're not attempting to notify the remote client that this MessageClient has
                    // been invalidated, remove any associated messages from the queue.
                    if (!messageClient.isAttemptingInvalidationClientNotification()) {
                        Object messageClientId = messageClient.getClientId();
                        for (Iterator<Message> iter = queue.messages.iterator(); iter.hasNext(); ) {
                            Message message = iter.next();
                            if (message.getClientId().equals(messageClientId))
                                iter.remove();
                        }
                    }

                    // If no active subscriptions require the queue, clean it up if possible.
                    if (queue.messageClientRefCount == 0) {
                        if (queue.messages.isEmpty() || messageClient.isClientChannelDisconnected()) {
                            if (queue.asyncPoll != null) // Close out async long-poll if one is registered.
                            {
                                FlushResult flushResult = internalFlush(queue);
                                // If the MessageClient isn't attempting client notification, override
                                // and do so in this case to suppress the next poll request from the remote client
                                // which will fail triggering an unnecessary channel disconnect on the client.
                                if (!messageClient.isAttemptingInvalidationClientNotification()) {
                                    CommandMessage msg = new CommandMessage();
                                    msg.setClientId(messageClient.getClientId());
                                    msg.setOperation(CommandMessage.SUBSCRIPTION_INVALIDATE_OPERATION);
                                    List<Message> messages = flushResult.getMessages();
                                    if (messages == null)
                                        messages = new ArrayList<Message>(1);
                                    messages.add(msg);
                                }
                                completeAsyncPoll(queue.asyncPoll, flushResult);
                            }

                            // Remove the empty, unused queue.
                            outboundQueues.remove(endpointId);
                        }
                        // Otherwise, the queue is being used by a polling client or contains messages
                        // that will be written by a delayed flush.
                        // Leave it in place. Once the next poll request or delayed flush occurs the
                        // queue will be cleaned up at that point. See internalFlush() and shutdownQueue().
                    }

                    // Make sure to notify any threads waiting on this queue that may be associated
                    // with the subscription that's gone away.
                    synchronized (queue) {
                        queue.notifyAll();
                    }
                }
                // And if this subscription was associated with an endpoint push handler, unregister it.
                if (endpointPushHandlers != null) {
                    EndpointPushHandler handler = endpointPushHandlers.get(endpointId);
                    if (handler != null)
                        handler.unregisterMessageClient(messageClient);
                }
            }
        }
    }

    //--------------------------------------------------------------------------
    //
    // Protected Methods
    //
    //--------------------------------------------------------------------------

    /**
     * Utility method that tests validity and throws an exception if the instance
     * has been invalidated.
     */
    protected void checkValid() {
        synchronized (lock) {
            if (!valid) {
                MessageException e = new MessageException();
                e.setMessage(FLEX_CLIENT_INVALIDATED);
                throw e;
            }
        }
    }

    /**
     * Invoked to clean up a timed out or closed async poll.
     *
     * @param asyncPoll The async poll to complete.
     * @param result    The FlushResult for the poll response.
     */
    protected void completeAsyncPoll(AsyncPollWithTimeout asyncPoll, FlushResult result) {
        synchronized (lock) {
            asyncPoll.cancelTimeout();
            EndpointQueue queue = asyncPoll.getEndpointQueue();
            if (queue.asyncPoll.equals(asyncPoll))
                queue.asyncPoll = null;
            FlexSession session = asyncPoll.getFlexSession();
            synchronized (session) {
                if (session.asyncPollMap != null)
                    session.asyncPollMap.remove(asyncPoll.getEndpointId());
            }
            asyncPoll.getHandler().asyncPollComplete(result);
        }
    }

    /**
     * Invoked to flush queued outbound messages to a client directly using a session
     * that supports real-time push.
     * Called by push() or delayed flush tasks for push-enabled sessions/connections.
     */
    protected void directFlush(EndpointQueue queue) {
        synchronized (lock) {
            // No need to invoke flush if the FlexClient has been invalidated.
            if (!valid)
                return;

            // If this invocation is a callback from a flush task, null out the task ref on
            // the queue to allow a subsequent delayed flush to be scheduled.
            if (queue.flushTask != null)
                queue.flushTask = null;

            FlushResult flushResult = internalFlush(queue, null, false /* updateLastUse */);
            if (flushResult == null) // If there's no flush result, return.
                return;

            // Pass any messages that are ready to flush off to the network layer.
            List<Message> messages = flushResult.getMessages();
            if (messages != null && !messages.isEmpty()) {
                if (queue.pushSession != null) {
                    if (queue.pushSession instanceof ConnectionAwareSession) {
                        // Update last use only if we're actually writing back to the client.
                        if (((ConnectionAwareSession) queue.pushSession).isConnected())
                            updateLastUse();
                    }
                    for (Message msg : messages)
                        queue.pushSession.push(msg);
                } else if (endpointPushHandlers != null) {
                    updateLastUse();
                    EndpointPushHandler handler = endpointPushHandlers.get(queue.endpointId);
                    handler.pushMessages(messages);
                }
            }

            // Schedule a delayed flush if necessary.
            int flushWaitTime = flushResult.getNextFlushWaitTimeMillis();
            if (flushWaitTime > 0) // Set up and schedule the delayed flush task.
                queue.flushTask = new FlexClientScheduledFlushForPush(queue, flushWaitTime);
        }
    }

    /**
     * Utility method to initialize an EndpointQueue (if necessary) and associate a subscription (MessageClient) with it.
     */
    protected EndpointQueue getOrCreateEndpointQueueAndRegisterSubscription(MessageClient messageClient, String endpointId) {
        EndpointQueue newQueue;
        if (!outboundQueues.containsKey(endpointId)) {
            newQueue = new EndpointQueue();
            newQueue.flexClient = this;
            newQueue.endpointId = endpointId;
            newQueue.endpoint = flexClientManager.getMessageBroker().getEndpoint(endpointId);
            newQueue.messages = new ArrayList<Message>(); /* Default size of 10 is fine */
            FlexSession session = messageClient.getFlexSession();
            if (session.isPushSupported())
                newQueue.pushSession = session;
            newQueue.processor = flexClientManager.createOutboundQueueProcessor(this, endpointId);
            newQueue.messageClientRefCount = 1;

            outboundQueues.put(endpointId, newQueue);
        } else {
            newQueue = outboundQueues.get(endpointId);
            newQueue.messageClientRefCount++;
            // Resubscribes as a result of network connectivity issues may arrive over the same
            // endpoint but use a new session.
            FlexSession session = messageClient.getFlexSession();
            if (session.isPushSupported())
                newQueue.pushSession = session;
        }
        return newQueue;
    }

    /**
     * Utility method to flush the outbound queue and log any problems.
     * Any exceptions are logged and then rethrown.
     *
     * @param queue The outbound queue to flush.
     */
    protected FlushResult internalFlush(EndpointQueue queue) {
        return internalFlush(queue, null);
    }

    /**
     * Utility method to flush the outbound queue and log any problems.
     * If a specific client is passed, we need to invoke a client-specific flush.
     * If the passed client is null, we do a general flush of the queue.
     * Any exceptions are logged and then rethrown.
     *
     * @param queue  The outbound queue to flush.
     * @param client The client to flush for.
     */
    protected FlushResult internalFlush(EndpointQueue queue, MessageClient client) {
        return internalFlush(queue, null, true);
    }

    /**
     * Utility method to flush the outbound queue and log any problems.
     * If a specific client is passed, we need to invoke a client-specific flush.
     * If the passed client is null, we do a general flush of the queue.
     * Any exceptions are logged and then rethrown.
     *
     * @param queue         The outbound queue to flush.
     * @param client        The client to flush for.
     * @param updateLastUse Whether the last-use timestamp of the FlexClient should
     *                      be updated.
     */
    protected FlushResult internalFlush(EndpointQueue queue, MessageClient client,
                                        boolean updateLastUse) {
        FlushResult flushResult;
        try {
            synchronized (queue) {
                flushResult = queue.processor.flush(client, queue.messages);
                shutdownQueue(queue);
            }

            if (updateLastUse)
                updateLastUseIfNecessary(flushResult);
        } catch (RuntimeException e) {
            if (Log.isError())
                Log.getLogger(FLEX_CLIENT_LOG_CATEGORY).error("Failed to flush an outbound queue for FlexClient '" + getId() + "'.", e);
            throw e;
        }
        return flushResult;
    }

    /**
     * Utility method to flush messages in response to a poll request with
     * regular and wait poll.
     *
     * @param queue The endpoint queue to flush messages for.
     * @return The flush result with messages, or null if there are no messages.
     */
    protected FlushResult internalPoll(EndpointQueue queue) {
        List<Message> allMessages = new ArrayList<Message>();

        // First, add the previously flushed messages.
        if (queue.flushedMessagesBetweenPolls != null && queue.flushedMessagesBetweenPolls.size() > 0) {
            allMessages.addAll(queue.flushedMessagesBetweenPolls);
            queue.flushedMessagesBetweenPolls.clear();
        }

        // Then, check for regularly queued messages. We call internalFlush
        // even if the queue is empty so the queue processor could know
        // about the incoming poll request regardless.
        FlushResult internalFlushResult = internalFlush(queue);
        List<Message> flushedMessages = internalFlushResult.getMessages();
        if (flushedMessages != null && !flushedMessages.isEmpty())
            allMessages.addAll(flushedMessages);

        // Schedule a delayed flush, if necessary.
        int flushWaitTime = internalFlushResult.getNextFlushWaitTimeMillis();
        if (flushWaitTime > 0)
            queue.flushTask = new FlexClientScheduledFlushForPoll(queue, flushWaitTime);

        if (allMessages.size() > 0) // Flush, if there are messages.
        {
            FlushResult flushResult = new FlushResult();
            flushResult.setMessages(allMessages);
            return flushResult;
        }
        return null;
    }

    /**
     * Notify attribute listeners that an attribute has been added.
     *
     * @param name  The name of the attribute.
     * @param value The new value of the attribute.
     */
    protected void notifyAttributeAdded(String name, Object value) {
        if (attributeListeners != null && !attributeListeners.isEmpty()) {
            FlexClientBindingEvent event = new FlexClientBindingEvent(this, name, value);
            // CopyOnWriteArrayList is iteration-safe from ConcurrentModificationExceptions.
            for (FlexClientAttributeListener attribListener : attributeListeners)
                attribListener.attributeAdded(event);
        }
    }

    /**
     * Notify binding listener that it has been bound to the FlexClient.
     *
     * @param name  The attribute name.
     * @param value The attribute that has been bound.
     */
    protected void notifyAttributeBound(String name, Object value) {
        if ((value != null) && (value instanceof FlexClientBindingListener)) {
            FlexClientBindingEvent bindingEvent = new FlexClientBindingEvent(this, name);
            ((FlexClientBindingListener) value).valueBound(bindingEvent);
        }
    }

    /**
     * Notify attribute listeners that an attribute has been removed.
     *
     * @param name  The name of the attribute.
     * @param value The previous value of the attribute.
     */
    protected void notifyAttributeRemoved(String name, Object value) {
        if (attributeListeners != null && !attributeListeners.isEmpty()) {
            FlexClientBindingEvent event = new FlexClientBindingEvent(this, name, value);
            // CopyOnWriteArrayList is iteration-safe from ConcurrentModificationExceptions.
            for (FlexClientAttributeListener attribListener : attributeListeners)
                attribListener.attributeRemoved(event);
        }
    }

    /**
     * Notify attribute listeners that an attribute has been replaced.
     *
     * @param name  The name of the attribute.
     * @param value The previous value of the attribute.
     */
    protected void notifyAttributeReplaced(String name, Object value) {
        if (attributeListeners != null && !attributeListeners.isEmpty()) {
            FlexClientBindingEvent event = new FlexClientBindingEvent(this, name, value);
            // CopyOnWriteArrayList is iteration-safe from ConcurrentModificationExceptions.
            for (FlexClientAttributeListener attribListener : attributeListeners)
                attribListener.attributeReplaced(event);
        }
    }

    /**
     * Notify binding listener that it has been unbound from the FlexClient.
     *
     * @param name  The attribute name.
     * @param value The attribute that has been unbound.
     */
    protected void notifyAttributeUnbound(String name, Object value) {
        if ((value != null) && (value instanceof FlexClientBindingListener)) {
            FlexClientBindingEvent bindingEvent = new FlexClientBindingEvent(this, name);
            ((FlexClientBindingListener) value).valueUnbound(bindingEvent);
        }
    }

    /**
     * Invoked by FlexClientManager after this new FlexClient has been constructed and
     * is fully configured.
     */
    protected void notifyCreated() {
        if (!createdListeners.isEmpty()) {
            // CopyOnWriteArrayList is iteration-safe from ConcurrentModificationExceptions.
            for (FlexClientListener createListener : createdListeners)
                createListener.clientCreated(this);
        }
    }

    /**
     * Utility method used to shutdown endpoint queues accessed via polling channels
     * that have no more active subscriptions and no more pending outbound messages.
     *
     * @param queue The queue to potentially shutdown.
     * @return true if the queue was cleaned up/removed; otherwise false.
     */
    protected boolean shutdownQueue(EndpointQueue queue) {
        // If no more subscriptions are using the queue and it is empty, shut it down.
        if (queue.messageClientRefCount == 0 && queue.messages.isEmpty()) {
            outboundQueues.remove(queue.endpointId);
            // Notify any threads waiting on this queue.
            synchronized (queue) {
                queue.notifyAll();
            }
            return true;
        }
        return false;
    }

    /**
     * Utility method to throw a not subscribed exception back to the client
     * if they issue a poll request to an endpoint that they haven't subscribed over.
     * <p>
     * This method should not be called when you hold an internal thread lock. It iterates
     * over all the FlexClients in the current session and will not work if two or more
     * FlexClients in the same session call it simultaneously.
     *
     * @param endpointId The endpoint Id.
     */
    protected void throwNotSubscribedException(String endpointId) {
        // Pre-3.1 versions of the client library did not handle URL session tokens properly
        // and may incorrectly issue a poll, after subscribing, that does not contain the proper
        // FlexClient id.
        // This scenario looks like a poll from a client that is not subscribed, but it is not,
        // and deserves a more useful error message.
        // We determine this by checking for an (orphaned) FlexClient instance associated with the
        // current session that has a subscription established through the target endpoint.
        List<FlexClient> flexClients = FlexContext.getFlexSession().getFlexClients();
        for (FlexClient otherClient : flexClients) {
            if (!otherClient.equals(this)) {
                List<MessageClient> otherSubs = otherClient.getMessageClients();
                for (MessageClient otherSub : otherSubs) {
                    if (otherSub.getEndpointId().equals(endpointId)) {
                        // Throw not-subscribed exception with extra guidance.
                        FlexClientNotSubscribedException e = new FlexClientNotSubscribedException();
                        e.setMessage(10036, new Object[]{endpointId});
                        e.setCode(MessageService.NOT_SUBSCRIBED_CODE);
                        throw e;
                    }
                }
            }
        }

        // Throw general not-subscribed exception.
        FlexClientNotSubscribedException e = new FlexClientNotSubscribedException();
        e.setMessage(10028, new Object[]{endpointId});
        e.setCode(MessageService.NOT_SUBSCRIBED_CODE);
        throw e;
    }

    /**
     * Updates the last-use timestamp if there are messages in the flush result.
     *
     * @param flushResult The flush result.
     */
    protected void updateLastUseIfNecessary(FlushResult flushResult) {
        List<Message> messages = flushResult != null ? flushResult.getMessages() : null;
        if (messages != null && !messages.isEmpty())
            updateLastUse();
    }

    //--------------------------------------------------------------------------
    //
    // Private Methods
    //
    //--------------------------------------------------------------------------

    private Set<String> getSessionAttributeNames(FlexSession session) {
        Set<String> attributeNames = new HashSet<String>();
        Enumeration<String> currentAttributeNames = session.getAttributeNames();
        while (currentAttributeNames.hasMoreElements())
            attributeNames.add(currentAttributeNames.nextElement());
        return attributeNames;
    }

    private Object getSessionAttributeInCurrentSession(String name) {
        return FlexContext.getFlexSession().getAttribute(name);
    }

    private Object getSessionAttributeInOtherSessions(String name) {
        FlexSession currentSession = FlexContext.getFlexSession();
        for (FlexSession session : sessions) {
            if (session == currentSession)
                continue;

            Object attributeValue = session.getAttribute(name);
            if (attributeValue != null)
                return attributeValue;
        }
        return null;
    }

    private void setSessionAttributeInCurrentSession(String name, Object value) {
        FlexContext.getFlexSession().setAttribute(name, value);
    }

    private void setSessionAttributeInServletBasedSession(String name, Object value) {
        for (FlexSession session : sessions) {
            if (isServletBasedSession(session)) {
                session.setAttribute(name, value);
                return;
            }
        }
    }

    private boolean isCurrentSessionServletBased() {
        return isServletBasedSession(FlexContext.getFlexSession());
    }

    private boolean isServletBasedSession(FlexSession session) {
        return session instanceof HttpFlexSession;
    }

    //--------------------------------------------------------------------------
    //
    // Inner Classes
    //
    //--------------------------------------------------------------------------

    /**
     * Helper class for handling async poll requests. This class allows the response for an async poll
     * to be delayed until data arrives to return to the client or the specified wait interval elapses.
     * Wait timeouts are monitored by the <tt>FlexClientManager</tt> which contains a <tt>TimeoutManager</tt>
     * instance that is started and stopped during application bootstrap and shutdown. Managing timeouts
     * locally or statically isn't a good option because they lack a useful shutdown hook that's necessary
     * in order to close down the timeout manager cleanly.
     */
    public class AsyncPollWithTimeout extends TimeoutAbstractObject {
        /**
         * Constructor.
         *
         * @param flexClient         flex client
         * @param session            flex session
         * @param queue              endpoint queue
         * @param handler            poll handler
         * @param waitIntervalMillis wait interval
         * @param endpointId         endpoint
         */
        public AsyncPollWithTimeout(FlexClient flexClient, FlexSession session, EndpointQueue queue, AsyncPollHandler handler, long waitIntervalMillis, String endpointId) {
            this.flexClient = flexClient;
            this.session = session;
            this.queue = queue;
            this.handler = handler;
            setTimeoutPeriod(waitIntervalMillis);
            flexClientManager.monitorAsyncPollTimeout(this);
            this.endpointId = endpointId;
        }

        private final FlexClient flexClient;

        /**
         * Return client.
         *
         * @return flex client
         */
        public FlexClient getFlexClient() {
            return flexClient;
        }

        private final FlexSession session;

        /**
         * Return session.
         *
         * @return flex session
         */
        public FlexSession getFlexSession() {
            return session;
        }

        private final EndpointQueue queue;

        /**
         * Return endpoint queue.
         *
         * @return the queue
         */
        public EndpointQueue getEndpointQueue() {
            return queue;
        }

        private final AsyncPollHandler handler;

        /**
         * Return handler.
         *
         * @return the handler
         */
        public AsyncPollHandler getHandler() {
            return handler;
        }

        private final String endpointId;

        /**
         * Return endpoint ID.
         *
         * @return the id
         */
        public String getEndpointId() {
            return endpointId;
        }

        /**
         * Trigger a timeout.
         */
        public void timeout() {
            completeAsyncPoll(this, null /* nothing to return */);
        }
    }

    /**
     * Helper class to flush a FlexClient's outbound queue after a specified delay.
     * Delayed flushes are handled by the <tt>FlexClientManager</tt>
     * using <tt>TimeoutManager</tt>.
     */
    abstract class FlexClientScheduledFlush extends TimeoutAbstractObject {
        final EndpointQueue queue;

        public FlexClientScheduledFlush(EndpointQueue queue, long waitIntervalMillis) {
            this.queue = queue;
            setTimeoutPeriod(waitIntervalMillis);
            flexClientManager.monitorScheduledFlush(this);
        }

        abstract void performFlushTask();

        public void timeout() {
            FlexContext.setThreadLocalFlexClient(FlexClient.this);
            performFlushTask();
            FlexContext.setThreadLocalFlexClient(null);
        }
    }

    /**
     * Helper class for push channels to directly flush a FlexClient's outbound
     * queue after a specified delay.
     */
    class FlexClientScheduledFlushForPush extends FlexClientScheduledFlush {
        public FlexClientScheduledFlushForPush(EndpointQueue queue, long waitIntervalMillis) {
            super(queue, waitIntervalMillis);
        }

        @Override
        void performFlushTask() {
            synchronized (lock) {
                synchronized (queue) {
                    directFlush(queue);
                }
            }
        }

    }

    /**
     * Helper class for polling channels to flush a FlexClient's outbound
     * queue to flushedMessagesBetweenPolls queue after a specified delay.
     * When the next poll happens, the flushedMessagesBetweenPolls will be
     * drained first.
     */
    class FlexClientScheduledFlushForPoll extends FlexClientScheduledFlush {
        public FlexClientScheduledFlushForPoll(EndpointQueue queue, long waitIntervalMillis) {
            super(queue, waitIntervalMillis);
        }

        @Override
        void performFlushTask() {
            synchronized (lock) {
                // No need to invoke flush if the FlexClient has been invalidated.
                if (!valid)
                    return;

                // If this invocation is a callback from a flush task, null out the task ref on
                // the queue to allow a subsequent delayed flush to be scheduled.
                if (queue.flushTask != null)
                    queue.flushTask = null;

                FlushResult flushResult = internalFlush(queue);
                if (flushResult == null)
                    return;

                List<Message> messages = flushResult.getMessages();
                if (messages != null && messages.size() > 0) {
                    if (queue.asyncPoll != null) {
                        completeAsyncPoll(queue.asyncPoll, flushResult);
                    } else {
                        if (queue.flushedMessagesBetweenPolls == null)
                            queue.flushedMessagesBetweenPolls = new ArrayList<Message>();
                        queue.flushedMessagesBetweenPolls.addAll(messages);
                    }
                }

                // Schedule a delayed flush, if necessary.
                int flushWaitTime = flushResult.getNextFlushWaitTimeMillis();
                if (flushWaitTime > 0)
                    queue.flushTask = new FlexClientScheduledFlushForPoll(queue, flushWaitTime);
            }
        }
    }

    /**
     * Helper class that stores per-endpoint outbound queue state including:
     * <ul>
     *   <li>flexClient - The <tt>FlexClient</tt> the queue is used by.</li>
     *   <li>messages - The outbound queue of messages for the endpoint.</li>
     *   <li>flushedMessagesBetweenPolls - Keeps track of flushed (more precisely
     *       drained buffered) messages between polls. A seperate list is needed
     *       from messages list to avoid regular flush handling.</li>
     *   <li>flushedMessagesBetweenPolls - Keeps track of flushed messages between polls.</li>
     *   <li>processor - The processor that handles adding messages to the queue as well as flushing
     *       them to the network.</li>
     *   <li>asyncPoll - The async poll to timeout or callback when messages arrive
     *       (null if the endpoint or session supports direct push).</li>
     *   <li>pushSession - A reference to a pushSession to use for direct writes to the
     *       client (null if the endpoint uses polling or handles push directly).</li>
     *
     *   <li>flushTask - A reference to a pending flush task that will perform a delayed flush of the queue;
     *       null if no delayed flush has been scheduled.</li>
     *   <li>messageClientRefCount - A reference count of MessageClients subcribed over this endpoint.
     *       Once all MessageClients unsubscribe this queue can be shut down.</li>
     *   <li>avoidBusyPolling - Used to signal poll result generation for the queue to avoid busy polling.</li>
     * </ul>
     */
    public static class EndpointQueue {
        public FlexClient flexClient;
        public String endpointId;
        public Endpoint endpoint;
        public List<Message> messages;
        public List<Message> flushedMessagesBetweenPolls;
        public FlexClientOutboundQueueProcessor processor;
        public AsyncPollWithTimeout asyncPoll;
        public boolean waitPoll;
        public FlexSession pushSession;
        public TimeoutAbstractObject flushTask;
        public int messageClientRefCount;
        public boolean avoidBusyPolling;
    }
}