/**
 * 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.proton.engine.impl;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.UnsignedShort;
import org.apache.qpid.proton.codec.AMQPDefinedTypes;
import org.apache.qpid.proton.codec.CompositeWritableBuffer;
import org.apache.qpid.proton.codec.DecoderImpl;
import org.apache.qpid.proton.codec.EncoderImpl;
import org.apache.qpid.proton.codec.WritableBuffer;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.EndpointError;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.EngineFactory;
import org.apache.qpid.proton.engine.ProtonJTransport;
import org.apache.qpid.proton.engine.Sasl;
import org.apache.qpid.proton.engine.Ssl;
import org.apache.qpid.proton.engine.SslDomain;
import org.apache.qpid.proton.engine.SslPeerDetails;
import org.apache.qpid.proton.engine.TransportException;
import org.apache.qpid.proton.engine.impl.ssl.ProtonSslEngineProvider;
import org.apache.qpid.proton.engine.impl.ssl.SslImpl;
import org.apache.qpid.proton.framing.TransportFrame;
import org.apache.qpid.proton.amqp.transport.Attach;
import org.apache.qpid.proton.amqp.transport.Begin;
import org.apache.qpid.proton.amqp.transport.Close;
import org.apache.qpid.proton.amqp.transport.Detach;
import org.apache.qpid.proton.amqp.transport.Disposition;
import org.apache.qpid.proton.amqp.transport.End;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.amqp.transport.Flow;
import org.apache.qpid.proton.amqp.transport.FrameBody;
import org.apache.qpid.proton.amqp.transport.Open;
import org.apache.qpid.proton.amqp.transport.Role;
import org.apache.qpid.proton.amqp.transport.Transfer;

public class TransportImpl extends EndpointImpl implements ProtonJTransport, FrameBody.FrameBodyHandler<Integer>,FrameTransport
{
    public static final byte[] HEADER = new byte[8];
    public static final org.apache.qpid.proton.amqp.messaging.Accepted ACCEPTED =
            new org.apache.qpid.proton.amqp.messaging.Accepted();

    static
    {
        HEADER[0] = (byte) 'A';
        HEADER[1] = (byte) 'M';
        HEADER[2] = (byte) 'Q';
        HEADER[3] = (byte) 'P';
        HEADER[4] = 0;
        HEADER[5] = 1;
        HEADER[6] = 0;
        HEADER[7] = 0;
    }

    private ConnectionImpl _connectionEndpoint;

    private boolean _isOpenSent;
    private boolean _isCloseSent;

    private boolean _headerWritten;
    private TransportSession[] _remoteSessions;
    private TransportSession[] _localSessions;

    private TransportInput _inputProcessor;
    private TransportOutput _outputProcessor;

    private Map<SessionImpl, TransportSession> _transportSessionState = new HashMap<SessionImpl, TransportSession>();
    private Map<LinkImpl, TransportLink> _transportLinkState = new HashMap<LinkImpl, TransportLink>();


    private DecoderImpl _decoder = new DecoderImpl();
    private EncoderImpl _encoder = new EncoderImpl(_decoder);

    private int _maxFrameSize = 16 * 1024;

    private final ByteBuffer _overflowBuffer = ByteBuffer.wrap(new byte[_maxFrameSize]);
    private static final byte AMQP_FRAME_TYPE = 0;
    private boolean _closeReceived;
    private Open _open;
    private SaslImpl _sasl;
    private SslImpl _ssl;
    private TransportException _inputException;
    private ProtocolTracer _protocolTracer = null;


    {
        AMQPDefinedTypes.registerAllTypes(_decoder, _encoder);
        _overflowBuffer.flip();
    }

    /**
     * @deprecated This constructor's visibility will be reduced to the default scope in a future release.
     * Client code outside this module should use a {@link EngineFactory} instead
     */
    @Deprecated public TransportImpl()
    {
        FrameParser frameParser = new FrameParser(this);

        _inputProcessor = frameParser;
        _outputProcessor = new TransportOutput()
                    {
                        @Override
                        public int output(byte[] bytes, int offset, int size)
                        {
                            return transportOutput(bytes, offset, size);
                        }
                    };
    }

    @Override
    public void bind(Connection conn)
    {
        // TODO - check if already bound
        ((ConnectionImpl) conn).setBound(true);
        _connectionEndpoint = (ConnectionImpl) conn;


        _localSessions = new TransportSession[_connectionEndpoint.getMaxChannels()+1];
        _remoteSessions = new TransportSession[_connectionEndpoint.getMaxChannels()+1];


        if(getRemoteState() != EndpointState.UNINITIALIZED)
        {
            _connectionEndpoint.handleOpen(_open);
            if(getRemoteState() == EndpointState.CLOSED)
            {
                _connectionEndpoint.setRemoteState(EndpointState.CLOSED);
            }
            _inputProcessor.input(new byte[0],0,0);
        }
    }

    @Override
    public int input(byte[] bytes, int offset, int length)
    {
        if(_inputException != null)
        {
            throw _inputException;
        }
        if(length == 0)
        {
            if(_connectionEndpoint == null || _connectionEndpoint.getRemoteState() != EndpointState.CLOSED)
            {
                throw new TransportException("Unexpected EOS: connection aborted");
            }
        }
        try
        {
            return  _inputProcessor.input(bytes, offset, length);
        }
        catch (TransportException e)
        {
            _inputException = e;
            throw e;
        }
    }

    //==================================================================================================================
    // Process model state to generate output

    @Override
    public int output(byte[] bytes, final int offset, final int size)
    {
        try
        {
            return _outputProcessor.output(bytes, offset, size);
        }
        catch (RuntimeException e)
        {
            e.printStackTrace();
            throw e;
        }
    }

    private int transportOutput(byte[] bytes, int offset, int size)
    {
        int written = 0;

        if(_overflowBuffer.hasRemaining())
        {
            final int overflowWritten = Math.min(size, _overflowBuffer.remaining());
            _overflowBuffer.get(bytes, offset, overflowWritten);
            written+=overflowWritten;
        }
        if(!_overflowBuffer.hasRemaining())
        {
            _overflowBuffer.clear();

            CompositeWritableBuffer outputBuffer =
                    new CompositeWritableBuffer(
                       new WritableBuffer.ByteBufferWrapper(ByteBuffer.wrap(bytes, offset + written, size - written)),
                       new WritableBuffer.ByteBufferWrapper(_overflowBuffer));

            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processHeader(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processOpen(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processBegin(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processAttach(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processReceiverDisposition(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processReceiverFlow(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processMessageData(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processSenderDisposition(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processSenderFlow(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processDetach(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processEnd(outputBuffer);
            if( outputBuffer.remaining() >= _maxFrameSize ) { written += processClose(outputBuffer);
            }}}}}}}}}}}}

            _overflowBuffer.flip();
            written -= _overflowBuffer.remaining();
        }

        return written;
    }

    public Sasl sasl()
    {
        if(_sasl == null)
        {
            _sasl = new SaslImpl();
            TransportWrapper transportWrapper = _sasl.wrap(_inputProcessor, _outputProcessor);
            _inputProcessor = transportWrapper;
            _outputProcessor = transportWrapper;
        }
        return _sasl;

    }

    /**
     * {@inheritDoc}
     *
     * <p>Note that sslDomain must implement {@link ProtonSslEngineProvider}. This is not possible
     * enforce at the API level because {@link ProtonSslEngineProvider} is not part of the
     * public Proton API.</p>
     */
    @Override
    public Ssl ssl(SslDomain sslDomain, SslPeerDetails sslPeerDetails)
    {
        if (_ssl == null)
        {
            _ssl = new SslImpl(sslDomain, sslPeerDetails);
            TransportWrapper transportWrapper = _ssl.wrap(_inputProcessor, _outputProcessor);
            _inputProcessor = transportWrapper;
            _outputProcessor = transportWrapper;
        }
        return _ssl;
    }

    @Override
    public Ssl ssl(SslDomain sslDomain)
    {
        return ssl(sslDomain, null);
    }

    private void clearTransportWorkList()
    {
        DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
        while(delivery != null)
        {
            DeliveryImpl transportWorkNext = delivery.getTransportWorkNext();
            delivery.clearTransportWork();
            delivery = transportWorkNext;
        }
    }

    private int processDetach(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
            while(endpoint != null && buffer.remaining() >= _maxFrameSize)
            {

                if(endpoint instanceof LinkImpl)
                {
                    LinkImpl link = (LinkImpl) endpoint;
                    TransportLink transportLink = getTransportState(link);
                    SessionImpl session = link.getSession();
                    TransportSession transportSession = getTransportState(session);

                    if(link.getLocalState() == EndpointState.CLOSED
                       && transportLink.isLocalHandleSet())
                    {
                        if(!(link instanceof SenderImpl)
                           || link.getQueued() == 0
                           || transportLink.detachReceived()
                           || transportSession.endReceived()
                           || _closeReceived)
                        {
                            UnsignedInteger localHandle = transportLink.getLocalHandle();
                            transportLink.clearLocalHandle();
                            transportSession.freeLocalHandle(localHandle);


                            Detach detach = new Detach();
                            detach.setHandle(localHandle);
                            // TODO - need an API for detaching rather than closing the link
                            detach.setClosed(true);

                            EndpointError localError = link.getLocalError();
                            if( localError !=null ) {
                                ErrorCondition error = new ErrorCondition();
                                error.setCondition(Symbol.getSymbol(localError.getName()));
                                error.setDescription(localError.getDescription());
                                detach.setError(error);
                            }


                            int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), detach, null, null);
                            written += frameBytes;
                            endpoint.clearModified();

                            // TODO - temporary hack for PROTON-154, this line should be removed and replaced
                            //        with proper handling for closed links
                            link.free();
                        }
                    }

                }
                endpoint = endpoint.transportNext();
            }
        }
        return written;
    }

    private int processSenderFlow(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
            while(endpoint != null && buffer.remaining() >= _maxFrameSize)
            {

                if(endpoint instanceof SenderImpl)
                {
                    SenderImpl sender = (SenderImpl) endpoint;
                    if(sender.getDrain() && sender.clearDrained())
                    {
                        TransportSender transportLink = sender.getTransportLink();
                        TransportSession transportSession = sender.getSession().getTransportSession();
                        UnsignedInteger credits = transportLink.getLinkCredit();
                        transportLink.setLinkCredit(UnsignedInteger.valueOf(0));
                        transportLink.setDeliveryCount(transportLink.getDeliveryCount().add(credits));
                        transportLink.setLinkCredit(UnsignedInteger.ZERO);

                        Flow flow = new Flow();
                        flow.setHandle(transportLink.getLocalHandle());
                        flow.setNextIncomingId(transportSession.getNextIncomingId());
                        flow.setIncomingWindow(transportSession.getIncomingWindowSize());
                        flow.setOutgoingWindow(transportSession.getOutgoingWindowSize());
                        flow.setDeliveryCount(transportLink.getDeliveryCount());
                        flow.setLinkCredit(transportLink.getLinkCredit());
                        flow.setDrain(sender.getDrain());
                        flow.setNextOutgoingId(transportSession.getNextOutgoingId());
                        int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
                        written += frameBytes;
                        endpoint.clearModified();
                    }

                }

                endpoint = endpoint.transportNext();
            }
        }
        return written;  //TODO - Implement
    }

    private int processSenderDisposition(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
            while(delivery != null && buffer.remaining() >= _maxFrameSize )
            {
                if((delivery.getLink() instanceof SenderImpl) && delivery.isLocalStateChange() && delivery.getTransportDelivery() != null)
                {
                    TransportDelivery transportDelivery = delivery.getTransportDelivery();
                    Disposition disposition = new Disposition();
                    disposition.setFirst(transportDelivery.getDeliveryId());
                    disposition.setLast(transportDelivery.getDeliveryId());
                    disposition.setRole(Role.SENDER);
                    disposition.setSettled(delivery.isSettled());
                    if(delivery.isSettled())
                    {
                        transportDelivery.settled();
                    }
                    disposition.setState(delivery.getLocalState());
                    int frameBytes = writeFrame(buffer, delivery.getLink().getSession()
                                                                      .getTransportSession().getLocalChannel(),
                                       disposition, null, null);
                    written += frameBytes;
                    delivery = delivery.clearTransportWork();
                }
                else
                {
                    delivery = delivery.getTransportWorkNext();
                }
            }
        }
        return written;

    }

    private int processMessageData(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();

            while(delivery != null && buffer.remaining() >= _maxFrameSize)
            {
                if((delivery.getLink() instanceof SenderImpl) && !(delivery.isDone() && delivery.getDataLength() == 0)
                   && delivery.getLink().getSession().getTransportSession().hasOutgoingCredit())
                {
                    SenderImpl sender = (SenderImpl) delivery.getLink();

                    sender.decrementQueued();


                    TransportLink transportLink = sender.getTransportLink();

                    UnsignedInteger deliveryId = transportLink.getDeliveryCount();
                    TransportDelivery transportDelivery = new TransportDelivery(deliveryId, delivery, transportLink);
                    delivery.setTransportDelivery(transportDelivery);

                    final Transfer transfer = new Transfer();
                    transfer.setDeliveryId(deliveryId);
                    transfer.setDeliveryTag(new Binary(delivery.getTag()));
                    transfer.setHandle(transportLink.getLocalHandle());

                    if(delivery.isSettled())
                    {
                        transfer.setSettled(Boolean.TRUE);
                    }
                    else
                    {
                        sender.getSession().getTransportSession().addUnsettledOutgoing(deliveryId, delivery);
                    }

                    if(delivery.getLink().current() == delivery)
                    {
                        transfer.setMore(true);
                    }

                    transfer.setMessageFormat(UnsignedInteger.ZERO);

                    // TODO - large frames
                    ByteBuffer payload = delivery.getData() ==  null ? null : ByteBuffer.wrap(delivery.getData(), delivery.getDataOffset(), delivery.getDataLength());

                    int frameBytes = writeFrame(buffer,
                                                sender.getSession().getTransportSession().getLocalChannel(),
                                                transfer, payload, new PartialTransfer(transfer));
                    sender.getSession().getTransportSession().incrementOutgoingId();

                    written += frameBytes;

                    if(payload == null || !payload.hasRemaining())
                    {
                        delivery.setData(null);
                        delivery.setDataLength(0);
                        delivery.setDone();

                        if(delivery.getLink().current() != delivery)
                        {
                            transportLink.setDeliveryCount(transportLink.getDeliveryCount().add(UnsignedInteger.ONE));
                            transportLink.setLinkCredit(transportLink.getLinkCredit().subtract(UnsignedInteger.ONE));
                        }

                        delivery = delivery.clearTransportWork();

                    }
                    else
                    {
                        delivery.setDataOffset(delivery.getDataOffset()+delivery.getDataLength()-payload.remaining());
                        delivery.setDataLength(payload.remaining());
                    }


                }
                else
                {
                    delivery = delivery.getTransportWorkNext();
                }
            }
        }
        return written;
    }

    private int processReceiverDisposition(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
            while(delivery != null && buffer.remaining() >= _maxFrameSize)
            {
                boolean remove = false;
                if((delivery.getLink() instanceof ReceiverImpl) && delivery.isLocalStateChange())
                {
                    remove = true;
                    TransportDelivery transportDelivery = delivery.getTransportDelivery();
                    Disposition disposition = new Disposition();
                    disposition.setFirst(transportDelivery.getDeliveryId());
                    disposition.setLast(transportDelivery.getDeliveryId());
                    disposition.setRole(Role.RECEIVER);
                    disposition.setSettled(delivery.isSettled());

                    disposition.setState(delivery.getLocalState());
                    int frameBytes = writeFrame(buffer, delivery.getLink().getSession()
                                                                      .getTransportSession().getLocalChannel(),
                                       disposition, null, null);
                    written += frameBytes;
                    if(delivery.isSettled())
                    {
                        transportDelivery.settled();
                    }
                    delivery = delivery.clearTransportWork();

                }
                else
                {
                    delivery = delivery.getTransportWorkNext();
                }
            }
        }
        return written;
    }

    private int processReceiverFlow(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
            while(endpoint != null && buffer.remaining() >= _maxFrameSize)
            {

                if(endpoint instanceof ReceiverImpl)
                {

                    ReceiverImpl receiver = (ReceiverImpl) endpoint;
                    TransportLink transportLink = getTransportState(receiver);
                    TransportSession transportSession = getTransportState(receiver.getSession());

                    if(receiver.getLocalState() == EndpointState.ACTIVE)
                    {
                        int credits = receiver.clearUnsentCredits();
                        transportSession.getSession().clearIncomingWindowResize();
                        if(credits != 0 || receiver.getDrain())
                        {
                            transportLink.addCredit(credits);
                            Flow flow = new Flow();
                            flow.setHandle(transportLink.getLocalHandle());
                            flow.setNextIncomingId(transportSession.getNextIncomingId());
                            flow.setIncomingWindow(transportSession.getIncomingWindowSize());
                            flow.setOutgoingWindow(transportSession.getOutgoingWindowSize());
                            flow.setDeliveryCount(transportLink.getDeliveryCount());
                            flow.setLinkCredit(transportLink.getLinkCredit());
                            flow.setDrain(receiver.getDrain());
                            flow.setNextOutgoingId(transportSession.getNextOutgoingId());
                            int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
                            written += frameBytes;
                            if(receiver.getLocalState() == EndpointState.ACTIVE)
                            {
                                endpoint.clearModified();
                            }
                        }

                    }
                }
                endpoint = endpoint.transportNext();
            }
            endpoint = _connectionEndpoint.getTransportHead();
            while(endpoint != null && buffer.remaining() >= _maxFrameSize)
            {

                if(endpoint instanceof SessionImpl)
                {

                    SessionImpl session = (SessionImpl) endpoint;
                    TransportSession transportSession = getTransportState(session);

                    if(session.getLocalState() == EndpointState.ACTIVE)
                    {
                        boolean windowResized = session.clearIncomingWindowResize();
                        if(windowResized)
                        {
                            Flow flow = new Flow();
                            flow.setIncomingWindow(transportSession.getIncomingWindowSize());
                            flow.setOutgoingWindow(transportSession.getOutgoingWindowSize());
                            flow.setNextOutgoingId(transportSession.getNextOutgoingId());
                            flow.setNextIncomingId(transportSession.getNextIncomingId());
                            int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
                            written += frameBytes;
                        }
                    }
                }
                endpoint = endpoint.transportNext();
            }
        }
        return written;
    }

    private int processAttach(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            EndpointImpl endpoint = _connectionEndpoint.getTransportHead();

            while(endpoint != null && buffer.remaining() >= _maxFrameSize)
            {
                if(endpoint instanceof LinkImpl)
                {

                    LinkImpl link = (LinkImpl) endpoint;
                    TransportLink transportLink = getTransportState(link);
                    if(link.getLocalState() != EndpointState.UNINITIALIZED && !transportLink.attachSent())
                    {

                        if( (link.getRemoteState() == EndpointState.ACTIVE
                            && !transportLink.isLocalHandleSet()) || link.getRemoteState() == EndpointState.UNINITIALIZED)
                        {

                            SessionImpl session = link.getSession();
                            TransportSession transportSession = getTransportState(session);
                            UnsignedInteger localHandle = transportSession.allocateLocalHandle(transportLink);

                            if(link.getRemoteState() == EndpointState.UNINITIALIZED)
                            {
                                transportSession.addHalfOpenLink(transportLink);
                            }

                            Attach attach = new Attach();
                            attach.setHandle(localHandle);
                            attach.setName(transportLink.getName());

                            if(link.getSenderSettleMode() != null)
                            {
                                attach.setSndSettleMode(link.getSenderSettleMode());
                            }

                            if(link.getReceiverSettleMode() != null)
                            {
                                attach.setRcvSettleMode(link.getReceiverSettleMode());
                            }

                            if(link.getSource() != null)
                            {
                                attach.setSource(link.getSource());
                            }

                            if(link.getTarget() != null)
                            {
                                attach.setTarget(link.getTarget());
                            }

                            attach.setRole(endpoint instanceof ReceiverImpl ? Role.RECEIVER : Role.SENDER);

                            if(link instanceof SenderImpl)
                            {
                                attach.setInitialDeliveryCount(UnsignedInteger.ZERO);
                            }

                            int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), attach, null, null);
                            written += frameBytes;
                            transportLink.sentAttach();
                            if(link.getLocalState() == EndpointState.ACTIVE && (link instanceof SenderImpl || !link.hasCredit()))
                            {
                                endpoint.clearModified();
                            }
                        }
                    }

                }

                endpoint = endpoint.transportNext();
            }
        }
        return written;
    }

    private void clearInterestList()
    {
        EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
        while(endpoint != null)
        {
            endpoint.clearModified();
            endpoint = endpoint.transportNext();
        }
    }

    private int processHeader(WritableBuffer buffer)
    {
        if(!_headerWritten)
        {
            buffer.put(HEADER, 0, HEADER.length);
            _headerWritten = true;
            return HEADER.length;
        }
        else
        {
            return 0;
        }
    }

    private int processOpen(WritableBuffer buffer)
    {
        if(_connectionEndpoint != null && _connectionEndpoint.getLocalState() != EndpointState.UNINITIALIZED && !_isOpenSent)
        {
            Open open = new Open();
            open.setContainerId(_connectionEndpoint.getLocalContainerId());
            open.setHostname(_connectionEndpoint.getHostname());
            // TODO - populate;

            _isOpenSent = true;

            return  writeFrame(buffer, 0, open, null, null);

        }
        return 0;
    }

    private int processBegin(WritableBuffer buffer)
    {
        int written = 0;

        if(_connectionEndpoint != null)
        {
            EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
            while(endpoint != null && buffer.remaining() >= _maxFrameSize)
            {
                if(endpoint instanceof SessionImpl)
                {
                    SessionImpl session = (SessionImpl) endpoint;
                    TransportSession transportSession = getTransportState(session);
                    if(session.getLocalState() != EndpointState.UNINITIALIZED && !transportSession.beginSent())
                    {
                        int channelId = allocateLocalChannel(transportSession);
                        Begin begin = new Begin();

                        if(session.getRemoteState() != EndpointState.UNINITIALIZED)
                        {
                            begin.setRemoteChannel(UnsignedShort.valueOf((short) transportSession.getRemoteChannel()));
                        }
                        begin.setHandleMax(transportSession.getHandleMax());
                        begin.setIncomingWindow(transportSession.getIncomingWindowSize());
                        begin.setOutgoingWindow(transportSession.getOutgoingWindowSize());
                        begin.setNextOutgoingId(transportSession.getNextOutgoingId());

                        written += writeFrame(buffer, channelId, begin, null, null);
                        transportSession.sentBegin();
                        if(session.getLocalState() == EndpointState.ACTIVE)
                        {
                            endpoint.clearModified();
                        }
                    }
                }
                endpoint = endpoint.transportNext();
            }
        }
        return written;
    }

    private TransportSession getTransportState(SessionImpl session)
    {
        TransportSession transportSession = _transportSessionState.get(session);
        if(transportSession == null)
        {
            transportSession = new TransportSession(session);
            session.setTransportSession(transportSession);
            _transportSessionState.put(session, transportSession);
        }
        return transportSession;
    }

    private TransportLink getTransportState(LinkImpl link)
    {
        TransportLink transportLink = _transportLinkState.get(link);
        if(transportLink == null)
        {
            transportLink = TransportLink.createTransportLink(link);
            _transportLinkState.put(link, transportLink);
        }
        return transportLink;
    }

    private int allocateLocalChannel(TransportSession transportSession)
    {
        for( int i=0; i < _localSessions.length; i++)
        {
            if( _localSessions[i] == null )
            {
                _localSessions[i] = transportSession;
                transportSession.setLocalChannel(i);
                return i;
            }
        }
        return -1;
    }

    private int freeLocalChannel(TransportSession transportSession)
    {
        final int channel = transportSession.getLocalChannel();
        _localSessions[channel] = null;
        transportSession.freeLocalChannel();
        return channel;
    }

    private int processEnd(WritableBuffer buffer)
    {
        int written = 0;
        if(_connectionEndpoint != null)
        {
            EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
            while(endpoint != null && buffer.remaining() >= _maxFrameSize)
            {
                SessionImpl session;
                TransportSession transportSession;

                if((endpoint instanceof SessionImpl)
                   && (session = (SessionImpl)endpoint).getLocalState() == EndpointState.CLOSED
                   && (transportSession = session.getTransportSession()).isLocalChannelSet()
                   && !hasSendableMessages(session))
                {
                    int channel = freeLocalChannel(transportSession);
                    End end = new End();
                    int frameBytes = writeFrame(buffer, channel, end, null, null);
                    written += frameBytes;
                    endpoint.clearModified();
                }

                endpoint = endpoint.transportNext();

            }
        }
        return written;
    }

    private boolean hasSendableMessages(SessionImpl session)
    {

        if(!_closeReceived && (session == null || !session.getTransportSession().endReceived()))
        {
            EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
            while(endpoint != null)
            {
                if(endpoint instanceof SenderImpl)
                {
                    SenderImpl sender = (SenderImpl) endpoint;
                    if((session == null || sender.getSession() == session)
                       && sender.getQueued() != 0
                        && !getTransportState(sender).detachReceived())
                    {
                        return true;
                    }
                }
                endpoint = endpoint.transportNext();
            }
        }
        return false;
    }

    private int processClose(WritableBuffer buffer)
    {
        if(_connectionEndpoint != null && _connectionEndpoint.getLocalState() == EndpointState.CLOSED && !_isCloseSent)
        {
            if(!hasSendableMessages(null))
            {
                Close close = new Close();

                // TODO - populate;

                _isCloseSent = true;

                return  writeFrame(buffer, 0, close, null, null);
            }
        }
        return 0;

    }

    private int writeFrame(WritableBuffer buffer,
                           int channel,
                           FrameBody frameBody,
                           ByteBuffer payload,
                           Runnable onPayloadTooLarge)
    {
        int oldPosition = buffer.position();
        buffer.position(oldPosition+8);
        _encoder.setByteBuffer(buffer);

        _encoder.writeObject(frameBody);

        if(payload != null && (payload.remaining() + buffer.position() - oldPosition) > _maxFrameSize)
        {
            if(onPayloadTooLarge != null)
            {
                onPayloadTooLarge.run();
            }
            buffer.position(oldPosition+8);
            _encoder.writeObject(frameBody);
        }

        if( _protocolTracer!=null )
        {
            ByteBuffer originalPayload = null;
            if( payload!=null )
            {
                originalPayload = payload.duplicate();
            }
            _protocolTracer.sentFrame(new TransportFrame(channel, (FrameBody) frameBody, Binary.create(originalPayload)));
        }

        int payloadSize = Math.min(payload == null ? 0 : payload.remaining(), _maxFrameSize - (buffer.position() - oldPosition));
        if(payloadSize > 0)
        {
            int oldLimit = payload.limit();
            payload.limit(payload.position() + payloadSize);
            buffer.put(payload);
            payload.limit(oldLimit);
        }
        int frameSize = buffer.position() - oldPosition;
        int limit = buffer.position();
        buffer.position(oldPosition);
        buffer.putInt(frameSize);
        buffer.put((byte) 2);
        buffer.put(AMQP_FRAME_TYPE);
        buffer.putShort((short) channel);
        buffer.position(limit);

        return frameSize;
    }

    //==================================================================================================================

    @Override
    protected ConnectionImpl getConnectionImpl()
    {
        return _connectionEndpoint;
    }

    @Override
    public void free()
    {
        super.free();
    }

    //==================================================================================================================
    // handle incoming amqp data


    @Override
    public void handleOpen(Open open, Binary payload, Integer channel)
    {
        setRemoteState(EndpointState.ACTIVE);
        if(_connectionEndpoint != null)
        {
            _connectionEndpoint.handleOpen(open);
        }
        else
        {
            _open = open;
        }

        if(open.getMaxFrameSize().longValue() > 0 && open.getMaxFrameSize().longValue() < (long) _maxFrameSize)
        {
            _maxFrameSize = (int) open.getMaxFrameSize().longValue();
        }
    }

    @Override
    public void handleBegin(Begin begin, Binary payload, Integer channel)
    {
        // TODO - check channel < max_channel
        TransportSession transportSession = _remoteSessions[channel];
        if(transportSession != null)
        {
            // TODO - fail due to begin on begun session
        }
        else
        {
            SessionImpl session;
            if(begin.getRemoteChannel() == null)
            {
                session = _connectionEndpoint.session();
                transportSession = getTransportState(session);
            }
            else
            {
                // TODO check null
                transportSession = _localSessions[begin.getRemoteChannel().intValue()];
                session = transportSession.getSession();

            }
            transportSession.setRemoteChannel(channel);
            session.setRemoteState(EndpointState.ACTIVE);
            transportSession.setNextIncomingId(begin.getNextOutgoingId());
            _remoteSessions[channel] = transportSession;


        }

    }

    @Override
    public void handleAttach(Attach attach, Binary payload, Integer channel)
    {
        TransportSession transportSession = _remoteSessions[channel];
        if(transportSession == null)
        {
            // TODO - fail due to attach on non-begun session
        }
        else
        {
            SessionImpl session = transportSession.getSession();
            TransportLink transportLink = transportSession.getLinkFromRemoteHandle(attach.getHandle());
            LinkImpl link = null;

            if(transportLink != null)
            {
                // TODO - fail - attempt attach on a handle which is in use
            }
            else
            {
                transportLink = transportSession.resolveHalfOpenLink(attach.getName());
                if(transportLink == null)
                {

                    link = (attach.getRole() == Role.RECEIVER)
                           ? session.sender(attach.getName())
                           : session.receiver(attach.getName());
                    transportLink = getTransportState(link);
                }
                else
                {
                    link = transportLink.getLink();
                }
                if(attach.getRole() == Role.SENDER)
                {
                    transportLink.setDeliveryCount(attach.getInitialDeliveryCount());
                }

                link.setRemoteState(EndpointState.ACTIVE);
                link.setRemoteSource(attach.getSource());
                link.setRemoteTarget(attach.getTarget());

                link.setRemoteReceiverSettleMode(attach.getRcvSettleMode());
                link.setRemoteSenderSettleMode(attach.getSndSettleMode());

                transportLink.setName(attach.getName());
                transportLink.setRemoteHandle(attach.getHandle());
                transportSession.addLinkRemoteHandle(transportLink, attach.getHandle());

            }
        }
    }

    @Override
    public void handleFlow(Flow flow, Binary payload, Integer channel)
    {
        TransportSession transportSession = _remoteSessions[channel];
        if(transportSession == null)
        {
            // TODO - fail due to attach on non-begun session
        }
        else
        {
            transportSession.handleFlow(flow);
        }

    }

    @Override
    public void handleTransfer(Transfer transfer, Binary payload, Integer channel)
    {
        // TODO - check channel < max_channel
        TransportSession transportSession = _remoteSessions[channel];
        if(transportSession != null)
        {
            transportSession.handleTransfer(transfer, payload);
        }
        else
        {
            // TODO - fail due to begin on begun session
        }
    }

    @Override
    public void handleDisposition(Disposition disposition, Binary payload, Integer channel)
    {
        TransportSession transportSession = _remoteSessions[channel];
        if(transportSession == null)
        {
            // TODO - fail due to attach on non-begun session
        }
        else
        {
            transportSession.handleDisposition(disposition);
        }
    }

    @Override
    public void handleDetach(Detach detach, Binary payload, Integer channel)
    {
        TransportSession transportSession = _remoteSessions[channel];
        if(transportSession == null)
        {
            // TODO - fail due to attach on non-begun session
        }
        else
        {
            TransportLink transportLink = transportSession.getLinkFromRemoteHandle(detach.getHandle());

            if(transportLink != null)
            {
                LinkImpl link = transportLink.getLink();
                transportLink.receivedDetach();
                transportSession.freeRemoteHandle(transportLink.getRemoteHandle());
                link.setRemoteState(EndpointState.CLOSED);

            }
            else
            {
                // TODO - fail - attempt attach on a handle which is in use
            }
        }
    }

    @Override
    public void handleEnd(End end, Binary payload, Integer channel)
    {
        TransportSession transportSession = _remoteSessions[channel];
        if(transportSession == null)
        {
            // TODO - fail due to attach on non-begun session
        }
        else
        {
            _remoteSessions[channel] = null;
            transportSession.receivedEnd();
            transportSession.getSession().setRemoteState(EndpointState.CLOSED);

        }
    }

    @Override
    public void handleClose(Close close, Binary payload, Integer channel)
    {
        _closeReceived = true;
        setRemoteState(EndpointState.CLOSED);
        if(_connectionEndpoint != null)
        {
            _connectionEndpoint.setRemoteState(EndpointState.CLOSED);
        }

    }

    @Override
    public boolean input(TransportFrame frame)
    {
        if( _protocolTracer!=null )
        {
            _protocolTracer.receivedFrame(frame);
        }
        if(_connectionEndpoint != null || getRemoteState() == EndpointState.UNINITIALIZED)
        {
            frame.getBody().invoke(this,frame.getPayload(), frame.getChannel());
            return true;
        }
        else
        {
            return false;
        }
    }

    private static class PartialTransfer implements Runnable
    {
        private final Transfer _transfer;

        public PartialTransfer(Transfer transfer)
        {
            _transfer = transfer;
        }

        public void run()
        {
            _transfer.setMore(true);
        }
    }

    @Override
    public ProtocolTracer getProtocolTracer()
    {
        return _protocolTracer;
    }

    @Override
    public void setProtocolTracer(ProtocolTracer protocolTracer)
    {
        this._protocolTracer = protocolTracer;
    }

}