/*
 * 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.endpoints;

import flex.messaging.FlexContext;
import flex.messaging.FlexSession;
import flex.messaging.HttpFlexSession;
import flex.messaging.MessageException;
import flex.messaging.client.EndpointPushNotifier;
import flex.messaging.client.FlexClient;
import flex.messaging.client.FlushResult;
import flex.messaging.client.UserAgentSettings;
import flex.messaging.config.ConfigMap;
import flex.messaging.log.Log;
import flex.messaging.messages.AcknowledgeMessage;
import flex.messaging.messages.AsyncMessage;
import flex.messaging.messages.CommandMessage;
import flex.messaging.messages.Message;
import flex.messaging.messages.MessagePerformanceInfo;
import flex.messaging.messages.MessagePerformanceUtils;
import flex.messaging.util.TimeoutManager;
import flex.messaging.util.UserAgentManager;

import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;

/**
 * Base class for HTTP-based endpoints that support streaming HTTP connections to
 * connected clients.
 * Each streaming connection managed by this endpoint consumes one of the request
 * handler threads provided by the servlet container, so it is not highly scalable
 * but offers performance advantages over client polling for clients receiving
 * a steady, rapid stream of pushed messages.
 * This endpoint does not support polling clients and will fault any poll requests
 * that are received. To support polling clients use subclasses of
 * BaseHTTPEndpoint instead.
 */
public abstract class BaseStreamingHTTPEndpoint extends BaseHTTPEndpoint
{
    //--------------------------------------------------------------------------
    //
    // Private Static Constants
    //
    //--------------------------------------------------------------------------

    /**
     * This token is used in chunked HTTP responses frequently so initialize it statically for general use.
     */
    private static final byte[] CRLF_BYTES = {(byte)13, (byte)10};

    /**
     * This token is used for the terminal chunk within a chunked response.
     */
    private static final byte ZERO_BYTE = (byte)48;

    /**
     * This token is used to signal that a chunk of data should be skipped by the client.
     */
    private static final byte NULL_BYTE = (byte)0;

    /**
     * Parameter name for 'command' passed in a request for a new streaming connection.
     */
    private static final String COMMAND_PARAM_NAME = "command";

    /**
     * This is the token at the end of the HTTP request line that indicates that it is
     * a stream connection that should be held open to push data back to the client,
     * as opposed to a regular request-response message.
     */
    private static final String OPEN_COMMAND = "open";

    /**
     * This is the token at the end of the HTTP request line that indicates that it is
     * a stream connection that should be closed.
     */
    private static final String CLOSE_COMMAND = "close";

    /**
     *  Parameter name for the stream ID; passed with commands for an existing streaming connection.
     */
    private static final String STREAM_ID_PARAM_NAME = "streamId";

    /**
     * Constant for HTTP/1.0.
     */
    private static final String HTTP_1_0 = "HTTP/1.0";

    /**
     * Thread name suffix for request threads that are servicing a pinned open streaming connection.
     */
    private static final String STREAMING_THREAD_NAME_EXTENSION = "-in-streaming-mode";

    /**
     * Configuration constants.
     */
    private static final String PROPERTY_CONNECTION_IDLE_TIMEOUT_MINUTES = "connection-idle-timeout-minutes";
    private static final String PROPERTY_LEGACY_CONNECTION_IDLE_TIMEOUT_MINUTES = "idle-timeout-minutes";
    private static final String MAX_STREAMING_CLIENTS = "max-streaming-clients";
    private static final String SERVER_TO_CLIENT_HEARTBEAT_MILLIS = "server-to-client-heartbeat-millis";
    private static final String PROPERTY_INVALIDATE_MESSAGECLIENT_ON_STREAMING_CLOSE = "invalidate-messageclient-on-streaming-close";

    /**
     * Defaults.
     */
    private static final boolean DEFAULT_INVALIDATE_MESSAGECLIENT_ON_STREAMING_CLOSE = false;
    private static final int DEFAULT_SERVER_TO_CLIENT_HEARTBEAT_MILLIS = 5000;
    private static final int DEFAULT_MAX_STREAMING_CLIENTS = 10;

    /**
     * Errors.
     */
    public static final String POLL_NOT_SUPPORTED_CODE = "Server.PollNotSupported";
    public static final int POLL_NOT_SUPPORTED_MESSAGE = 10034;


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

    /**
     * Constructs an unmanaged <code>BaseStreamingHTTPEndpoint</code>.
     */
    public BaseStreamingHTTPEndpoint()
    {
        this(false);
    }

    /**
     * Constructs an <code>BaseStreamingHTTPEndpoint</code> with the indicated management.
     *
     * @param enableManagement <code>true</code> if the <code>BaseStreamingHTTPEndpoint</code>
     * is manageable; <code>false</code> otherwise.
     */
    public BaseStreamingHTTPEndpoint(boolean enableManagement)
    {
        super(enableManagement);
    }

    //--------------------------------------------------------------------------
    //
    // Initialize, validate, start, and stop methods.
    //
    //--------------------------------------------------------------------------

