/*
 *
 * 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.server.protocol.v1_0;

import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.security.AccessController;
import java.security.Principal;
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import javax.security.sasl.SaslException;
import javax.security.sasl.SaslServer;

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

import org.apache.qpid.server.model.AuthenticationProvider;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.protocol.v1_0.codec.DescribedTypeConstructorRegistry;
import org.apache.qpid.server.protocol.v1_0.codec.FrameWriter;
import org.apache.qpid.server.protocol.v1_0.codec.ProtocolHandler;
import org.apache.qpid.server.protocol.v1_0.codec.ValueWriter;
import org.apache.qpid.server.protocol.v1_0.framing.AMQFrame;
import org.apache.qpid.server.protocol.v1_0.framing.FrameHandler;
import org.apache.qpid.server.protocol.v1_0.framing.OversizeFrameException;
import org.apache.qpid.server.protocol.v1_0.framing.SASLFrame;
import org.apache.qpid.server.protocol.v1_0.framing.TransportFrame;
import org.apache.qpid.server.protocol.v1_0.type.Binary;
import org.apache.qpid.server.protocol.v1_0.type.FrameBody;
import org.apache.qpid.server.protocol.v1_0.type.SaslFrameBody;
import org.apache.qpid.server.protocol.v1_0.type.Symbol;
import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.common.ServerPropertyNames;
import org.apache.qpid.configuration.CommonProperties;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
import org.apache.qpid.server.protocol.v1_0.type.UnsignedShort;
import org.apache.qpid.server.protocol.v1_0.type.codec.AMQPDescribedTypeRegistry;
import org.apache.qpid.server.protocol.v1_0.type.security.SaslChallenge;
import org.apache.qpid.server.protocol.v1_0.type.security.SaslCode;
import org.apache.qpid.server.protocol.v1_0.type.security.SaslInit;
import org.apache.qpid.server.protocol.v1_0.type.security.SaslMechanisms;
import org.apache.qpid.server.protocol.v1_0.type.security.SaslOutcome;
import org.apache.qpid.server.protocol.v1_0.type.security.SaslResponse;
import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
import org.apache.qpid.server.protocol.v1_0.type.transport.Begin;
import org.apache.qpid.server.protocol.v1_0.type.transport.Close;
import org.apache.qpid.server.protocol.v1_0.type.transport.ConnectionError;
import org.apache.qpid.server.protocol.v1_0.type.transport.Detach;
import org.apache.qpid.server.protocol.v1_0.type.transport.Disposition;
import org.apache.qpid.server.protocol.v1_0.type.transport.End;
import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
import org.apache.qpid.server.protocol.v1_0.type.transport.Flow;
import org.apache.qpid.server.protocol.v1_0.type.transport.Open;
import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
import org.apache.qpid.server.security.SubjectCreator;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.security.auth.AuthenticationResult;
import org.apache.qpid.server.security.auth.SubjectAuthenticationResult;
import org.apache.qpid.server.security.auth.manager.AnonymousAuthenticationManager;
import org.apache.qpid.server.security.auth.manager.ExternalAuthenticationManagerImpl;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.transport.AbstractAMQPConnection;
import org.apache.qpid.server.transport.ProtocolEngine;
import org.apache.qpid.server.transport.ServerNetworkConnection;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException;
import org.apache.qpid.transport.ByteBufferSender;
import org.apache.qpid.server.transport.AggregateTicker;

public class AMQPConnection_1_0 extends AbstractAMQPConnection<AMQPConnection_1_0>
        implements FrameOutputHandler, DescribedTypeConstructorRegistry.Source,
                   ValueWriter.Registry.Source,
                   ErrorHandler,
                   SASLEndpoint,
                   ConnectionHandler
{

    private static Logger LOGGER = LoggerFactory.getLogger(AMQPConnection_1_0.class);
    private static final Logger FRAME_LOGGER = LoggerFactory.getLogger("FRM");

    private static final long CLOSE_RESPONSE_TIMEOUT = 10000L;

    private final AtomicBoolean _stateChanged = new AtomicBoolean();
    private final AtomicReference<Action<ProtocolEngine>> _workListener = new AtomicReference<>();


    private static final byte[] SASL_HEADER = new byte[]
            {
                    (byte) 'A',
                    (byte) 'M',
                    (byte) 'Q',
                    (byte) 'P',
                    (byte) 3,
                    (byte) 1,
                    (byte) 0,
                    (byte) 0
            };

    private static final byte[] AMQP_HEADER = new byte[]
            {
                    (byte) 'A',
                    (byte) 'M',
                    (byte) 'Q',
                    (byte) 'P',
                    (byte) 0,
                    (byte) 1,
                    (byte) 0,
                    (byte) 0
            };

    private FrameWriter _frameWriter;
    private ProtocolHandler _frameHandler;
    private volatile boolean _transportBlockedForWriting;
    private volatile SubjectAuthenticationResult _successfulAuthenticationResult;

    private enum FrameReceivingState
    {
        AMQP_OR_SASL_HEADER,
        SASL_INIT_ONLY,
        SASL_RESPONSE_ONLY,
        AMQP_HEADER,
        OPEN_ONLY,
        ANY_FRAME,
        CLOSED
    }

    private volatile FrameReceivingState _frameReceivingState = FrameReceivingState.AMQP_OR_SASL_HEADER;

    private static final short CONNECTION_CONTROL_CHANNEL = (short) 0;
    private static final QpidByteBuffer EMPTY_BYTE_BUFFER = QpidByteBuffer.wrap(new byte[0]);

    private static final int DEFAULT_CHANNEL_MAX = Math.min(Integer.getInteger("amqp.channel_max", 255), 0xFFFF);
    private static final int DEFAULT_MAX_FRAME = Integer.getInteger("amqp.max_frame_size", 1 << 15);

    private AmqpPort<?> _port;
    private SubjectCreator _subjectCreator;
    private Transport _transport;
    private long _connectionId;

    private Container _container;

    private int _channelMax = DEFAULT_CHANNEL_MAX;
    private int _maxFrameSize = 4096;
    private String _remoteContainerId;

    private SocketAddress _remoteAddress;

    // positioned by the *outgoing* channel
    private Session_1_0[] _sendingSessions;

    // positioned by the *incoming* channel
    private Session_1_0[] _receivingSessions;
    private boolean _closedForInput;
    private boolean _closedForOutput;

    private long _idleTimeout;

    private ConnectionState _connectionState = ConnectionState.UNOPENED;

    private AMQPDescribedTypeRegistry _describedTypeRegistry = AMQPDescribedTypeRegistry.newInstance()
            .registerTransportLayer()
            .registerMessagingLayer()
            .registerTransactionLayer()
            .registerSecurityLayer();


    private Map _properties;
    private SaslServerProvider _saslServerProvider;
    private boolean _saslComplete;

    private SaslServer _saslServer;
    private String _localHostname;
    private long _desiredIdleTimeout;
    private UnsignedInteger _handleMax = UnsignedInteger.MAX_VALUE;
    private Error _remoteError;

    private static final long MINIMUM_SUPPORTED_IDLE_TIMEOUT = 1000L;

    private Map _remoteProperties;

    private final AtomicBoolean _orderlyClose = new AtomicBoolean(false);

    private final Collection<Session_1_0>
            _sessions = Collections.synchronizedCollection(new ArrayList<Session_1_0>());

    private final Object _reference = new Object();

    private final Queue<Action<? super ConnectionHandler>> _asyncTaskList =
            new ConcurrentLinkedQueue<>();

    private boolean _closedOnOpen;


    AMQPConnection_1_0(final Broker<?> broker,
                       final ServerNetworkConnection network,
                       AmqpPort<?> port, Transport transport, long id,
                       final AggregateTicker aggregateTicker,
                       final boolean useSASL)
    {
        super(broker, network, port, transport, Protocol.AMQP_1_0, id, aggregateTicker);
        _container = new Container(broker.getId().toString());

        _subjectCreator = port.getAuthenticationProvider().getSubjectCreator(transport.isSecure());

        _saslServerProvider = useSASL ? asSaslServerProvider(_subjectCreator, network) : null;
        _port = port;
        _transport = transport;
        _connectionId = id;

        Map<Symbol,Object> serverProperties = new LinkedHashMap<>();
        serverProperties.put(Symbol.valueOf(ServerPropertyNames.PRODUCT), CommonProperties.getProductName());
        serverProperties.put(Symbol.valueOf(ServerPropertyNames.VERSION), CommonProperties.getReleaseVersion());
        serverProperties.put(Symbol.valueOf(ServerPropertyNames.QPID_BUILD), CommonProperties.getBuildVersion());
        serverProperties.put(Symbol.valueOf(ServerPropertyNames.QPID_INSTANCE_NAME), broker.getName());

        setProperties(serverProperties);

        setRemoteAddress(network.getRemoteAddress());

        setDesiredIdleTimeout(1000L * broker.getConnection_heartBeatDelay());

        _frameWriter =  new FrameWriter(getDescribedTypeRegistry(), getSender());


    }


    private void setUserPrincipal(final Principal user)
    {
        setSubject(_subjectCreator.createSubjectWithGroups(user));
    }

    private long getDesiredIdleTimeout()
    {
        return _desiredIdleTimeout;
    }

    public void receiveAttach(final short channel, final Attach attach)
    {
        assertState(FrameReceivingState.ANY_FRAME);
        final Session_1_0 session = getSession(channel);
        if (session != null)
        {
            AccessController.doPrivileged(new PrivilegedAction<Object>()
            {
                @Override
                public Object run()
                {
                    session.receiveAttach(attach);
                    return null;
                }
            }, session.getAccessControllerContext());
        }
        else
        {
            // TODO - error
        }
    }

    public void receive(final short channel, final Object frame)
    {
        FRAME_LOGGER.debug("RECV[{}|{}] : {}", _remoteAddress, channel, frame);
        if (frame instanceof FrameBody)
        {
            ((FrameBody) frame).invoke(channel, this);
        }
        else if (frame instanceof SaslFrameBody)
        {
            ((SaslFrameBody) frame).invoke(channel, this);
        }
    }

    private void closeSaslWithFailure()
    {
        _saslComplete = true;
        _frameReceivingState = FrameReceivingState.CLOSED;
        setClosedForInput(true);
        close();
    }

    public void receiveSaslChallenge(final SaslChallenge saslChallenge)
    {
        // TODO - log unexpected frame
        closeSaslWithFailure();
    }

    public void receiveClose(final short channel, final Close close)
    {
        assertState(FrameReceivingState.ANY_FRAME);
        _frameReceivingState = FrameReceivingState.CLOSED;
        setClosedForInput(true);
        closeReceived();
        switch (_connectionState)
        {
            case UNOPENED:
            case AWAITING_OPEN:
                Error error = new Error();
                error.setCondition(ConnectionError.CONNECTION_FORCED);
                error.setDescription("Connection close sent before connection was opened");
                closeConnection(error);
                break;
            case OPEN:
                _connectionState = ConnectionState.CLOSE_RECEIVED;
                // TODO - we should log the error we received from the client if present
                sendClose(new Close());
                _connectionState = ConnectionState.CLOSED;
                _orderlyClose.set(true);
                break;
            case CLOSE_SENT:
                _connectionState = ConnectionState.CLOSED;
                _orderlyClose.set(true);

            default:
        }
        _remoteError = close.getError();

    }

    private void closeReceived()
    {
        Collection<Session_1_0> sessions = new ArrayList<>(_sessions);

        for(final Session_1_0 session : sessions)
        {
            AccessController.doPrivileged(new PrivilegedAction<Object>()
            {
                @Override
                public Object run()
                {
                    session.remoteEnd(new End());
                    return null;
                }
            }, session.getAccessControllerContext());
        }
    }

    private void setClosedForInput(final boolean closed)
    {
        _closedForInput = closed;
    }

    public void receiveSaslMechanisms(final SaslMechanisms saslMechanisms)
    {
        // TODO - log unexpected frame
        closeSaslWithFailure();
    }

    public void receiveSaslResponse(final SaslResponse saslResponse)
    {
        final Binary responseBinary = saslResponse.getResponse();
        byte[] response = responseBinary == null ? new byte[0] : responseBinary.getArray();

        assertState(FrameReceivingState.SASL_RESPONSE_ONLY);

        processSaslResponse(response);
    }

    public AMQPDescribedTypeRegistry getDescribedTypeRegistry()
    {
        return _describedTypeRegistry;
    }

    private void closeSessionAsync(final Session_1_0 session, final AMQConstant cause, final String message)
    {
        addAsyncTask(new Action<ConnectionHandler>()
        {
            @Override
            public void performAction(final ConnectionHandler object)
            {
                session.close(cause, message);
            }
        });
    }


    private boolean closedForOutput()
    {
        return _closedForOutput;
    }

    public boolean isClosed()
    {
        return _connectionState == ConnectionState.CLOSED
               || _connectionState == ConnectionState.CLOSE_RECEIVED;
    }

    public boolean closedForInput()
    {
        return _closedForInput;
    }

    void sessionEnded(final Session_1_0 session)
    {
        if(!_closedOnOpen)
        {
            _sessions.remove(session);
            sessionRemoved(session);
        }
    }

    public int send(final short channel, final FrameBody body, final QpidByteBuffer payload)
    {
        return sendFrame(channel, body, payload);
    }

    private void inputClosed()
    {
        List<Runnable> postLockActions;

        if (!_closedForInput)
        {
            _closedForInput = true;
            FRAME_LOGGER.debug("RECV[{}] : {}", _remoteAddress, "Underlying connection closed");
            switch (_connectionState)
            {
                case UNOPENED:
                case AWAITING_OPEN:
                case CLOSE_SENT:
                    _connectionState = ConnectionState.CLOSED;
                    closeSender();
                    break;
                case OPEN:
                    _connectionState = ConnectionState.CLOSE_RECEIVED;
                case CLOSED:
                    // already sent our close - too late to do anything more
                    break;
                default:
            }
            closeReceived();
        }


    }

    private void closeSender()
    {
        setClosedForOutput(true);
        close();
    }

    String getRemoteContainerId()
    {
        return _remoteContainerId;
    }

    private void setDesiredIdleTimeout(final long desiredIdleTimeout)
    {
        _desiredIdleTimeout = desiredIdleTimeout;
    }

    public boolean isOpen()
    {
        return _connectionState == ConnectionState.OPEN;
    }

    void sendEnd(final short channel, final End end, final boolean remove)
    {
        sendFrame(channel, end);
        if (remove)
        {
            _sendingSessions[channel] = null;
        }

    }

    public void receiveSaslOutcome(final SaslOutcome saslOutcome)
    {
        // TODO - log unexpected frame
        closeSaslWithFailure();
    }

    public void receiveEnd(final short channel, final End end)
    {

        assertState(FrameReceivingState.ANY_FRAME);
        Session_1_0 endpoint = _receivingSessions[channel];
        if (endpoint != null)
        {
            _receivingSessions[channel] = null;

            endpoint.receiveEnd(end);
        }
        else
        {
            // TODO error
        }
    }

    public void receiveDisposition(final short channel,
                                   final Disposition disposition)
    {
        assertState(FrameReceivingState.ANY_FRAME);
        final Session_1_0 session = getSession(channel);
        if (session != null)
        {
            AccessController.doPrivileged(new PrivilegedAction<Object>()
            {
                @Override
                public Object run()
                {
                    session.receiveDisposition(disposition);
                    return null;
                }
            }, session.getAccessControllerContext());
        }
        else
        {
            // TODO - error
        }

    }

    public void receiveBegin(final short channel, final Begin begin)
    {

        assertState(FrameReceivingState.ANY_FRAME);
        short myChannelId;
        if (begin.getRemoteChannel() != null)
        {
            final Error error = new Error();
            error.setCondition(ConnectionError.FRAMING_ERROR);
            error.setDescription("BEGIN received on channel " + channel + " with given remote-channel "
                                 + begin.getRemoteChannel() + ". Since the broker does not spontaneously start channels, this must be an error.");
            closeConnection(error);

        }
        else // Peer requesting session creation
        {

            if (_receivingSessions[channel] == null)
            {
                myChannelId = getFirstFreeChannel();
                if (myChannelId == -1)
                {
                    final Error error = new Error();
                    error.setCondition(ConnectionError.FRAMING_ERROR);
                    error.setDescription("BEGIN received on channel " + channel + ". There are no free channels for the broker to responsd on.");
                    closeConnection(error);

                }
                Session_1_0 session = new Session_1_0(this, begin);

                _receivingSessions[channel] = session;
                _sendingSessions[myChannelId] = session;

                Begin beginToSend = new Begin();

                session.setReceivingChannel(channel);
                session.setSendingChannel(myChannelId);
                beginToSend.setRemoteChannel(UnsignedShort.valueOf(channel));
                beginToSend.setNextOutgoingId(session.getNextOutgoingId());
                beginToSend.setOutgoingWindow(session.getOutgoingWindowSize());
                beginToSend.setIncomingWindow(session.getIncomingWindowSize());
                sendFrame(myChannelId, beginToSend);

                _sessions.add(session);
                sessionAdded(session);

            }
            else
            {
                final Error error = new Error();
                error.setCondition(ConnectionError.FRAMING_ERROR);
                error.setDescription("BEGIN received on channel " + channel + " which is already in use.");
                closeConnection(error);
            }

        }

    }

    private short getFirstFreeChannel()
    {
        for (int i = 0; i <= _channelMax; i++)
        {
            if (_sendingSessions[i] == null)
            {
                return (short) i;
            }
        }
        return -1;
    }

    public void handleError(final Error error)
    {
        if (!closedForOutput())
        {
            Close close = new Close();
            close.setError(error);
            sendFrame((short) 0, close);

            setClosedForOutput(true);
        }

    }

    public void receiveTransfer(final short channel, final Transfer transfer)
    {
        assertState(FrameReceivingState.ANY_FRAME);
        final Session_1_0 session = getSession(channel);
        if (session != null)
        {
            AccessController.doPrivileged(new PrivilegedAction<Object>()
            {
                @Override
                public Object run()
                {
                    session.receiveTransfer(transfer);
                    return null;
                }
            }, session.getAccessControllerContext());
        }
        else
        {
            // TODO - error
        }
    }

    public void receiveFlow(final short channel, final Flow flow)
    {
        assertState(FrameReceivingState.ANY_FRAME);
        final Session_1_0 session = getSession(channel);
        if (session != null)
        {
            AccessController.doPrivileged(new PrivilegedAction<Object>()
            {
                @Override
                public Object run()
                {
                    session.receiveFlow(flow);
                    return null;
                }
            }, session.getAccessControllerContext());
        }
        else
        {
            // TODO - error
        }

    }

    public void receiveOpen(final short channel, final Open open)
    {
        assertState(FrameReceivingState.OPEN_ONLY);
        _frameReceivingState = FrameReceivingState.ANY_FRAME;
        _channelMax = open.getChannelMax() == null ? _channelMax
                : open.getChannelMax().intValue() < _channelMax
                        ? open.getChannelMax().intValue()
                        : _channelMax;
        if (_receivingSessions == null)
        {
            _receivingSessions = new Session_1_0[_channelMax + 1];
            _sendingSessions = new Session_1_0[_channelMax + 1];
        }
        _maxFrameSize = open.getMaxFrameSize() == null ? DEFAULT_MAX_FRAME : open.getMaxFrameSize().intValue();
        _remoteContainerId = open.getContainerId();
        _localHostname = open.getHostname();
        if (open.getIdleTimeOut() != null)
        {
            _idleTimeout = open.getIdleTimeOut().longValue();
        }
        _remoteProperties = open.getProperties();
        if (_remoteProperties != null)
        {
            if (_remoteProperties.containsKey(Symbol.valueOf("product")))
            {
                setClientProduct(_remoteProperties.get(Symbol.valueOf("product")).toString());
            }
            if (_remoteProperties.containsKey(Symbol.valueOf("version")))
            {
                setClientVersion(_remoteProperties.get(Symbol.valueOf("version")).toString());
            }
            setClientId(_remoteContainerId);
        }
        if (_idleTimeout != 0L && _idleTimeout < MINIMUM_SUPPORTED_IDLE_TIMEOUT)
        {
            closeConnection(new Error(ConnectionError.CONNECTION_FORCED,
                                      "Requested idle timeout of "
                                      + _idleTimeout
                                      + " is too low. The minimum supported timeout is"
                                      + MINIMUM_SUPPORTED_IDLE_TIMEOUT));
            close();
            _closedOnOpen = true;
        }
        else
        {
            long desiredIdleTimeout = getDesiredIdleTimeout();
            initialiseHeartbeating(_idleTimeout / 2L, desiredIdleTimeout);
            final NamedAddressSpace addressSpace = ((AmqpPort) _port).getAddressSpace(_localHostname);
            if (addressSpace == null)
            {
                closeWithError(AmqpError.NOT_FOUND, "Unknown hostname in connection open: '" + _localHostname + "'");
            }
            else
            {
                if (!addressSpace.isActive())
                {
                    final Error err = new Error();
                    err.setCondition(AmqpError.NOT_FOUND);
                    closeConnection(err);

                    _closedOnOpen = true;

                    populateConnectionRedirect(addressSpace, err);

                    closeConnection(err);

                    close();

                    _closedOnOpen = true;

                }
                else
                {
                    if (AuthenticatedPrincipal.getOptionalAuthenticatedPrincipalFromSubject(getSubject()) == null)
                    {
                        closeWithError(AmqpError.NOT_ALLOWED, "Connection has not been authenticated");
                    }
                    else
                    {
                        try
                        {
                            setAddressSpace(addressSpace);
                        }
                        catch (VirtualHostUnavailableException e)
                        {
                            closeWithError(AmqpError.NOT_ALLOWED, e.getMessage());
                        }
                    }
                }
            }
        }
        switch (_connectionState)
        {
            case UNOPENED:
                sendOpen(_channelMax, _maxFrameSize);
            case AWAITING_OPEN:
                _connectionState = ConnectionState.OPEN;
            default:
                // TODO bad stuff (connection already open)

        }

    }

    private void populateConnectionRedirect(final NamedAddressSpace addressSpace, final Error err)
    {
        final String redirectHost = addressSpace.getRedirectHost(((AmqpPort) _port));

        if(redirectHost == null)
        {
            err.setDescription("Virtual host '" + _localHostname + "' is not active");
        }
        else
        {
            String networkHost;
            int port;
            if(redirectHost.matches("\\[[0-9a-f:]+\\](:[0-9]+)?"))
            {
                // IPv6 case
                networkHost = redirectHost.substring(1, redirectHost.indexOf("]"));
                if(redirectHost.contains("]:"))
                {
                    port = Integer.parseInt(redirectHost.substring(redirectHost.indexOf("]")+2));
                }
                else
                {
                    port = -1;
                }
            }
            else
            {
                if(redirectHost.contains(":"))
                {
                    networkHost = redirectHost.substring(0, redirectHost.lastIndexOf(":"));
                    try
                    {
                        String portString = redirectHost.substring(redirectHost.lastIndexOf(":")+1);
                        port = Integer.parseInt(portString);
                    }
                    catch (NumberFormatException e)
                    {
                        port = -1;
                    }
                }
                else
                {
                    networkHost = redirectHost;
                    port = -1;
                }
            }
            final Map<Symbol, Object> infoMap = new HashMap<>();
            infoMap.put(Symbol.valueOf("network-host"), networkHost);
            if(port > 0)
            {
                infoMap.put(Symbol.valueOf("port"), UnsignedInteger.valueOf(port));
            }
            err.setInfo(infoMap);
        }
    }

    public void receiveDetach(final short channel, final Detach detach)
    {
        assertState(FrameReceivingState.ANY_FRAME);
        final Session_1_0 session = getSession(channel);
        if (session != null)
        {
            AccessController.doPrivileged(new PrivilegedAction<Object>()
            {
                @Override
                public Object run()
                {
                    session.receiveDetach(detach);
                    return null;
                }
            }, session.getAccessControllerContext());
        }
        else
        {
            // TODO - error
        }
    }

    private void transportStateChanged()
    {
        for (Session_1_0 session : _sessions)
        {
            session.transportStateChanged();
        }
    }

    public void close(final Error error)
    {
        closeConnection(error);
    }

    private void setRemoteAddress(final SocketAddress remoteAddress)
    {
        _remoteAddress = remoteAddress;
    }

    public void setProperties(final Map<Symbol, Object> properties)
    {
        _properties = properties;
    }

    private void setClosedForOutput(final boolean closed)
    {
        _closedForOutput = closed;
    }

    public void receiveSaslInit(final SaslInit saslInit)
    {
        assertState(FrameReceivingState.SASL_INIT_ONLY);
        String mechanism = saslInit.getMechanism() == null ? null : saslInit.getMechanism().toString();
        final Binary initialResponse = saslInit.getInitialResponse();
        byte[] response = initialResponse == null ? new byte[0] : initialResponse.getArray();


        try
        {
            _saslServer = _saslServerProvider.getSaslServer(mechanism, "localhost");
            processSaslResponse(response);
        }
        catch (SaslException e)
        {
            handleSaslError();
        }
    }

    private void processSaslResponse(final byte[] response)
    {
        byte[] challenge = null;
        SubjectAuthenticationResult authenticationResult = _successfulAuthenticationResult;
        if (authenticationResult == null)
        {
            authenticationResult = _subjectCreator.authenticate(_saslServer, response != null ? response : new byte[0]);
            challenge = authenticationResult.getChallenge();
        }

        if (authenticationResult.getStatus() == AuthenticationResult.AuthenticationStatus.SUCCESS)
        {
            _successfulAuthenticationResult = authenticationResult;
            if (challenge == null || challenge.length == 0)
            {
                setSubject(_successfulAuthenticationResult.getSubject());
                SaslOutcome outcome = new SaslOutcome();
                outcome.setCode(SaslCode.OK);
                send(new SASLFrame(outcome), null);
                _saslComplete = true;
                _frameReceivingState = FrameReceivingState.AMQP_HEADER;
            }
            else
            {
                continueSaslNegotiation(challenge);
            }
        }
        else if(authenticationResult.getStatus() == AuthenticationResult.AuthenticationStatus.CONTINUE)
        {
            continueSaslNegotiation(challenge);
        }
        else
        {
            handleSaslError();
        }
    }

    private void continueSaslNegotiation(final byte[] challenge)
    {
        SaslChallenge challengeBody = new SaslChallenge();
        challengeBody.setChallenge(new Binary(challenge));
        send(new SASLFrame(challengeBody), null);

        _frameReceivingState = FrameReceivingState.SASL_RESPONSE_ONLY;
    }

    private void handleSaslError()
    {
        SaslOutcome outcome = new SaslOutcome();
        outcome.setCode(SaslCode.AUTH);
        send(new SASLFrame(outcome), null);
        _saslComplete = true;
        closeSaslWithFailure();
    }

    public int getMaxFrameSize()
    {
        return _maxFrameSize;
    }

    Object getReference()
    {
        return _reference;
    }

    private void endpointClosed()
    {
        try
        {
            performDeleteTasks();
            closeReceived();
        }
        finally
        {
            NamedAddressSpace virtualHost = getAddressSpace();
            if (virtualHost != null)
            {
                virtualHost.deregisterConnection(this);
            }
        }
    }

    private void closeConnection()
    {
        switch (_connectionState)
        {
            case AWAITING_OPEN:
            case OPEN:
                Close closeToSend = new Close();
                sendClose(closeToSend);
                _connectionState = ConnectionState.CLOSE_SENT;
                break;
            case CLOSE_SENT:
            default:
        }
    }

    private void closeConnection(final Error error)
    {
        Close close = new Close();
        close.setError(error);
        switch (_connectionState)
        {
            case UNOPENED:
                sendOpen(0, 0);
                sendClose(close);
                _connectionState = ConnectionState.CLOSED;
                break;
            case AWAITING_OPEN:
            case OPEN:
                sendClose(close);
                _connectionState = ConnectionState.CLOSE_SENT;
            case CLOSE_SENT:
            case CLOSED:
                // already sent our close - too late to do anything more
                break;
            default:
                // TODO Unknown state
        }
    }

    int sendFrame(final short channel, final FrameBody body, final QpidByteBuffer payload)
    {
        if (!_closedForOutput)
        {
            ValueWriter<FrameBody> writer = _describedTypeRegistry.getValueWriter(body);
            int size = writer.writeToBuffer(EMPTY_BYTE_BUFFER);
            QpidByteBuffer payloadDup = payload == null ? null : payload.duplicate();
            int payloadSent = _maxFrameSize - (size + 9);
            try
            {
                if (payloadSent < (payload == null ? 0 : payload.remaining()))
                {

                    if (body instanceof Transfer)
                    {
                        ((Transfer) body).setMore(Boolean.TRUE);
                    }

                    writer = _describedTypeRegistry.getValueWriter(body);
                    size = writer.writeToBuffer(EMPTY_BYTE_BUFFER);
                    payloadSent = _maxFrameSize - (size + 9);

                    payloadDup.limit(payloadDup.position() + payloadSent);
                }
                else
                {
                    payloadSent = payload == null ? 0 : payload.remaining();
                }
                send(AMQFrame.createAMQFrame(channel, body, payloadDup));
            }
            finally
            {
                if (payloadDup != null)
                {
                    payloadDup.dispose();
                }
            }
            return payloadSent;
        }
        else
        {
            return -1;
        }
    }

    void sendFrame(final short channel, final FrameBody body)
    {
        sendFrame(channel, body, null);
    }

    public ByteBufferSender getSender()
    {
        return getNetwork().getSender();
    }

    @Override
    public void writerIdle()
    {
        send(TransportFrame.createAMQFrame((short)0,null));
    }

    @Override
    public void readerIdle()
    {
        AccessController.doPrivileged(new PrivilegedAction<Object>()
        {
            @Override
            public Object run()
            {
                getEventLogger().message(ConnectionMessages.IDLE_CLOSE("", false));
                getNetwork().close();
                return null;
            }
        }, getAccessControllerContext());
    }

    @Override
    public void encryptedTransport()
    {
    }

    private static SaslServerProvider asSaslServerProvider(final SubjectCreator subjectCreator,
                                                           final ServerNetworkConnection network)
    {
        return new SaslServerProvider()
        {
            @Override
            public SaslServer getSaslServer(String mechanism, String fqdn) throws SaslException
            {
                return subjectCreator.createSaslServer(mechanism, fqdn, network.getPeerPrincipal());
            }
        };
    }

    public String getAddress()
    {
        return getNetwork().getRemoteAddress().toString();
    }


    public void received(final QpidByteBuffer msg)
    {

        AccessController.doPrivileged(new PrivilegedAction<Object>()
        {
            @Override
            public Object run()
            {
                updateLastReadTime();
                try
                {
                    int remaining;

                    do
                    {
                        remaining = msg.remaining();

                        switch (_frameReceivingState)
                        {
                            case AMQP_OR_SASL_HEADER:
                            case AMQP_HEADER:
                                if (remaining >= 8)
                                {
                                    processProtocolHeader(msg);
                                }
                                break;
                            case OPEN_ONLY:
                            case ANY_FRAME:
                            case SASL_INIT_ONLY:
                            case SASL_RESPONSE_ONLY:
                                _frameHandler.parse(msg);
                                break;
                            case CLOSED:
                                // ignore;
                                break;
                        }


                    }
                    while (msg.remaining() != remaining);
                }
                catch (IllegalArgumentException | IllegalStateException e)
                {
                    throw new ConnectionScopedRuntimeException(e);
                }
                catch (StoreException e)
                {
                    if (getAddressSpace().isActive())
                    {
                        throw new ServerScopedRuntimeException(e);
                    }
                    else
                    {
                        throw new ConnectionScopedRuntimeException(e);
                    }
                }
                return null;
            }
        }, getAccessControllerContext());

    }

    private void processProtocolHeader(final QpidByteBuffer msg)
    {
        if(msg.remaining() >= 8)
        {
            byte[] header = new byte[8];
            msg.get(header);

            final AuthenticationProvider authenticationProvider = getPort().getAuthenticationProvider();
            final SubjectCreator subjectCreator = authenticationProvider.getSubjectCreator(getTransport().isSecure());

            if(Arrays.equals(header, SASL_HEADER))
            {
                if(_saslComplete)
                {
                    throw new ConnectionScopedRuntimeException("SASL Layer header received after SASL already established");
                }

                getSender().send(QpidByteBuffer.wrap(SASL_HEADER));

                SaslMechanisms mechanisms = new SaslMechanisms();
                ArrayList<Symbol> mechanismsList = new ArrayList<Symbol>();
                for (String name :  subjectCreator.getMechanisms())
                {
                    mechanismsList.add(Symbol.valueOf(name));
                }
                mechanisms.setSaslServerMechanisms(mechanismsList.toArray(new Symbol[mechanismsList.size()]));
                send(new SASLFrame(mechanisms), null);

                _frameReceivingState = FrameReceivingState.SASL_INIT_ONLY;
                _frameHandler = new FrameHandler(this, true);
            }
            else if(Arrays.equals(header, AMQP_HEADER))
            {
                if(!_saslComplete)
                {
                    final List<String> mechanisms = subjectCreator.getMechanisms();

                    if(mechanisms.contains(ExternalAuthenticationManagerImpl.MECHANISM_NAME) && getNetwork().getPeerPrincipal() != null)
                    {
                        setUserPrincipal(new AuthenticatedPrincipal(getNetwork().getPeerPrincipal()));
                    }
                    else if(mechanisms.contains(AnonymousAuthenticationManager.MECHANISM_NAME))
                    {
                        setUserPrincipal(new AuthenticatedPrincipal(((AnonymousAuthenticationManager) authenticationProvider).getAnonymousPrincipal()));
                    }
                    else
                    {
                        // TODO - log auth failure / close
                        getNetwork().close();
                    }

                }
                getSender().send(QpidByteBuffer.wrap(AMQP_HEADER));
                _frameReceivingState = FrameReceivingState.OPEN_ONLY;
                _frameHandler = new FrameHandler(this, false);

            }
            else
            {
                throw new ConnectionScopedRuntimeException("Unknown protocol header");
            }

        }

    }


    public void closed()
    {
        try
        {
            inputClosed();
        }
        catch(RuntimeException e)
        {
            LOGGER.error("Exception while closing", e);
        }
        finally
        {
            try
            {
                endpointClosed();
            }
            finally
            {
                markTransportClosed();
            }
        }
    }

    public boolean canSend()
    {
        return true;
    }

    public void send(final AMQFrame amqFrame)
    {
        send(amqFrame, null);
    }


    public void send(final AMQFrame amqFrame, ByteBuffer buf)
    {
        updateLastWriteTime();
        FRAME_LOGGER.debug("SEND[{}|{}] : {}",
                           getNetwork().getRemoteAddress(),
                           amqFrame.getChannel(),
                           amqFrame.getFrameBody());

        int size = _frameWriter.send(amqFrame);
        if (size > getMaxFrameSize())
        {
            throw new OversizeFrameException(amqFrame, size);
        }
    }

    public void send(short channel, FrameBody body)
    {
        AMQFrame frame = AMQFrame.createAMQFrame(channel, body);
        send(frame);

    }

    public void close()
    {
        getAggregateTicker().addTicker(new ConnectionClosingTicker(System.currentTimeMillis() + CLOSE_RESPONSE_TIMEOUT,
                                                                   getNetwork()));

        // trigger a wakeup to ensure the ticker will be taken into account
        notifyWork();
    }

    @Override
    public boolean isTransportBlockedForWriting()
    {
        return _transportBlockedForWriting;
    }
    @Override
    public void setTransportBlockedForWriting(final boolean blocked)
    {
        if(_transportBlockedForWriting != blocked)
        {
            _transportBlockedForWriting = blocked;
            transportStateChanged();
        }

    }

    @Override
    public Iterator<Runnable> processPendingIterator()
    {
        if (isIOThread())
        {
            return new ProcessPendingIterator();
        }
        else
        {
            return Collections.emptyIterator();
        }
    }

    @Override
    public boolean hasWork()
    {
        return _stateChanged.get();
    }

    @Override
    public void notifyWork()
    {
        _stateChanged.set(true);

        final Action<ProtocolEngine> listener = _workListener.get();
        if(listener != null)
        {
            listener.performAction(this);
        }
    }

    @Override
    public void clearWork()
    {
        _stateChanged.set(false);
    }

    @Override
    public void setWorkListener(final Action<ProtocolEngine> listener)
    {
        _workListener.set(listener);
    }

    public boolean hasSessionWithName(final byte[] name)
    {
        return false;
    }

    public void sendConnectionCloseAsync(final AMQConstant cause, final String message)
    {
        Action<ConnectionHandler> action = new Action<ConnectionHandler>()
        {
            @Override
            public void performAction(final ConnectionHandler object)
            {
                closeConnection();

            }
        };
        addAsyncTask(action);
    }

    public void closeSessionAsync(final AMQSessionModel<?> session,
                                  final AMQConstant cause, final String message)
    {
        closeSessionAsync((Session_1_0) session, cause, message);
    }

    public void block()
    {
        // TODO
    }

    public String getRemoteContainerName()
    {
        return _remoteContainerId;
    }

    public Collection<? extends Session_1_0> getSessionModels()
    {
        return Collections.unmodifiableCollection(_sessions);
    }

    public void unblock()
    {
        // TODO
    }

    public long getSessionCountLimit()
    {
        return 0;  // TODO
    }

    @Override
    public boolean isOrderlyClose()
    {
        return _orderlyClose.get();
    }

    private void addAsyncTask(final Action<ConnectionHandler> action)
    {
        _asyncTaskList.add(action);
        notifyWork();
    }


    private void sendOpen(final int channelMax, final int maxFrameSize)
    {
        Open open = new Open();

        if (_receivingSessions == null)
        {
            _receivingSessions = new Session_1_0[channelMax + 1];
            _sendingSessions = new Session_1_0[channelMax + 1];
        }
        if (channelMax < _channelMax)
        {
            _channelMax = channelMax;
        }
        open.setChannelMax(UnsignedShort.valueOf((short) channelMax));
        open.setContainerId(_container.getId());
        open.setMaxFrameSize(UnsignedInteger.valueOf(maxFrameSize));
        // TODO - should we try to set the hostname based on the connection information?
        // open.setHostname();
        open.setIdleTimeOut(UnsignedInteger.valueOf(_desiredIdleTimeout));
        if (_properties != null)
        {
            open.setProperties(_properties);
        }

        sendFrame(CONNECTION_CONTROL_CHANNEL, open);
    }

    private void closeWithError(final AmqpError amqpError, final String errorDescription)
    {
        final Error err = new Error();
        err.setCondition(amqpError);
        err.setDescription(errorDescription);
        closeConnection(err);
        close();
        _closedOnOpen = true;
    }

    private Session_1_0 getSession(final short channel)
    {
        Session_1_0 session = _receivingSessions[channel];
        if (session == null)
        {
            Error error = new Error();
            error.setCondition(ConnectionError.FRAMING_ERROR);
            error.setDescription("Frame received on channel " + channel + " which is not known as a begun session.");
            handleError(error);
        }

        return session;
    }

    private void sendClose(Close closeToSend)
    {
        sendFrame(CONNECTION_CONTROL_CHANNEL, closeToSend);
        closeSender();
    }

    @Override
    public String toString()
    {
        NamedAddressSpace virtualHost = getAddressSpace();
        return "Connection_1_0["
               + _connectionId
               + " "
               + getAddress()
               + (virtualHost == null ? "" : (" vh : " + virtualHost.getName()))
               + ']';
    }


    private void assertState(final FrameReceivingState state)
    {
        if(_frameReceivingState != state)
        {
            throw new ConnectionScopedRuntimeException("Unexpected state, client has sent frame in an illegal order.  Required state: " + state + ", actual state: " + _frameReceivingState);
        }
    }


    private class ProcessPendingIterator implements Iterator<Runnable>
    {
        private final Collection<? extends AMQSessionModel<?>> _sessionsWithPending;
        private Iterator<? extends AMQSessionModel<?>> _sessionIterator;
        private ProcessPendingIterator()
        {
            _sessionsWithPending = new ArrayList<>(getSessionModels());
            _sessionIterator = _sessionsWithPending.iterator();
        }

        @Override
        public boolean hasNext()
        {
            return !(_sessionsWithPending.isEmpty() && _asyncTaskList.isEmpty());
        }

        @Override
        public Runnable next()
        {
            if(!_sessionsWithPending.isEmpty())
            {
                if(!_sessionIterator.hasNext())
                {
                    _sessionIterator = _sessionsWithPending.iterator();
                }
                final AMQSessionModel<?> session = _sessionIterator.next();
                return new Runnable()
                {
                    @Override
                    public void run()
                    {
                        if(!session.processPending())
                        {
                            _sessionIterator.remove();
                        }
                    }
                };
            }
            else if(!_asyncTaskList.isEmpty())
            {
                final Action<? super ConnectionHandler> asyncAction = _asyncTaskList.poll();
                return new Runnable()
                {
                    @Override
                    public void run()
                    {
                        asyncAction.performAction(AMQPConnection_1_0.this);
                    }
                };
            }
            else
            {
                throw new NoSuchElementException();
            }
        }

        @Override
        public void remove()
        {
            throw new UnsupportedOperationException();
        }
    }

    @Override
    public void initialiseHeartbeating(final long writerDelay, final long readerDelay)
    {
        super.initialiseHeartbeating(writerDelay, readerDelay);
    }
}