/*
 *
 * 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 org.apache.qpid.client;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.qpid.AMQConnectionClosedException;
import org.apache.qpid.AMQDisconnectedException;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQTimeoutException;
import org.apache.qpid.QpidException;
import org.apache.qpid.client.failover.FailoverException;
import org.apache.qpid.client.failover.FailoverState;
import org.apache.qpid.client.protocol.AMQProtocolSession;
import org.apache.qpid.client.protocol.BlockingMethodFrameListener;
import org.apache.qpid.client.state.AMQState;
import org.apache.qpid.client.state.AMQStateManager;
import org.apache.qpid.client.state.StateWaiter;
import org.apache.qpid.client.state.listener.SpecificMethodFrameListener;
import org.apache.qpid.codec.ClientDecoder;
import org.apache.qpid.configuration.ClientProperties;
import org.apache.qpid.framing.AMQBody;
import org.apache.qpid.framing.AMQDataBlock;
import org.apache.qpid.framing.AMQFrame;
import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.framing.AMQProtocolHeaderException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.ConnectionCloseBody;
import org.apache.qpid.framing.ConnectionCloseOkBody;
import org.apache.qpid.framing.HeartbeatBody;
import org.apache.qpid.framing.MethodRegistry;
import org.apache.qpid.framing.ProtocolInitiation;
import org.apache.qpid.framing.ProtocolVersion;
import org.apache.qpid.protocol.ErrorCodes;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.protocol.AMQMethodListener;
import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.ByteBufferSender;
import org.apache.qpid.transport.ConnectionSettings;
import org.apache.qpid.transport.ExceptionHandlingByteBufferReceiver;
import org.apache.qpid.transport.TransportException;
import org.apache.qpid.transport.network.NetworkConnection;
import org.apache.qpid.transport.network.TransportActivity;

public class AMQProtocolHandler implements ExceptionHandlingByteBufferReceiver, TransportActivity
{
    /** Used for debugging. */
    private static final Logger _logger = LoggerFactory.getLogger(AMQProtocolHandler.class);

    private static final long MAXIMUM_STATE_WAIT_TIME = Long.parseLong(System.getProperty("amqj.MaximumStateWait", "30000"));
    private static final String AMQJ_DEFAULT_SYNCWRITE_TIMEOUT = "amqj.default_syncwrite_timeout";

    /**
     * The connection that this protocol handler is associated with. There is a 1-1 mapping between connection
     * instances and protocol handler instances.
     */
    private final AMQConnection _connection;

    /** Our wrapper for a protocol session that provides access to session values in a typesafe manner. */
    private final AMQProtocolSession _protocolSession;

    /** Holds the state of the protocol session. */
    private AMQStateManager _stateManager;

    /** Holds the method listeners, */
    private final CopyOnWriteArraySet<AMQMethodListener> _frameListeners = new CopyOnWriteArraySet<AMQMethodListener>();

    /**
     * We create the failover handler when the session is created since it needs a reference to the IoSession in order
     * to be able to send errors during failover back to the client application. The session won't be available in the
     * case where we failing over due to a Connection.Redirect message from the broker.
     */
    private final FailoverHandler _failoverHandler;

    /**
     * This flag is used to track whether failover is being attempted. It is used to prevent the application constantly
     * attempting failover where it is failing.
     */
    private FailoverState _failoverState = FailoverState.NOT_STARTED;

    /** Used to provide a condition to wait upon for operations that are required to wait for failover to complete. */
    private CountDownLatch _failoverLatch;

    /** Object to lock on when changing the _failoverLatch  */
    private final Object _failoverLatchChange = new Object();

    /** The last failover exception that occurred */
    private FailoverException _lastFailoverException;

    /** Defines the default timeout to use for synchronous protocol commands. */
    private final long DEFAULT_SYNC_TIMEOUT = Long.getLong(ClientProperties.QPID_SYNC_OP_TIMEOUT,
                                                           Long.getLong(AMQJ_DEFAULT_SYNCWRITE_TIMEOUT,
                                                                        ClientProperties.DEFAULT_SYNC_OPERATION_TIMEOUT));
    private ClientDecoder _decoder;

    private ProtocolVersion _suggestedProtocolVersion;

    private long _writtenBytes;
    private long _readBytes;

    private int _messageReceivedCount;
    private int _messagesOut;


    private NetworkConnection _network;
    private ByteBufferSender _sender;
    private long _lastReadTime = System.currentTimeMillis();
    private long _lastWriteTime = System.currentTimeMillis();
    private HeartbeatListener _heartbeatListener = HeartbeatListener.DEFAULT;
    private Throwable _initialConnectionException;

    private static final int REUSABLE_BYTE_BUFFER_CAPACITY = 65 * 1024;
    private final byte[] _reusableBytes = new byte[REUSABLE_BYTE_BUFFER_CAPACITY];

    private int _queueId = 1;
    private final Object _queueIdLock = new Object();

    /**
     * Creates a new protocol handler, associated with the specified client connection instance.
     *
     * @param con The client connection that this is the event handler for.
     */
    public AMQProtocolHandler(AMQConnection con)
    {
        _connection = con;
        _protocolSession = new AMQProtocolSession(this, _connection);
        _stateManager = new AMQStateManager(_protocolSession);
        _failoverHandler = new FailoverHandler(this);
    }

    /**
     * Called when the network connection is closed. This can happen, either because the client explicitly requested
     * that the connection be closed, in which case nothing is done, or because the connection died. In the case
     * where the connection died, an attempt to failover automatically to a new connection may be started. The failover
     * process will be started, provided that it is the clients policy to allow failover, and provided that a failover
     * has not already been started or failed.
     * <p>
     * TODO  Clarify: presumably exceptionCaught is called when the client is sending during a connection failure and
     * not otherwise? The above comment doesn't make that clear.
     */
    public void closed()
    {
        if (_connection.isClosed())
        {
            _logger.debug("Session closed called by client");
        }
        else
        {
            // Use local variable to keep flag whether fail-over allowed or not,
            // in order to execute AMQConnection#exceptionRecievedout out of synchronization block,
            // otherwise it might deadlock with failover mutex
            boolean failoverNotAllowed = false;
            boolean failedWithoutConnecting = false;
            Throwable initialConnectionException = null;
            synchronized (this)
            {
                if (_logger.isDebugEnabled())
                {
                    _logger.debug("Session closed called with failover state " + _failoverState);
                }

                // reconnetablility was introduced here so as not to disturb the client as they have made their intentions
                // known through the policy settings.
                if (_failoverState == FailoverState.NOT_STARTED)
                {
                    // close the sender
                    try
                    {
                        _sender.close();
                    }
                    catch (Exception e)
                    {
                        _logger.warn("Exception occurred on closing the sender", e);
                    }
                    if (_connection.failoverAllowed())
                    {
                        _failoverState = FailoverState.IN_PROGRESS;

                        _logger.debug("FAILOVER STARTING");
                        startFailoverThread();
                    }
                    else if (_connection.isConnected())
                    {
                        failoverNotAllowed = true;
                        if (_logger.isDebugEnabled())
                        {
                            _logger.debug("Failover not allowed by policy:" + _connection.getFailoverPolicy());
                        }
                    }
                    else
                    {
                        failedWithoutConnecting = true;
                        initialConnectionException = _initialConnectionException;
                        _logger.debug("We are in process of establishing the initial connection");
                    }
                    _initialConnectionException = null;
                }
                else
                {
                    _logger.debug("Not starting the failover thread as state currently " + _failoverState);
                }
            }

            if (failoverNotAllowed)
            {
                _connection.closed(new AMQDisconnectedException(
                        "Server closed connection and reconnection not permitted.", _stateManager.getLastException()));
            }
            else if(failedWithoutConnecting)
            {
                if(initialConnectionException == null)
                {
                    initialConnectionException = _stateManager.getLastException();
                }
                String message = initialConnectionException == null ? "" : initialConnectionException.getMessage();
                _connection.exceptionReceived(new QpidException(
                        "Connection could not be established: " + message, initialConnectionException));
            }
        }

        if (_logger.isDebugEnabled())
        {
            _logger.debug("Protocol Session [" + this + "] closed");
        }
    }

    /** See {@link FailoverHandler} to see rationale for separate thread. */
    private void startFailoverThread()
    {
        if(!_connection.isClosed())
        {
            final Thread failoverThread;
            try
            {
                failoverThread = Threading.getThreadFactory().createThread(
                        new Runnable()
                        {
                            @Override
                            public void run()
                            {

                                if (Thread.currentThread().isDaemon())
                                {
                                    throw new IllegalStateException("FailoverHandler must run on a non-daemon thread.");
                                }

                                // Create a latch, upon which tasks that must not run in parallel with a failover can wait for completion of
                                // the fail over.
                                final CountDownLatch failoverLatch = new CountDownLatch(1);
                                setFailoverLatch(failoverLatch);

                                try
                                {
                                    // We wake up listeners. If they can handle failover, they will extend the
                                    // FailoverRetrySupport class and will in turn block on the latch until failover
                                    // has completed before retrying the operation.
                                    notifyFailoverStarting();

                                    getConnection().doWithAllLocks(_failoverHandler);
                                }
                                finally
                                {
                                    failoverLatch.countDown();
                                    setFailoverLatch(null);
                                }
                            }
                        });
            }
            catch (Exception e)
            {
                throw new RuntimeException("Failed to create thread", e);
            }
            failoverThread.setName("Failover");
            // Do not inherit daemon-ness from current thread as this can be a daemon
            // thread such as a AnonymousIoService thread.
            failoverThread.setDaemon(false);
            failoverThread.start();
        }
    }

    public void readerIdle()
    {
        _logger.debug("Protocol Session [" + this + "] idle: reader");
        //  failover:
        _logger.warn("Timed out while waiting for heartbeat from peer.");
        _network.close();
    }

    public void writerIdle()
    {
        _logger.debug("Protocol Session [" + this + "] idle: writer");
        writeFrame(HeartbeatBody.FRAME);
        _heartbeatListener.heartbeatSent();
    }

    /**
     * Invoked when any exception is thrown by the NetworkDriver
     */
    public void exception(Throwable cause)
    {
        boolean causeIsAConnectionProblem =
                cause instanceof AMQConnectionClosedException ||
                cause instanceof IOException ||
                cause instanceof TransportException;

        if (causeIsAConnectionProblem)
        {
            //ensure the IoSender and IoReceiver are closed
            try
            {
                _network.close();
            }
            catch (Exception e)
            {
                //ignore
            }
        }
        FailoverState state = getFailoverState();
        if (state == FailoverState.NOT_STARTED)
        {
            if (causeIsAConnectionProblem)
            {
                _logger.info("Connection exception caught therefore going to attempt failover: " + cause, cause);
                _initialConnectionException = cause;
            }
            else
            {
                _connection.exceptionReceived(cause);
            }

            // FIXME Need to correctly handle other exceptions. Things like ...
            // AMQChannelClosedException
            // which will cause the JMSSession to end due to a channel close and so that Session needs
            // to be removed from the map so we can correctly still call close without an exception when trying to close
            // the server closed session.  See also CloseChannelMethodHandler as the sessionClose is never called on exception
        }
        // we reach this point if failover was attempted and failed therefore we need to let the calling app
        // know since we cannot recover the situation
        else if (state == FailoverState.FAILED)
        {
            _logger.error("Exception caught by protocol handler: " + cause, cause);

            // we notify the state manager of the error in case we have any clients waiting on a state
            // change. Those "waiters" will be interrupted and can handle the exception
            AMQDisconnectedException amqe = new AMQDisconnectedException("Failover could not re-establish connectivity: " + cause, cause);
            propagateExceptionToAllWaiters(amqe);
            _connection.closed(amqe);
        }
        else
        {
            _logger.warn("Exception caught by protocol handler: " + cause, cause);
        }
    }

    /**
     * There are two cases where we have other threads potentially blocking for events to be handled by this class.
     * These are for the state manager (waiting for a state change) or a frame listener (waiting for a particular type
     * of frame to arrive). When an error occurs we need to notify these waiters so that they can react appropriately.
     *
     * This should be called only when the exception is fatal for the connection.
     *
     * @param e the exception to propagate
     *
     * @see #propagateExceptionToFrameListeners
     */
    public void propagateExceptionToAllWaiters(Exception e)
    {
        getStateManager().error(e);

        propagateExceptionToFrameListeners(e);
    }

    /**
     * This caters for the case where we only need to propagate an exception to the the frame listeners to interupt any
     * protocol level waits.
     *
     * This will would normally be used to notify all Frame Listeners that Failover is about to occur and they should
     * stop waiting and relinquish the Failover lock. See {@link FailoverHandler}.
     *
     * Once the {@link FailoverHandler} has re-established the connection then the listeners will be able to re-attempt
     * their protocol request and so listen again for the correct frame.
     *
     * @param e the exception to propagate
     */
    public void propagateExceptionToFrameListeners(Exception e)
    {
        synchronized (_frameListeners)
        {
            if (!_frameListeners.isEmpty())
            {
                final Iterator it = _frameListeners.iterator();
                while (it.hasNext())
                {
                    final AMQMethodListener ml = (AMQMethodListener) it.next();
                    ml.error(e);
                }
            }
        }
    }

    public void notifyFailoverStarting()
    {
        // Set the last exception in the sync block to ensure the ordering with add.
        // either this gets done and the add does the ml.error
        // or the add completes first and the iterator below will do ml.error
        synchronized (_frameListeners)
        {
            _lastFailoverException = new FailoverException("Failing over about to start");
        }

        //Only notify the Frame listeners that failover is going to occur as the State listeners shouldn't be
        // interrupted unless failover cannot restore the state.
        propagateExceptionToFrameListeners(_lastFailoverException);
    }

    public void failoverInProgress()
    {
        _lastFailoverException = null;
    }

    public void received(ByteBuffer msg)
    {
        _readBytes += msg.remaining();
        _lastReadTime = System.currentTimeMillis();
        final List<AMQDataBlock> dataBlocks = _protocolSession.getMethodProcessor().getProcessedMethods();
        try
        {
            _decoder.decodeBuffer(msg);

            // Decode buffer
            int size = dataBlocks.size();
            for (int i = 0; i < size; i++)
            {
                AMQDataBlock message = dataBlocks.get(i);
                _logger.debug("RECV: {}", message);

                if(message instanceof AMQFrame)
                {

                    final long msgNumber = ++_messageReceivedCount;

                    if (((msgNumber % 1000) == 0) && _logger.isDebugEnabled())
                    {
                        _logger.debug("Received {} protocol messages", _messageReceivedCount);
                    }

                    AMQFrame frame = (AMQFrame) message;

                    final AMQBody bodyFrame = frame.getBodyFrame();

                    bodyFrame.handle(frame.getChannel(), _protocolSession);

                    _connection.bytesReceived(_readBytes);
                }
                else if (message instanceof ProtocolInitiation)
                {
                    // We get here if the server sends a response to our initial protocol header
                    // suggesting an alternate ProtocolVersion; the server will then close the
                    // connection.
                    try
                    {
                        ProtocolInitiation protocolInit = (ProtocolInitiation) message;
                        _suggestedProtocolVersion = protocolInit.checkVersion();
                        _logger.debug("Broker suggested using protocol version: {} ", _suggestedProtocolVersion);

                        // get round a bug in old versions of qpid whereby the connection is not closed
                        _stateManager.changeState(AMQState.CONNECTION_CLOSED);

                    }
                    catch (AMQProtocolHeaderException e)
                    {
                        _stateManager.error(e);
                        throw e;
                    }
                }
            }
        }
        catch (Exception e)
        {
            _logger.error("Exception processing frame", e);
            propagateExceptionToFrameListeners(e);
            _stateManager.propagateExceptionToStateWaiters(e);
            exception(e);
        }
        finally
        {
            dataBlocks.clear();
        }


    }

    public void methodBodyReceived(final int channelId, final AMQBody bodyFrame)
            throws QpidException
    {
        final AMQMethodEvent<AMQMethodBody> evt =
                new AMQMethodEvent<AMQMethodBody>(channelId, (AMQMethodBody) bodyFrame);

        try
        {

            boolean wasAnyoneInterested = getStateManager().methodReceived(evt);
            synchronized (_frameListeners)
            {
                if (!_frameListeners.isEmpty())
                {
                    //This iterator is safe from the error state as the frame listeners always add before they send so their
                    // will be ready and waiting for this response.
                    Iterator it = _frameListeners.iterator();
                    while (it.hasNext())
                    {
                        final AMQMethodListener listener = (AMQMethodListener) it.next();
                        wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested;
                    }
                }
            }
            if (!wasAnyoneInterested)
            {
                throw new QpidException("AMQMethodEvent " + evt + " was not processed by any listener.  Listeners:"
                                             + _frameListeners, null);
            }
        }
        catch (QpidException e)
        {
            propagateExceptionToFrameListeners(e);
            _stateManager.propagateExceptionToStateWaiters(e);
            exception(e);
        }

    }

    public StateWaiter createWaiter(Set<AMQState> states) throws QpidException
    {
        return getStateManager().createWaiter(states);
    }

    public void writeFrame(AMQDataBlock frame)
    {
        writeFrame(frame, true);
    }

    public  synchronized void writeFrame(AMQDataBlock frame, boolean flush)
    {
        _lastWriteTime = System.currentTimeMillis();
        _writtenBytes += frame.getSize();
        frame.writePayload(_sender);
        if(flush)
        {
            _sender.flush();
        }

        _logger.debug("SEND: {}", frame);

        final long sentMessages = _messagesOut++;

        final boolean debug = _logger.isDebugEnabled();

        if (debug && ((sentMessages % 1000) == 0))
        {
            _logger.debug("Sent {} protocol messages", _messagesOut);
        }

        _connection.bytesSent(_writtenBytes);

    }


    /**
     * Convenience method that writes a frame to the protocol session and waits for a particular response. Equivalent to
     * calling getProtocolSession().write() then waiting for the response.
     *
     * @param frame
     * @param listener the blocking listener. Note the calling thread will block.
     */
    public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQDataBlock frame, BlockingMethodFrameListener listener)
            throws QpidException, FailoverException
    {
        return writeCommandFrameAndWaitForReply(frame, listener, DEFAULT_SYNC_TIMEOUT);
    }

    /**
     * Convenience method that writes a frame to the protocol session and waits for a particular response. Equivalent to
     * calling getProtocolSession().write() then waiting for the response.
     *
     * @param frame
     * @param listener the blocking listener. Note the calling thread will block.
     */
    public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQDataBlock frame, BlockingMethodFrameListener listener,
                                                           long timeout) throws QpidException, FailoverException
    {
        try
        {
            synchronized (_frameListeners)
            {
                if (_lastFailoverException != null)
                {
                    throw _lastFailoverException;
                }

                if(_stateManager.getCurrentState() == AMQState.CONNECTION_CLOSED ||
                        _stateManager.getCurrentState() == AMQState.CONNECTION_CLOSING)
                {
                    Exception e = _stateManager.getLastException();
                    if (e != null)
                    {
                        if (e instanceof QpidException)
                        {
                            QpidException amqe = (QpidException) e;
                            throw amqe.cloneForCurrentThread();
                        }
                        else
                        {
                            throw new AMQException(ErrorCodes.INTERNAL_ERROR, e.getMessage(), e);
                        }
                    }
                }

                _frameListeners.add(listener);
                //FIXME: At this point here we should check or before add we should check _stateManager is in an open
                // state so as we don't check we are likely just to time out here as I believe is being seen in QPID-1255
            }
            writeFrame(frame);

            long actualTimeout = timeout == -1 ? DEFAULT_SYNC_TIMEOUT : timeout;
            return listener.blockForFrame(actualTimeout);
            // When control resumes before this line, a reply will have been received
            // that matches the criteria defined in the blocking listener
        }
        finally
        {
            // If we don't removeKey the listener then no-one will
            _frameListeners.remove(listener);
        }

    }

    /** More convenient method to write a frame and wait for it's response. */
    public AMQMethodEvent syncWrite(AMQFrame frame, Class responseClass) throws QpidException, FailoverException
    {
        return syncWrite(frame, responseClass, DEFAULT_SYNC_TIMEOUT);
    }

    /** More convenient method to write a frame and wait for it's response. */
    public AMQMethodEvent syncWrite(AMQFrame frame, Class responseClass, long timeout) throws QpidException, FailoverException
    {
        return writeCommandFrameAndWaitForReply(frame, new SpecificMethodFrameListener(frame.getChannel(), responseClass),
                                                timeout);
    }

    public void closeSession(AMQSession session) throws QpidException
    {
        _protocolSession.closeSession(session);
    }

    /**
     * Closes the connection.
     * <p>
     * If a failover exception occurs whilst closing the connection it is ignored, as the connection is closed
     * anyway.
     *
     * @param timeout The timeout to wait for an acknowledgment to the close request.
     *
     * @throws QpidException If the close fails for any reason.
     */
    public void closeConnection(long timeout) throws QpidException
    {
        if (getStateManager().getCurrentState().equals(AMQState.CONNECTION_OPEN))
        {
            // Connection is already closed then don't do a syncWrite
            try
            {
                final ConnectionCloseBody body = _protocolSession.getMethodRegistry().createConnectionCloseBody(
                        ErrorCodes.REPLY_SUCCESS,
                        // replyCode
                        new AMQShortString("JMS client is closing the connection."),
                        0,
                        0);
                final AMQFrame frame = body.generateFrame(0);

                syncWrite(frame, ConnectionCloseOkBody.class, timeout);
                _network.close();
                closed();
            }
            catch (AMQTimeoutException e)
            {
                closed();
            }
            catch (FailoverException e)
            {
                _logger.debug("FailoverException interrupted connection close, ignoring as connection closed anyway.");
            }
        }
    }

    /** @return the number of bytes read from this protocol session */
    public long getReadBytes()
    {
        return _readBytes;
    }

    /** @return the number of bytes written to this protocol session */
    public long getWrittenBytes()
    {
        return _writtenBytes;
    }

    public void blockUntilNotFailingOver() throws InterruptedException
    {
        CountDownLatch failoverLatch = getFailoverLatch();
        if (failoverLatch != null)
        {
            if (!failoverLatch.await(MAXIMUM_STATE_WAIT_TIME, TimeUnit.MILLISECONDS))
            {
                _logger.debug("Timed out after waiting {}ms for failover to complete.", MAXIMUM_STATE_WAIT_TIME);
            }
        }
    }

    public String generateQueueName()
    {
        int id;
        synchronized (_queueIdLock)
        {
            id = _queueId++;
        }
        // convert '.', '/', ':' and ';' to single '_', for spec compliance and readability
        String localAddress = getLocalAddress().toString().replaceAll("[./:;]", "_");
        String queueName = "tmp_" + localAddress + "_" + id;
        return queueName.replaceAll("_+", "_");
    }

    CountDownLatch getFailoverLatch()
    {
        synchronized (_failoverLatchChange)
        {
            return _failoverLatch;
        }
    }

    void setFailoverLatch(CountDownLatch failoverLatch)
    {
        synchronized (_failoverLatchChange)
        {
            _failoverLatch = failoverLatch;
        }
    }

    public AMQConnection getConnection()
    {
        return _connection;
    }

    public AMQStateManager getStateManager()
    {
        return _stateManager;
    }

    public void setStateManager(AMQStateManager stateManager)
    {
        _stateManager = stateManager;
        _stateManager.setProtocolSession(_protocolSession);
    }

    public AMQProtocolSession getProtocolSession()
    {
        return _protocolSession;
    }

    synchronized FailoverState getFailoverState()
    {
        return _failoverState;
    }

    public synchronized void setFailoverState(FailoverState failoverState)
    {
        _failoverState= failoverState;
    }

    public MethodRegistry getMethodRegistry()
    {
        return _protocolSession.getMethodRegistry();
    }

    public ProtocolVersion getProtocolVersion()
    {
        return _protocolSession.getProtocolVersion();
    }

    public SocketAddress getRemoteAddress()
    {
        return _network.getRemoteAddress();
    }

    public SocketAddress getLocalAddress()
    {
        return _network.getLocalAddress();
    }

    public void setNetworkConnection(NetworkConnection network)
    {
        setNetworkConnection(network, network.getSender());
    }

    public void setNetworkConnection(NetworkConnection network, ByteBufferSender sender)
    {
        _network = network;
        _sender = sender;
        _protocolSession.setSender(sender);
    }

    @Override
    public long getLastReadTime()
    {
        return _lastReadTime;
    }

    @Override
    public long getLastWriteTime()
    {
        return _lastWriteTime;
    }

    protected ByteBufferSender getSender()
    {
        return _sender;
    }

    public NetworkConnection getNetworkConnection()
    {
        return _network;
    }

    public ProtocolVersion getSuggestedProtocolVersion()
    {
        return _suggestedProtocolVersion;
    }


    public void setHeartbeatListener(HeartbeatListener listener)
    {
        _heartbeatListener = listener == null ? HeartbeatListener.DEFAULT : listener;
    }

    public void heartbeatBodyReceived()
    {
        _heartbeatListener.heartbeatReceived();
    }

    public void setMaxFrameSize(final long frameMax)
    {
        _decoder.setMaxFrameSize(frameMax == 0l || frameMax > (long) Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) frameMax);
    }

    public void init(final ConnectionSettings settings)
    {
        _decoder = new ClientDecoder(_protocolSession.getMethodProcessor());
        _protocolSession.init(settings);
    }

    public long getDefaultTimeout()
    {
        return DEFAULT_SYNC_TIMEOUT;
    }
}