    /**
     * Initializes the <code>Endpoint</code> with the properties.
     * If subclasses override this method, they must call <code>super.initialize()</code>.
     *
     * @param id The ID of the <code>Endpoint</code>.
     * @param properties Properties for the <code>Endpoint</code>.
     */
    @Override
    public void initialize(String id, ConfigMap properties)
    {
        super.initialize(id, properties);

        if (properties == null || properties.size() == 0)
        {
            // Initialize default user agent manager settings.
            UserAgentManager.setupUserAgentManager(null, userAgentManager);

            return; // Nothing else to initialize.
        }

        // The interval that the server will check if the client is still available.
        serverToClientHeartbeatMillis = properties.getPropertyAsLong(SERVER_TO_CLIENT_HEARTBEAT_MILLIS, DEFAULT_SERVER_TO_CLIENT_HEARTBEAT_MILLIS);
        setServerToClientHeartbeatMillis(serverToClientHeartbeatMillis);

        setInvalidateMessageClientOnStreamingClose(properties.getPropertyAsBoolean(PROPERTY_INVALIDATE_MESSAGECLIENT_ON_STREAMING_CLOSE, DEFAULT_INVALIDATE_MESSAGECLIENT_ON_STREAMING_CLOSE));

        // Number of minutes a client can remain idle before the server times the connection out.
        int connectionIdleTimeoutMinutes = properties.getPropertyAsInt(PROPERTY_CONNECTION_IDLE_TIMEOUT_MINUTES, getConnectionIdleTimeoutMinutes());
        if (connectionIdleTimeoutMinutes != 0)
        {
            setConnectionIdleTimeoutMinutes(connectionIdleTimeoutMinutes);
        }
        else
        {
            connectionIdleTimeoutMinutes = properties.getPropertyAsInt(PROPERTY_LEGACY_CONNECTION_IDLE_TIMEOUT_MINUTES, getConnectionIdleTimeoutMinutes());
            if (connectionIdleTimeoutMinutes != 0)
                setConnectionIdleTimeoutMinutes(connectionIdleTimeoutMinutes);
        }

        // User-agent configuration for kick-start bytes and max streaming connections per session.
        UserAgentManager.setupUserAgentManager(properties, userAgentManager);

        // Maximum number of clients allowed to have streaming HTTP connections with the endpoint.
        maxStreamingClients = properties.getPropertyAsInt(MAX_STREAMING_CLIENTS, DEFAULT_MAX_STREAMING_CLIENTS);

        // Set initial state for the canWait flag based on whether we allow waits or not.
        canStream = (maxStreamingClients > 0);
    }


    @Override
    public void start()
    {
        if (isStarted())
            return;

        super.start();

        if (connectionIdleTimeoutMinutes > 0)
        {
            pushNotifierTimeoutManager = new TimeoutManager(new ThreadFactory()
                                                            {
                                                                int counter = 1;
                                                                public synchronized Thread newThread(Runnable runnable)
                                                                {
                                                                    Thread t = new Thread(runnable);
                                                                    t.setName(getId() + "-StreamingConnectionTimeoutThread-" + counter++);
                                                                    return t;
                                                                }
                                                            });
        }

        currentStreamingRequests = new ConcurrentHashMap<String, EndpointPushNotifier>();
    }

    /**
     * (non-JavaDoc)
     * @see flex.messaging.endpoints.AbstractEndpoint#stop()
     */
    @Override
    public void stop()
    {
        if (!isStarted())
            return;

        // Shutdown the timeout manager for streaming connections cleanly.
        if (pushNotifierTimeoutManager != null)
        {
            pushNotifierTimeoutManager.shutdown();
            pushNotifierTimeoutManager = null;
        }

        // Shutdown any currently open streaming connections.
        for (EndpointPushNotifier notifier : currentStreamingRequests.values())
            notifier.close();

        currentStreamingRequests = null;

        super.stop();
    }

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

    /**
     * Used to synchronize sets and gets to the number of streaming clients.
     */
    protected final Object lock = new Object();

    /**
     * Used to keep track of the mapping between user agent match strings and
     * the bytes needed to kickstart their streaming connections.
     */
    protected UserAgentManager userAgentManager = new UserAgentManager();

    /**
     * This flag is volatile to allow for consistent reads across thread without
     * needing to pay the cost for a synchronized lock for each read.
     */
    private volatile boolean canStream = true;

    /**
     * Manages timing out EndpointPushNotifier instances.
     */
    private volatile TimeoutManager pushNotifierTimeoutManager;

    /**
     * A Map(EndpointPushNotifier, Boolean.TRUE) containing all currently open streaming notifiers
     * for this endpoint.
     * Used for clean shutdown.
     */
    private ConcurrentHashMap<String, EndpointPushNotifier> currentStreamingRequests;

    //--------------------------------------------------------------------------
    //
    // Properties
    //
    //--------------------------------------------------------------------------

    //------------------------------------------
    //  invalidateMessageClientOnStreamingClose
    //-----------------------------------------

    private volatile boolean invalidateMessageClientOnStreamingClose = DEFAULT_INVALIDATE_MESSAGECLIENT_ON_STREAMING_CLOSE;

    /**
     * Returns whether invalidate-messageclient-on-streaming-close is enabled.
     * See {@link BaseStreamingHTTPEndpoint#setInvalidateMessageClientOnStreamingClose(boolean)}
     * for details.
     *
     * @return <code>true</code> if the invalidate-messageclient-on-streaming-close is enabled, <code>false</code> otherwise.
     */
    public boolean isInvalidateMessageClientOnStreamingClose()
    {
        return invalidateMessageClientOnStreamingClose;
    }

    /**
     * Sets the invalidate-messageclient-on-streaming close property. If enabled,
     * when the streaming connection is closed for whatever reason (for example, the client is gone),
     * the client's associated MessageClient on the server is invalidated immediately.
     * This is useful in scenarios where there is a constant stream of messages, the client is gone,
     * and the streaming connection is closed, but the session has not timed out on the server yet.
     * In that case, enabling this property will prevent messages accumulating on the session on behalf
     * of the MessageClient that will invalidate.
     * <p>
     * Important: Do not enable this property when reliable messaging is used, otherwise
     * reliable reconnect attempts will not happen correctly.</p>
     *
     * @param value The property value.
     */
    public void setInvalidateMessageClientOnStreamingClose(boolean value)
    {
        invalidateMessageClientOnStreamingClose = value;
    }

    //----------------------------------
    //  serverToClientHeartbeatMillis
    //----------------------------------

    private long serverToClientHeartbeatMillis = DEFAULT_SERVER_TO_CLIENT_HEARTBEAT_MILLIS;

    /**
     * Retrieves the number of milliseconds the server will wait before writing a
     * single <code>null</code> byte to the streaming connection, to ensure the client is
     * still available.
     * @return The server-to-client heartbeat time in milliseconds.
     */
    public long getServerToClientHeartbeatMillis()
    {
        return serverToClientHeartbeatMillis;
    }

    /**
     * Retrieves the number of milliseconds the server will wait before writing a
     * single <code>null</code> byte to the streaming connection to ensure the client is
     * still available when there are no new messages for the client.
     * A non-positive value means the server will wait forever for new messages and
     * it will not write the <code>null</code> byte to determine if the client is available.
     * @param serverToClientHeartbeatMillis The server-to-client heartbeat time in milliseconds.
     */
    public void setServerToClientHeartbeatMillis(long serverToClientHeartbeatMillis)
    {
        if (serverToClientHeartbeatMillis < 0)
            serverToClientHeartbeatMillis = 0;
        this.serverToClientHeartbeatMillis = serverToClientHeartbeatMillis;
    }

    //----------------------------------
    //  connectionIdleTimeoutMinutes
    //----------------------------------

    private int connectionIdleTimeoutMinutes = 0;

    /**
     * Retrieves the number of minutes a client can remain idle before the server
     * times the connection out. The default value is 0, indicating that connections
     * will not be timed out and must be closed by the client or server, either explicitly
     * or by either process terminating.
     *
     * @return The number of minutes a client can remain idle before the server
     * times the connection out.
     */
    public int getConnectionIdleTimeoutMinutes()
    {
        return connectionIdleTimeoutMinutes;
    }

    /**
     * Sets the number of minutes a client can remain idle before the server
     * times the connection out. A value of 0 or below indicates that
     * connections will not be timed out.
     *
     * @param value The number of minutes a client can remain idle
     * before the server times the connection out.
     */
    public void setConnectionIdleTimeoutMinutes(int value)
    {
        if (value < 0)
            value = 0;

        this.connectionIdleTimeoutMinutes = value;
    }

    /**
     * (non-JavaDoc)
     * @deprecated Use {@link BaseStreamingHTTPEndpoint#getConnectionIdleTimeoutMinutes()} instead.
     */
    public int getIdleTimeoutMinutes()
    {
        return getConnectionIdleTimeoutMinutes();
    }

    /**
     * (non-JavaDoc)
     * @deprecated Use {@link BaseStreamingHTTPEndpoint#setConnectionIdleTimeoutMinutes(int)} instead.
     */
    public void setIdleTimeoutMinutes(int value)
    {
        setConnectionIdleTimeoutMinutes(value);
    }

    //----------------------------------
    //  maxStreamingClients
    //----------------------------------

    private int maxStreamingClients = DEFAULT_MAX_STREAMING_CLIENTS;

    /**
     * Retrieves the maximum number of clients that will be allowed to establish
     * a streaming HTTP connection with the endpoint.
     *
     * @return The maximum number of clients that will be allowed to establish
     * a streaming HTTP connection with the endpoint.
     */
    public int getMaxStreamingClients()
    {
        return maxStreamingClients;
    }

    /**
     * Sets the maximum number of clients that will be allowed to establish
     * a streaming HTTP connection with the server.
     *
     * @param maxStreamingClients The maximum number of clients that will be allowed
     * to establish a streaming HTTP connection with the server.
     */
    public void setMaxStreamingClients(int maxStreamingClients)
    {
        this.maxStreamingClients = maxStreamingClients;
        canStream = (streamingClientsCount < maxStreamingClients);
    }

    //----------------------------------
    //  streamingClientsCount
    //----------------------------------

    protected int streamingClientsCount;

    /**
     * Retrieves the the number of clients that are currently in the streaming state.
     *
     * @return The number of clients that are currently in the streaming state.
     */
    public int getStreamingClientsCount()
    {
        return streamingClientsCount;
    }

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

    /**
     * Handles HTTP requests targetting this endpoint.
     * Two types or requests are supported. If the request is a regular request-response AMF/AMFX
     * message it is handled by the base logic in BaseHTTPEndpoint.service. However, if it is a
     * request to open a streaming HTTP connection to the client this endpoint performs some
     * validation checks and then holds the connection open to stream data back to the client
     * over.
     *
     * @param req The original servlet request.
     * @param res The active servlet response.
     */
    @Override
    public void service(HttpServletRequest req, HttpServletResponse res)
    {
        String command = req.getParameter(COMMAND_PARAM_NAME);
        if (command != null)
            serviceStreamingRequest(req, res);
        else // Let BaseHTTPEndpoint logic handle regular request-response messaging.
            super.service(req, res);
    }

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

    /**
     * If the message has MPI enabled, this method adds all the needed performance
     * headers to the message.
     *
     * @param message Message to add performance headers to.
     */
    protected void addPerformanceInfo(Message message)
    {
        MessagePerformanceInfo mpiOriginal = getMPI(message);
        if (mpiOriginal == null)
            return;

        MessagePerformanceInfo mpip = (MessagePerformanceInfo)mpiOriginal.clone();
        try
        {
            // Set the original message info as the pushed causer info.
            MessagePerformanceUtils.setMPIP(message, mpip);
            MessagePerformanceUtils.setMPII(message, null);
        }
        catch (Exception e)
        {
            if (Log.isDebug())
                log.debug("MPI exception while streaming the message: " + e.toString());
        }

        // Overhead only used when MPI is enabled for sizing
        MessagePerformanceInfo mpio = new MessagePerformanceInfo();
        if (mpip.recordMessageTimes)
        {
            mpio.sendTime = System.currentTimeMillis();
            mpio.infoType = "OUT";
        }
        mpio.pushedFlag = true;
        MessagePerformanceUtils.setMPIO(message, mpio);

        // If MPI sizing information is enabled serialize again so that we know size
        if (mpip.recordMessageSizes)
        {
            try
            {
                // Each subclass serializes the message in their own format to
                // get the message size for the MPIO.
                long serializationOverhead = System.currentTimeMillis();
                mpio.messageSize = getMessageSizeForPerformanceInfo(message);

                // Set serialization overhead to the time calculated during serialization above
                if (mpip.recordMessageTimes)
                {
                    serializationOverhead = System.currentTimeMillis() - serializationOverhead;
                    mpip.addToOverhead(serializationOverhead);
                    mpiOriginal.addToOverhead(serializationOverhead);
                    mpio.sendTime = System.currentTimeMillis();
                }
            }
            catch(Exception e)
            {
                log.debug("MPI exception while streaming the message: " + e.toString());
            }
        }
    }

    /**
     * Utility method to convert streamed push messages to their small versions
     * if the channel-endpoint combination supports small messages.
     *
     * @param message The regular message.
     * @return The small message if the message has a small version, or regular message
     * if it doesn't .
     */
    protected Message convertPushMessageToSmall(Message message)
    {
        FlexSession session = FlexContext.getFlexSession();
        if (session != null && session.useSmallMessages())
            return convertToSmallMessage(message);
        return message;
    }

    /**
     * Used internally for performance information gathering; not intended for
     * public use. The default implementation of this method returns zero.
     * Subclasses should overwrite if they want to accurately report message
     * size information in performance information gathering.
     *
     * @param message Message to get the size for.
     *
     * @return The size of the message after message is serialized.
     */
    protected long getMessageSizeForPerformanceInfo(Message message)
    {
        return 0;
    }

    /**
     * This streaming endpoint does not support polling clients.
     *
     * @param flexClient The FlexClient that issued the poll request.
     * @param pollCommand The poll command from the client.
     * @return The flush info used to build the poll response.
     */
    @Override
    protected FlushResult handleFlexClientPoll(FlexClient flexClient, CommandMessage pollCommand)
    {
        MessageException me = new MessageException();
        me.setMessage(POLL_NOT_SUPPORTED_MESSAGE);
        me.setDetails(POLL_NOT_SUPPORTED_MESSAGE);
        me.setCode(POLL_NOT_SUPPORTED_CODE);
        throw me;
    }

    /**
     * Handles streaming connection open command sent by the FlexClient.
     *
     * @param req The <code>HttpServletRequest</code> to service.
     * @param res The <code>HttpServletResponse</code> to be used in case an error
     * has to be sent back.
     * @param flexClient FlexClient that requested the streaming connection.
     */
    protected void handleFlexClientStreamingOpenRequest(HttpServletRequest req, HttpServletResponse res, FlexClient flexClient)
    {
        FlexSession session = FlexContext.getFlexSession();
        if (canStream && session.canStream)
        {
            // If canStream/session.canStream is true it means we currently have
            // less than the max number of allowed streaming threads, per endpoint/session.

            // We need to protect writes/reads to the stream count with the endpoint's lock.
            // Also, we have to be careful to handle the case where two threads get to this point when only
            // one streaming spot remains; one thread will win and the other needs to fault.
            boolean thisThreadCanStream;
            synchronized (lock)
            {
                ++streamingClientsCount;
                if (streamingClientsCount == maxStreamingClients)
                {
                    thisThreadCanStream = true; // This thread got the last spot.
                    canStream = false;
                }
                else if (streamingClientsCount > maxStreamingClients)
                {
                    thisThreadCanStream = false; // This thread was beaten out for the last spot.
                    --streamingClientsCount; // Decrement the count because we're not going to grant the streaming right to the client.
                }
                else
                {
                    // We haven't hit the limit yet, allow this thread to stream.
                    thisThreadCanStream = true;
                }
            }

            // If the thread cannot wait due to endpoint streaming connection
            // limit, inform the client and return.
            if (!thisThreadCanStream)
            {
                String errorMessage = "Endpoint with id '" + getId() + "' cannot grant streaming connection to FlexClient with id '"
                        + flexClient.getId() + "' because " + MAX_STREAMING_CLIENTS + " limit of '"
                        + maxStreamingClients + "' has been reached.";
                if (Log.isError())
                    log.error(errorMessage);
                try
                {
                    errorMessage = "Endpoint with id '" + getId() + "' cannot grant streaming connection to FlexClient with id '"
                    + flexClient.getId() + "' because " + MAX_STREAMING_CLIENTS + " limit has been reached.";
                    res.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, errorMessage);
                }
                catch (IOException ignore)
                {}
                return;
            }

            // Setup for specific user agents.
            byte[] kickStartBytesToStream = null;
            String userAgentValue = req.getHeader(UserAgentManager.USER_AGENT_HEADER_NAME);
            UserAgentSettings agentSettings = userAgentManager.match(userAgentValue);
            if (agentSettings != null)
            {
                synchronized (session)
                {
                    session.maxConnectionsPerSession = agentSettings.getMaxPersistentConnectionsPerSession();
                }

                int kickStartBytes = agentSettings.getKickstartBytes();
                if (kickStartBytes > 0)
                {
                    // Determine the minimum number of actual bytes that need to be sent to
                    // kickstart, taking into account transfer-encoding overhead.
                    try
                    {
                        int chunkLengthHeaderSize = Integer.toHexString(kickStartBytes).getBytes("ASCII").length;
                        int chunkOverhead = chunkLengthHeaderSize + 4; // 4 for the 2 wrapping CRLF tokens.
                        int minimumKickstartBytes = kickStartBytes - chunkOverhead;
                        kickStartBytesToStream = new byte[(minimumKickstartBytes > 0) ? minimumKickstartBytes :
                                kickStartBytes];
                    }
                    catch (UnsupportedEncodingException ignore)
                    {
                        kickStartBytesToStream = new byte[kickStartBytes];
                    }
                    Arrays.fill(kickStartBytesToStream, NULL_BYTE);
                }
            }

            // Now, check with the session before granting the streaming connection.
            synchronized(session)
            {
                ++session.streamingConnectionsCount;
                if (session.maxConnectionsPerSession == FlexSession.MAX_CONNECTIONS_PER_SESSION_UNLIMITED)
                {
                    thisThreadCanStream = true;
                }
                else if (session.streamingConnectionsCount == session.maxConnectionsPerSession)
                {
                    thisThreadCanStream = true; // This thread got the last spot in the session.
                    session.canStream = false;
                }
                else if (session.streamingConnectionsCount > session.maxConnectionsPerSession)
                {
                    thisThreadCanStream = false; // This thread was beaten out for the last spot.
                    --session.streamingConnectionsCount;
                    synchronized(lock)
                    {
                        // Decrement the endpoint count because we're not going to grant the streaming right to the client.
                        --streamingClientsCount;
                    }
                }
                else
                {
                    // We haven't hit the limit yet, allow this thread to stream.
                    thisThreadCanStream = true;
                }
            }

            // If the thread cannot wait due to session streaming connection
            // limit, inform the client and return.
            if (!thisThreadCanStream)
            {
                if (Log.isError())
                    log.error("Endpoint with id '" + getId() + "' cannot grant streaming connection to FlexClient with id '"
                            + flexClient.getId() + "' because " + UserAgentManager.MAX_PERSISTENT_CONNECTIONS_PER_SESSION + " limit of '" + session.maxConnectionsPerSession
                            + ((agentSettings != null) ? "' for user-agent '" + agentSettings.getMatchOn() + "'" : "") +  " has been reached." );
                try
                {
                    // Return an HTTP status code 400.
                    String errorMessage = "The server cannot grant streaming connection to this client because limit has been reached.";
                    res.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, errorMessage);
                }
                catch (IOException ignore)
                {
                    // NOWARN
                }
                return;
            }

            Thread currentThread = Thread.currentThread();
            String threadName = currentThread.getName();
            EndpointPushNotifier notifier = null;
            boolean suppressIOExceptionLogging = false; // Used to suppress logging for IO exception.
            try
            {
                currentThread.setName(threadName + STREAMING_THREAD_NAME_EXTENSION);

                // Open and commit response headers and get output stream.
                if (addNoCacheHeaders)
                    addNoCacheHeaders(req, res);
                res.setContentType(getResponseContentType());
                res.setHeader("Transfer-Encoding", "chunked");
                res.setHeader("Connection", "close");
                ServletOutputStream os = res.getOutputStream();
                res.flushBuffer();

                // If kickstart-bytes are specified, stream them.
                if (kickStartBytesToStream != null)
                {
                    if (Log.isDebug())
                        log.debug("Endpoint with id '" + getId() + "' is streaming " + kickStartBytesToStream.length
                                + " bytes (not counting chunk encoding overhead) to kick-start the streaming connection for FlexClient with id '"
                                + flexClient.getId() + "'.");

                    streamChunk(kickStartBytesToStream, os, res);
                }

                // Setup serialization and type marshalling contexts
                setThreadLocals();

                // Activate streaming helper for this connection.
                // Watch out for duplicate stream issues.
                try
                {
                    notifier = new EndpointPushNotifier(this, flexClient);
                }
                catch (MessageException me)
                {
                    if (me.getNumber() == 10033) // It's a duplicate stream request from the same FlexClient. Leave the current stream in place and fault this.
                    {
                        if (Log.isWarn())
                            log.warn("Endpoint with id '" + getId() + "' received a duplicate streaming connection request from, FlexClient with id '"
                                    + flexClient.getId() + "'. Faulting request.");

                        // Rollback counters and send an error response.
                        synchronized (lock)
                        {
                            --streamingClientsCount;
                            canStream = (streamingClientsCount < maxStreamingClients);
                            synchronized (session)
                            {
                                --session.streamingConnectionsCount;
                                session.canStream = (session.maxConnectionsPerSession == FlexSession.MAX_CONNECTIONS_PER_SESSION_UNLIMITED
                                        || session.streamingConnectionsCount < session.maxConnectionsPerSession);
                            }
                        }
                        try
                        {
                            res.sendError(HttpServletResponse.SC_BAD_REQUEST);
                        }
                        catch (IOException ignore)
                        {
                            // NOWARN
                        }
                        return; // Exit early.
                    }
                }
                if (connectionIdleTimeoutMinutes > 0)
                    notifier.setIdleTimeoutMinutes(connectionIdleTimeoutMinutes);
                notifier.setLogCategory(getLogCategory());
                monitorTimeout(notifier);
                currentStreamingRequests.put(notifier.getNotifierId(), notifier);

                // Push down an acknowledgement for the 'connect' request containing the unique id for this specific stream.
                AcknowledgeMessage connectAck = new AcknowledgeMessage();
                connectAck.setBody(notifier.getNotifierId());
                connectAck.setCorrelationId(BaseStreamingHTTPEndpoint.OPEN_COMMAND);
                ArrayList toPush = new ArrayList(1);
                toPush.add(connectAck);
                streamMessages(toPush, os, res);

                // Output session level streaming count.
                if (Log.isDebug())
                    Log.getLogger(FlexSession.FLEX_SESSION_LOG_CATEGORY).info("Number of streaming clients for FlexSession with id '"+ session.getId() +"' is " + session.streamingConnectionsCount + ".");

                // Output endpoint level streaming count.
                if (Log.isDebug())
                    log.debug("Number of streaming clients for endpoint with id '"+ getId() +"' is " + streamingClientsCount + ".");

                // And cycle in a wait-notify loop with the aid of the helper until it
                // is closed, we're interrupted or the act of streaming data to the client fails.
                while (!notifier.isClosed())
                {
                    try
                    {
                        // Drain any messages that might have been accumulated
                        // while the previous drain was being processed.
                        List<AsyncMessage> messages = null;
                        synchronized (notifier.pushNeeded)
                        {
                            messages = notifier.drainMessages();
                        }
                        streamMessages(messages, os, res);

                        synchronized (notifier.pushNeeded)
                        {
                            notifier.pushNeeded.wait(serverToClientHeartbeatMillis);
                        
                            messages = notifier.drainMessages();
                        }
                        // If there are no messages to send to the client, send an null
                        // byte as a heartbeat to make sure the client is still valid.
                        if (messages == null && serverToClientHeartbeatMillis > 0)
                        {
                            try
                            {
                                os.write(NULL_BYTE);
                                res.flushBuffer();
                            }
                            catch (IOException e)
                            {
                                if (Log.isWarn())
                                    log.warn("Endpoint with id '" + getId() + "' is closing the streaming connection to FlexClient with id '"
                                            + flexClient.getId() + "' because endpoint encountered a socket write error" +
                                            ", possibly due to an unresponsive FlexClient.", e);
                                break; // Exit the wait loop.
                            }
                        }
                        // Otherwise stream the messages to the client.
                        else
                        {
                            // Update the last time notifier was used to drain messages.
                            // Important for idle timeout detection.
                            notifier.updateLastUse();

                            streamMessages(messages, os, res);
                        }
                    }
                    catch (InterruptedException e)
                    {
                        if (Log.isWarn())
                            log.warn("Streaming thread '" + threadName + "' for endpoint with id '" + getId() + "' has been interrupted and the streaming connection will be closed.");
                        os.close();
                        break; // Exit the wait loop.
                    }

                    // Update the FlexClient last use time to prevent FlexClient from
                    // timing out when the client is still subscribed. It is important
                    // to do this outside synchronized(notifier.pushNeeded) to avoid
                    // thread deadlock!
                    flexClient.updateLastUse();
                }
                if (Log.isDebug())
                    log.debug("Streaming thread '" + threadName + "' for endpoint with id '" + getId() + "' is releasing connection and returning to the request handler pool.");
                suppressIOExceptionLogging = true;
                // Terminate the response.
                streamChunk(null, os, res);
            }
            catch (IOException e)
            {
                if (Log.isWarn() && !suppressIOExceptionLogging)
                    log.warn("Streaming thread '" + threadName + "' for endpoint with id '" + getId() + "' is closing connection due to an IO error.", e);
            }
            finally
            {
                currentThread.setName(threadName);

                // We're done so decrement the counts for streaming threads,
                // and update the canStream flag if necessary.
                synchronized (lock)
                {
                    --streamingClientsCount;
                    canStream = (streamingClientsCount < maxStreamingClients);
                    synchronized (session)
                    {
                        --session.streamingConnectionsCount;
                        session.canStream = (session.maxConnectionsPerSession == FlexSession.MAX_CONNECTIONS_PER_SESSION_UNLIMITED
                                || session.streamingConnectionsCount < session.maxConnectionsPerSession);
                    }
                }

                if (notifier != null && currentStreamingRequests != null)
                {
                    currentStreamingRequests.remove(notifier.getNotifierId());
                    notifier.close();
                }

                // Output session level streaming count.
                if (Log.isDebug())
                    Log.getLogger(FlexSession.FLEX_SESSION_LOG_CATEGORY).info("Number of streaming clients for FlexSession with id '"+ session.getId() +"' is " + session.streamingConnectionsCount + ".");

                // Output endpoint level streaming count.
                if (Log.isDebug())
                    log.debug("Number of streaming clients for endpoint with id '"+ getId() +"' is " + streamingClientsCount + ".");
            }
        }
        // Otherwise, client's streaming connection open request could not be granted.
        else
        {
            if (Log.isError())
            {
                String logString = null;
                if (!canStream)
                {
                    logString = "Endpoint with id '" + getId() + "' cannot grant streaming connection to FlexClient with id '"
                    + flexClient.getId() + "' because " + MAX_STREAMING_CLIENTS + " limit of '"
                    + maxStreamingClients + "' has been reached.";
                }
                else if (!session.canStream)
                {
                    logString = "Endpoint with id '" + getId() + "' cannot grant streaming connection to FlexClient with id '"
                    + flexClient.getId() + "' because " + UserAgentManager.MAX_STREAMING_CONNECTIONS_PER_SESSION + " limit of '"
                    + session.maxConnectionsPerSession + "' has been reached.";
                }
                if (logString != null)
                    log.error(logString);
            }

            try
            {
                // Return an HTTP status code 400 to indicate that client request can't be processed.
                String errorMessage = null;
                if (!canStream)
                {
                    errorMessage = "Endpoint with id '" + getId() + "' cannot grant streaming connection to FlexClient with id '"
                    + flexClient.getId() + "' because " + MAX_STREAMING_CLIENTS + " limit has been reached.";
                }
                else if (!session.canStream)
                {
                    errorMessage = "Endpoint with id '" + getId() + "' cannot grant streaming connection to FlexClient with id '"
                    + flexClient.getId() + "' because " + UserAgentManager.MAX_STREAMING_CONNECTIONS_PER_SESSION + " limit has been reached.";
                }
                res.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, errorMessage);
            }
            catch (IOException ignore)
            {}
        }
    }

    /**
     * Handles streaming connection close command sent by the FlexClient.
     *
     * @param req The <code>HttpServletRequest</code> to service.
     * @param res The <code>HttpServletResponse</code> to be used in case an error
     * has to be sent back.
     * @param flexClient FlexClient that requested the streaming connection.
     * @param streamId The id for the stream to close.
     */
    protected void handleFlexClientStreamingCloseRequest(HttpServletRequest req, HttpServletResponse res, FlexClient flexClient, String streamId)
    {
        if (streamId != null)
        {
            EndpointPushNotifier notifier = (EndpointPushNotifier)flexClient.getEndpointPushHandler(getId());
            if ((notifier != null) && notifier.getNotifierId().equals(streamId))
                notifier.close();
        }
    }

    /**
     * Service streaming connection commands.
     *
     * @param req The <code>HttpServletRequest</code> to service.
     * @param res The <code>HttpServletResponse</code> to be used in case an error
     * has to be sent back.
     */
    protected void serviceStreamingRequest(HttpServletRequest req, HttpServletResponse res)
    {
        // If this is a request for a streaming connection, make sure it's for a valid FlexClient
        // and that the FlexSession doesn't already have a streaming connection open.
        // Streaming requests are POSTs (to help prevent the possibility of caching) that carry the
        // following parameters:
        // command - Indicating a custom command for the endpoint; currently 'open' to request a new
        //           streaming connection be opened, and 'close' to request the streaming connection
        //           to close.
        // version - Indicates the streaming connection 'version' to use; it's here for backward comp. support
        //           if we need to change how commands are handled in a future product release.
        // DSId - The FlexClient id value that uniquely identifies the swf making the request.
        String command = req.getParameter(COMMAND_PARAM_NAME);

        // Only HTTP 1.1 is supported, disallow HTTP 1.0.
        if (req.getProtocol().equals(HTTP_1_0))
        {
            if (Log.isError())
                log.error("Endpoint with id '" + getId() + "' cannot service the streaming request made with " +
                " HTTP 1.0. Only HTTP 1.1 is supported.");

            try
            {
                // Return an HTTP status code 400 to indicate that the client's request was syntactically invalid (bad command).
                res.sendError(HttpServletResponse.SC_BAD_REQUEST);
            }
            catch (IOException ignore)
            {}
            return; // Abort further server processing.
        }

        if (!(command.equals(OPEN_COMMAND) || command.equals(CLOSE_COMMAND)))
        {
            if (Log.isError())
                log.error("Endpoint with id '" + getId() + "' cannot service the streaming request as the supplied command '"
                        + command + "' is invalid.");

            try
            {
                // Return an HTTP status code 400 to indicate that the client's request was syntactically invalid (bad command).
                res.sendError(HttpServletResponse.SC_BAD_REQUEST);
            }
            catch (IOException ignore)
            {}
            return; // Abort further server processing.
        }

        String flexClientId = req.getParameter(Message.FLEX_CLIENT_ID_HEADER);
        if (flexClientId == null)
        {
            if (Log.isError())
                log.error("Endpoint with id '" + getId() + "' cannot service the streaming request as no FlexClient id"
                        + " has been supplied in the request.");

            try
            {
                // Return an HTTP status code 400 to indicate that the client's request was syntactically invalid (missing id).
                res.sendError(HttpServletResponse.SC_BAD_REQUEST);
            }
            catch (IOException ignore)
            {}
            return; // Abort further server processing.
        }

        // Validate that the provided FlexClient id exists and is associated with the current session.
        // We don't do this validation with CLOSE_COMMAND because CLOSE_COMMAND can come in on a
        // different session. For example, when the session expires due to timeout, the streaming client
        // using that session sends a CLOSE_COMMAND on a new session to let the server know to clean client's
        // corresponding server constructs. In that case, server already knows that session has expired so
        // we can simply omit this validation.
        FlexClient flexClient = null;
        List<FlexClient> flexClients = FlexContext.getFlexSession().getFlexClients();
        boolean validFlexClientId = false;
        for (Iterator<FlexClient> iter = flexClients.iterator(); iter.hasNext();)
        {
            flexClient = iter.next();
            if (flexClient.getId().equals(flexClientId) && flexClient.isValid())
            {
                validFlexClientId = true;
                break;
            }
        }
        if (!command.equals(CLOSE_COMMAND) && !validFlexClientId)
        {
            if (Log.isError())
                log.error("Endpoint with id '" + getId() + "' cannot service the streaming request as either the supplied"
                        + " FlexClient id '" + flexClientId + " is not valid, or the FlexClient with that id is not valid.");

            try
            {
                // Return an HTTP status code 400 to indicate that the client's request was syntactically invalid (invalid id).
                res.sendError(HttpServletResponse.SC_BAD_REQUEST);
            }
            catch (IOException ignore)
            {}
            return; // Abort further server processing.
        }

        // If a close command is received and we don't have any flex clients registered simply invalidate 
        // the Flex Session. This will take care of the Flex Session that got created when the MB servlet 
        // was processing the CLOSE request.
        if (command.equals(CLOSE_COMMAND) && flexClients.size() == 0)
        {
            FlexSession flexSession = FlexContext.getFlexSession();
            if (flexSession instanceof HttpFlexSession)
            {
                ((HttpFlexSession)flexSession).invalidate(false);
            }
            return;
        }

        if (flexClient != null)
        {
            if (command.equals(OPEN_COMMAND))
                handleFlexClientStreamingOpenRequest(req, res, flexClient);
            else if (command.equals(CLOSE_COMMAND))
                handleFlexClientStreamingCloseRequest(req, res, flexClient, req.getParameter(STREAM_ID_PARAM_NAME));
        }
    }

    /**
     * Helper method to write a chunk of bytes to the output stream in an HTTP
     * "Transfer-Encoding: chunked" format.
     * If the bytes array is null or empty, a terminal chunk will be written to
     * signal the end of the response.
     * Once the chunk is written to the output stream, the stream will be flushed immediately (no buffering).
     *
     * @param bytes The array of bytes to write as a chunk in the response; or if null, the signal to write the final chunk to complete the response.
     * @param os The output stream the chunk will be written to.
     * @param response The HttpServletResponse, used to flush the chunk to the client.
     *
     * @throws IOException if writing the chunk to the output stream fails.
     */
    protected void streamChunk(byte[] bytes, ServletOutputStream os, HttpServletResponse response) throws IOException
    {
        if ((bytes != null) && (bytes.length > 0))
        {
            byte[] chunkLength = Integer.toHexString(bytes.length).getBytes("ASCII");
            os.write(chunkLength);
            os.write(CRLF_BYTES);
            os.write(bytes);
            os.write(CRLF_BYTES);
            response.flushBuffer();
        }
        else // Send final 'EOF' chunk for the response.
        {
            os.write(ZERO_BYTE);
            os.write(CRLF_BYTES);
            response.flushBuffer();
        }
    }

    /**
     * Helper method invoked by the endpoint request handler thread cycling in wait-notify.
     * Serializes messages and streams each to the client as a response chunk using streamChunk().
     *
     * @param messages The messages to serialize and push to the client.
     * @param os The output stream the chunk will be written to.
     * @param response The HttpServletResponse, used to flush the chunk to the client.
     */
    protected abstract void streamMessages(List messages, ServletOutputStream os, HttpServletResponse response) throws IOException;

    /**
     * Given a message, returns the MessagePerformanceInfo object if the message
     * performance gathering is enabled, returns null otherwise.
     *
     * @param message The message.
     * @return MessagePerformanceInfo if the message performance gathering is enabled,
     * null otherwise.
     */
    protected MessagePerformanceInfo getMPI(Message message)
    {
        return (isRecordMessageSizes() || isRecordMessageTimes())?
                MessagePerformanceUtils.getMPII(message) : null;
    }

    //--------------------------------------------------------------------------
    //
    // Private methods.
    //
    //--------------------------------------------------------------------------

    /**
     * Utility method used at EndpointPushNotifier construction to monitor it for timeout.
     *
     * @param notifier The EndpointPushNotifier to monitor.
     */
    private void monitorTimeout(EndpointPushNotifier notifier)
    {
        if (pushNotifierTimeoutManager != null)
            pushNotifierTimeoutManager.scheduleTimeout(notifier);
    }
}