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

import java.nio.charset.StandardCharsets;
import java.security.AccessControlException;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import com.google.common.util.concurrent.Futures;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.consumer.ConsumerOption;
import org.apache.qpid.server.exchange.ExchangeDefaults;
import org.apache.qpid.server.filter.AMQInvalidArgumentException;
import org.apache.qpid.server.filter.AMQPFilterTypes;
import org.apache.qpid.server.filter.ArrivalTimeFilter;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.filter.FilterManagerFactory;
import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.logging.messages.ChannelMessages;
import org.apache.qpid.server.logging.messages.ExchangeMessages;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.message.RejectType;
import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.model.AbstractConfiguredObject;
import org.apache.qpid.server.model.AlternateBinding;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.ExclusivityPolicy;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.NoFactoryForTypeException;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.protocol.ErrorCodes;
import org.apache.qpid.server.protocol.v0_10.transport.*;
import org.apache.qpid.server.queue.QueueArgumentsConverter;
import org.apache.qpid.server.store.MessageHandle;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.server.transport.ProtocolEngine;
import org.apache.qpid.server.txn.AlreadyKnownDtxException;
import org.apache.qpid.server.txn.DtxNotSelectedException;
import org.apache.qpid.server.txn.IncorrectDtxStateException;
import org.apache.qpid.server.txn.JoinAndResumeDtxException;
import org.apache.qpid.server.txn.NotAssociatedDtxException;
import org.apache.qpid.server.txn.RollbackOnlyDtxException;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.txn.SuspendAndFailDtxException;
import org.apache.qpid.server.txn.TimeoutDtxException;
import org.apache.qpid.server.txn.UnknownDtxBranchException;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.MessageDestinationIsAlternateException;
import org.apache.qpid.server.virtualhost.RequiredExchangeException;
import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
import org.apache.qpid.server.virtualhost.UnknownAlternateBindingException;
import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException;

public class ServerSessionDelegate extends MethodDelegate<ServerSession> implements ProtocolDelegate<ServerSession>
{
    private static final Logger LOGGER = LoggerFactory.getLogger(ServerSessionDelegate.class);
    private final AtomicBoolean _closed = new AtomicBoolean();

    public ServerSessionDelegate()
    {

    }

    @Override
    public void command(ServerSession session, Method method)
    {
        try
        {
            if(!session.isClosing())
            {
                Object asyncCommandMark = session.getAsyncCommandMark();
                command(session, method, false);
                Object newOutstanding = session.getAsyncCommandMark();
                if(newOutstanding == null || newOutstanding == asyncCommandMark)
                {
                    session.processed(method);
                }

                if(newOutstanding != null)
                {
                    session.completeAsyncCommands();
                }

                if (method.isSync())
                {
                    session.awaitCommandCompletion();
                    session.flushProcessed();
                }
            }
        }
        catch(ServerScopedRuntimeException | ConnectionScopedRuntimeException e)
        {
            throw e;
        }
        catch(RuntimeException e)
        {
            LOGGER.error("Exception processing command", e);
            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, "Exception processing command: " + e);
        }
    }

    @Override
    public void messageAccept(ServerSession session, MessageAccept method)
    {
        final ServerSession serverSession = session;
        serverSession.accept(method.getTransfers());
        if(!serverSession.isTransactional())
        {
            serverSession.recordFuture(Futures.<Void>immediateFuture(null),
                                       new CommandProcessedAction(serverSession, method));
        }
    }

    @Override
    public void messageReject(ServerSession session, MessageReject method)
    {
        session.reject(method.getTransfers());
    }

    @Override
    public void messageRelease(ServerSession session, MessageRelease method)
    {
        session.release(method.getTransfers(), method.getSetRedelivered());
    }

    @Override
    public void messageAcquire(ServerSession session, MessageAcquire method)
    {
        RangeSet acquiredRanges = session.acquire(method.getTransfers());

        Acquired result = new Acquired(acquiredRanges);


        session.executionResult((int) method.getId(), result);


    }

    @Override
    public void messageResume(ServerSession session, MessageResume method)
    {
        super.messageResume(session, method);
    }

    @Override
    public void messageSubscribe(ServerSession session, MessageSubscribe method)
    {
        /*
          TODO - work around broken Python tests
          Correct code should read like
          if not hasAcceptMode() exception ILLEGAL_ARGUMENT "Accept-mode not supplied"
          else if not method.hasAcquireMode() exception ExecutionErrorCode.ILLEGAL_ARGUMENT, "Acquire-mode not supplied"
        */
        if(!method.hasAcceptMode())
        {
            method.setAcceptMode(MessageAcceptMode.EXPLICIT);
        }
        if(!method.hasAcquireMode())
        {
            method.setAcquireMode(MessageAcquireMode.PRE_ACQUIRED);

        }

        if(!method.hasQueue())
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "queue not supplied");
        }
        else
        {
            String destination = method.getDestination();

            if (destination == null)
            {
                exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Subscriber must provide a destination. The protocol specification marking the destination argument as optional is considered a mistake.");
            }
            else if(session.getSubscription(destination) != null)
            {
                exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Subscription already exists with destination '"+destination+"'");
            }
            else
            {
                String queueName = method.getQueue();
                NamedAddressSpace addressSpace = getAddressSpace(session);

                final Collection<MessageSource> sources = new HashSet<>();
                final MessageSource queue = addressSpace.getAttainedMessageSource(queueName);

                if(method.getArguments() != null && method.getArguments().get("x-multiqueue") instanceof Collection)
                {
                    for(Object object : (Collection<Object>)method.getArguments().get("x-multiqueue"))
                    {
                        String sourceName = String.valueOf(object);
                        sourceName = sourceName.trim();
                        if(sourceName.length() != 0)
                        {
                            MessageSource source = addressSpace.getAttainedMessageSource(sourceName);
                            if(source == null)
                            {
                                sources.clear();
                                break;
                            }
                            else
                            {
                                sources.add(source);
                            }
                        }
                    }
                    queueName = method.getArguments().get("x-multiqueue").toString();
                }
                else if(queue != null)
                {
                    sources.add(queue);
                }

                if(sources.isEmpty())
                {
                    exception(session, method, ExecutionErrorCode.NOT_FOUND, "Queue: " + queueName + " not found");
                }
                else if(!verifySessionAccess(session, sources))
                {
                    exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
                }
                else
                {
                    ProtocolEngine protocolEngine = getServerConnection(session).getAmqpConnection();
                    FlowCreditManager_0_10 creditManager = new WindowCreditManager(0L,0L);

                    FilterManager filterManager = null;
                    try
                    {
                        filterManager = FilterManagerFactory.createManager(method.getArguments());
                    }
                    catch (AMQInvalidArgumentException amqe)
                    {
                        exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "Exception Creating FilterManager");
                        return;
                    }


                    if(method.hasArguments() && method.getArguments().containsKey(AMQPFilterTypes.REPLAY_PERIOD.toString()))
                    {
                        Object value = method.getArguments().get(AMQPFilterTypes.REPLAY_PERIOD.toString());
                        final long period;
                        if(value instanceof Number)
                        {
                            period = ((Number)value).longValue();
                        }
                        else if(value instanceof String)
                        {
                            try
                            {
                                period = Long.parseLong(value.toString());
                            }
                            catch (NumberFormatException e)
                            {
                                exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "Cannot parse value " + value + " as a number for filter " + AMQPFilterTypes.REPLAY_PERIOD.toString());
                                return;
                            }
                        }
                        else
                        {
                            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "Cannot parse value " + value + " as a number for filter " + AMQPFilterTypes.REPLAY_PERIOD.toString());
                            return;
                        }
                        final long startingFrom = System.currentTimeMillis() - (1000l * period);
                        if(filterManager == null)
                        {
                            filterManager = new FilterManager();
                        }
                        MessageFilter filter = new ArrivalTimeFilter(startingFrom, period == 0);
                        filterManager.add(filter.getName(), filter);

                    }

                    boolean multiQueue = sources.size()>1;
                    ConsumerTarget_0_10 target = new ConsumerTarget_0_10(session, destination,
                                                                         method.getAcceptMode(),
                                                                         method.getAcquireMode(),
                                                                         MessageFlowMode.WINDOW,
                                                                         creditManager,
                                                                         method.getArguments(),
                                                                         multiQueue
                    );

                    Integer priority = null;
                    if(method.hasArguments() && method.getArguments().containsKey("x-priority"))
                    {
                        Object value = method.getArguments().get("x-priority");
                        if(value instanceof Number)
                        {
                            priority = ((Number)value).intValue();
                        }
                        else if(value instanceof String)
                        {
                            try
                            {
                                priority = Integer.parseInt(value.toString());
                            }
                            catch (NumberFormatException e)
                            {
                            }
                        }
                    }

                    session.register(destination, target);
                    try
                    {
                        EnumSet<ConsumerOption> options = EnumSet.noneOf(ConsumerOption.class);
                        if(method.getAcquireMode() == MessageAcquireMode.PRE_ACQUIRED)
                        {
                            options.add(ConsumerOption.ACQUIRES);
                        }
                        if(method.getAcquireMode() != MessageAcquireMode.NOT_ACQUIRED || method.getAcceptMode() == MessageAcceptMode.EXPLICIT)
                        {
                            options.add(ConsumerOption.SEES_REQUEUES);
                        }
                        if(method.getExclusive())
                        {
                            options.add(ConsumerOption.EXCLUSIVE);
                        }
                        for(MessageSource source : sources)
                        {
                            source.addConsumer(target,
                                               filterManager,
                                               MessageTransferMessage.class,
                                               destination,
                                               options,
                                               priority);
                        }
                        target.updateNotifyWorkDesired();
                    }
                    catch (Queue.ExistingExclusiveConsumer existing)
                    {
                        exception(session, method, ExecutionErrorCode.RESOURCE_LOCKED, "Queue has an exclusive consumer");
                    }
                    catch (Queue.ExistingConsumerPreventsExclusive exclusive)
                    {
                        exception(session, method, ExecutionErrorCode.RESOURCE_LOCKED, "Queue has an existing consumer - can't subscribe exclusively");
                    }
                    catch (AccessControlException e)
                    {
                        exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
                    }
                    catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
                    {
                        exception(session, method, ExecutionErrorCode.RESOURCE_LOCKED, "Queue has an incompatible exclusivity policy");
                    }
                    catch (MessageSource.QueueDeleted queueDeleted)
                    {
                        exception(session, method, ExecutionErrorCode.NOT_FOUND, "Queue was deleted");
                    }
                }
            }
        }
    }

    protected boolean verifySessionAccess(final ServerSession session, final Collection<MessageSource> queues)
    {
        for(MessageSource source : queues)
        {
            if(!verifySessionAccess(session, source))
            {
                return false;
            }
        }
        return true;
    }

    protected boolean verifySessionAccess(final ServerSession session, final MessageSource queue)
    {
        return queue.verifySessionAccess(session.getModelObject());
    }

    private static String getMessageUserId(MessageTransfer xfr)
    {
        byte[] userIdBytes = xfr.getHeader() == null ? null : xfr.getHeader().getMessageProperties() == null ? null : xfr.getHeader().getMessageProperties().getUserId();
        return userIdBytes == null ? null : new String(userIdBytes, StandardCharsets.UTF_8);
    }

    @Override
    public void messageTransfer(ServerSession ssn, final MessageTransfer xfr)
    {
        try
        {
            if(ssn.blockingTimeoutExceeded())
            {
                getEventLogger(ssn).message(ChannelMessages.FLOW_CONTROL_IGNORED());

                ssn.close(ErrorCodes.MESSAGE_TOO_LARGE,
                          "Session flow control was requested, but not enforced by sender");
            }
            else if(xfr.getBodySize() > ssn.getConnection().getMaxMessageSize())
            {
                exception(ssn, xfr, ExecutionErrorCode.RESOURCE_LIMIT_EXCEEDED,
                          "Message size of " + xfr.getBodySize() + " greater than allowed maximum of " + ssn.getConnection().getMaxMessageSize());
            }
            else
            {
                final MessageDestination destination = getDestinationForMessage(ssn, xfr);

                final DeliveryProperties delvProps =
                        xfr.getHeader() == null ? null : xfr.getHeader().getDeliveryProperties();
                if (delvProps != null && delvProps.hasTtl() && !delvProps.hasExpiration())
                {
                    delvProps.setExpiration(System.currentTimeMillis() + delvProps.getTtl());
                }

                final MessageMetaData_0_10 messageMetaData = new MessageMetaData_0_10(xfr);

                final NamedAddressSpace virtualHost = getAddressSpace(ssn);
                try
                {
                    ssn.getAMQPConnection().checkAuthorizedMessagePrincipal(getMessageUserId(xfr));
                    ssn.authorisePublish(destination, messageMetaData.getRoutingKey(), messageMetaData.isImmediate(), ssn
                            .getAMQPConnection().getLastReadTime());

                }
                catch (AccessControlException e)
                {
                    ExecutionErrorCode errorCode = ExecutionErrorCode.UNAUTHORIZED_ACCESS;
                    exception(ssn, xfr, errorCode, e.getMessage());

                    return;
                }

                final MessageStore store = virtualHost.getMessageStore();
                final StoredMessage<MessageMetaData_0_10> storeMessage = createStoreMessage(xfr, messageMetaData, store);
                final MessageTransferMessage message =
                        new MessageTransferMessage(storeMessage, ssn.getReference());
                MessageReference<MessageTransferMessage> reference = message.newReference();

                try
                {
                    final InstanceProperties instanceProperties = new InstanceProperties()
                    {
                        @Override
                        public Object getProperty(final Property prop)
                        {
                            switch (prop)
                            {
                                case EXPIRATION:
                                    return message.getExpiration();
                                case IMMEDIATE:
                                    return message.isImmediate();
                                case MANDATORY:
                                    return (delvProps == null || !delvProps.getDiscardUnroutable())
                                           && xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT;
                                case PERSISTENT:
                                    return message.isPersistent();
                                case REDELIVERED:
                                    return delvProps.getRedelivered();
                            }
                            return null;
                        }
                    };

                    RoutingResult<MessageTransferMessage> routingResult = ssn.enqueue(message, instanceProperties, destination);

                    boolean explictlyRejected = routingResult.containsReject(RejectType.LIMIT_EXCEEDED);
                    if (!routingResult.hasRoutes() || explictlyRejected)
                    {
                        boolean closeWhenNoRoute = ssn.getAMQPConnection().getPort().getCloseWhenNoRoute();
                        boolean discardUnroutable = delvProps != null && delvProps.getDiscardUnroutable();
                        if (!discardUnroutable && xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT)
                        {
                            RangeSet rejects = RangeSetFactory.createRangeSet();
                            rejects.add(xfr.getId());
                            MessageReject reject = new MessageReject(rejects, MessageRejectCode.UNROUTABLE, "Unroutable");
                            ssn.invoke(reject);
                        }
                        else if (!discardUnroutable && closeWhenNoRoute && explictlyRejected)
                        {
                            ExecutionErrorCode code = ExecutionErrorCode.RESOURCE_LIMIT_EXCEEDED;
                            String errorMessage = String.format("No route for message with destination '%s' and routing key '%s' : %s",
                                                                xfr.getDestination(),
                                                                message.getInitialRoutingAddress(),
                                                                routingResult.getRejectReason());

                            ExecutionException ex = new ExecutionException();
                            ex.setErrorCode(code);
                            ex.setDescription(errorMessage);
                            ssn.invoke(ex);
                            ssn.close(ErrorCodes.RESOURCE_ERROR, errorMessage);
                            return;
                        }
                        else
                        {
                            getEventLogger(ssn).message(ExchangeMessages.DISCARDMSG(destination.getName(),
                                                                                    messageMetaData.getRoutingKey()));
                        }
                    }

                    // TODO: we currently do not send MessageAccept when AcceptMode is EXPLICIT
                    if (ssn.isTransactional())
                    {
                        ssn.processed(xfr);
                    }
                    else
                    {
                        ssn.recordFuture(Futures.immediateFuture(null),
                                         new CommandProcessedAction(ssn, xfr));
                    }
                }
                catch (VirtualHostUnavailableException e)
                {
                    getServerConnection(ssn).sendConnectionCloseAsync(ConnectionCloseCode.CONNECTION_FORCED, e.getMessage());
                }
                finally
                {
                    reference.release();
                }
            }
        }
        finally
        {
            xfr.dispose();
        }
    }

    private StoredMessage<MessageMetaData_0_10> createStoreMessage(final MessageTransfer xfr,
                                                                   final MessageMetaData_0_10 messageMetaData, final MessageStore store)
    {
        final MessageHandle<MessageMetaData_0_10> addedMessage = store.addMessage(messageMetaData);
        QpidByteBuffer body = xfr.getBody();
        if(body != null)
        {
            addedMessage.addContent(body);
        }
        final StoredMessage<MessageMetaData_0_10> storedMessage = addedMessage.allContentAdded();
        return storedMessage;
    }

    @Override
    public void messageCancel(ServerSession session, MessageCancel method)
    {
        String destination = method.getDestination();

        ConsumerTarget_0_10 sub = session.getSubscription(destination);

        if(sub == null)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
        }
        else
        {
            session.unregister(sub);
        }
    }

    @Override
    public void messageFlush(ServerSession session, MessageFlush method)
    {
        String destination = method.getDestination();

        ConsumerTarget_0_10 sub = session.getSubscription(destination);

        if(sub == null)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
        }
        else
        {
            sub.flush();
        }
    }

    @Override
    public void txSelect(ServerSession session, TxSelect method)
    {
        // TODO - check current tx mode
        session.selectTx();
    }

    @Override
    public void txCommit(ServerSession session, TxCommit method)
    {
        // TODO - check current tx mode
        session.commit();
    }

    @Override
    public void txRollback(ServerSession session, TxRollback method)
    {
        // TODO - check current tx mode
        session.rollback();
    }

    @Override
    public void dtxSelect(ServerSession session, DtxSelect method)
    {
        // TODO - check current tx mode
        session.selectDtx();
    }

    @Override
    public void dtxStart(ServerSession session, DtxStart method)
    {
        XaResult result = new XaResult();
        result.setStatus(DtxXaStatus.XA_OK);
        try
        {
            session.startDtx(method.getXid(), method.getJoin(), method.getResume());
            session.executionResult(method.getId(), result);
        }
        catch(JoinAndResumeDtxException e)
        {
            exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Unknown xid " + method.getXid());
        }
        catch(AlreadyKnownDtxException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Xid already started an neither join nor " +
                                                                       "resume set" + method.getXid());
        }
        catch(DtxNotSelectedException e)
        {
            exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
        }

    }

    @Override
    public void dtxEnd(ServerSession session, DtxEnd method)
    {
        XaResult result = new XaResult();
        result.setStatus(DtxXaStatus.XA_OK);
        try
        {
            try
            {
                session.endDtx(method.getXid(), method.getFail(), method.getSuspend());
            }
            catch (TimeoutDtxException e)
            {
                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
            }
            session.executionResult(method.getId(), result);
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
        }
        catch(NotAssociatedDtxException e)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
        }
        catch(DtxNotSelectedException e)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
        }
        catch(SuspendAndFailDtxException e)
        {
            exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
        }

    }

    @Override
    public void dtxCommit(ServerSession session, DtxCommit method)
    {
        XaResult result = new XaResult();
        result.setStatus(DtxXaStatus.XA_OK);
        try
        {
            try
            {
                session.commitDtx(method.getXid(), method.getOnePhase());
            }
            catch (RollbackOnlyDtxException e)
            {
                result.setStatus(DtxXaStatus.XA_RBROLLBACK);
            }
            catch (TimeoutDtxException e)
            {
                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
            }
            session.executionResult(method.getId(), result);
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
        }
        catch(IncorrectDtxStateException e)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
        }
        catch(StoreException e)
        {
            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
            throw e;
        }
    }

    @Override
    public void dtxForget(ServerSession session, DtxForget method)
    {
        try
        {
            session.forgetDtx(method.getXid());
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
        }
        catch(IncorrectDtxStateException e)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
        }

    }

    @Override
    public void dtxGetTimeout(ServerSession session, DtxGetTimeout method)
    {
        GetTimeoutResult result = new GetTimeoutResult();
        try
        {
            result.setTimeout(session.getTimeoutDtx(method.getXid()));
            session.executionResult(method.getId(), result);
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
        }
    }

    @Override
    public void dtxPrepare(ServerSession session, DtxPrepare method)
    {
        XaResult result = new XaResult();
        result.setStatus(DtxXaStatus.XA_OK);
        try
        {
            try
            {
                session.prepareDtx(method.getXid());
            }
            catch (RollbackOnlyDtxException e)
            {
                result.setStatus(DtxXaStatus.XA_RBROLLBACK);
            }
            catch (TimeoutDtxException e)
            {
                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
            }
            session.executionResult((int) method.getId(), result);
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
        }
        catch(IncorrectDtxStateException e)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
        }
        catch(StoreException e)
        {
            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
            throw e;
        }
    }

    @Override
    public void dtxRecover(ServerSession session, DtxRecover method)
    {
        RecoverResult result = new RecoverResult();
        List inDoubt = session.recoverDtx();
        result.setInDoubt(inDoubt);
        session.executionResult(method.getId(), result);
    }

    @Override
    public void dtxRollback(ServerSession session, DtxRollback method)
    {

        XaResult result = new XaResult();
        result.setStatus(DtxXaStatus.XA_OK);
        try
        {
            try
            {
                session.rollbackDtx(method.getXid());
            }
            catch (TimeoutDtxException e)
            {
                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
            }
            session.executionResult(method.getId(), result);
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
        }
        catch(IncorrectDtxStateException e)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
        }
        catch(StoreException e)
        {
            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
            throw e;
        }
    }

    @Override
    public void dtxSetTimeout(ServerSession session, DtxSetTimeout method)
    {
        try
        {
            session.setTimeoutDtx(method.getXid(), method.getTimeout());
        }
        catch(UnknownDtxBranchException e)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
        }
    }

    @Override
    public void executionSync(final ServerSession ssn, final ExecutionSync sync)
    {
        ssn.awaitCommandCompletion();
        ssn.syncPoint();
    }


    @Override
    public void exchangeDeclare(ServerSession session, ExchangeDeclare method)
    {
        String exchangeName = method.getExchange();
        NamedAddressSpace addressSpace = getAddressSpace(session);

        //we must check for any unsupported arguments present and throw not-implemented
        if(method.hasArguments())
        {
            Map<String,Object> args = method.getArguments();
            //QPID-3392: currently we don't support any!
            if(!args.isEmpty())
            {
                exception(session, method, ExecutionErrorCode.NOT_IMPLEMENTED, "Unsupported exchange argument(s) found " + args.keySet().toString());
                return;
            }
        }
        String alternateExchangeName = method.getAlternateExchange();
        if(nameNullOrEmpty(method.getExchange()))
        {
            // special case handling to fake the existence of the default exchange for 0-10
            if(!ExchangeDefaults.DIRECT_EXCHANGE_CLASS.equals(method.getType()))
            {
                exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
                          "Attempt to redeclare default exchange "
                          + " of type " + ExchangeDefaults.DIRECT_EXCHANGE_CLASS
                          + " to " + method.getType() +".");
            }
            if(!nameNullOrEmpty(alternateExchangeName))
            {
                exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
                          "Attempt to set alternate exchange of the default exchange "
                          + " to " + alternateExchangeName + ".");
            }
        }
        else
        {
            if(method.getPassive())
            {

                Exchange<?> exchange = getExchange(session, exchangeName);

                if(exchange == null)
                {
                    exception(session, method, ExecutionErrorCode.NOT_FOUND, "not-found: exchange-name '" + exchangeName + "'");
                }
                else
                {
                    if (!exchange.getType().equals(method.getType())
                            && (method.getType() != null && method.getType().length() > 0))
                    {
                        exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Attempt to redeclare exchange: "
                                + exchangeName + " of type " + exchange.getType() + " to " + method.getType() + ".");
                    }
                }
            }
            else
            {

                try
                {
                    Map<String,Object> attributes = new HashMap<String, Object>();

                    attributes.put(org.apache.qpid.server.model.Exchange.NAME, method.getExchange());
                    attributes.put(org.apache.qpid.server.model.Exchange.TYPE, method.getType());
                    attributes.put(org.apache.qpid.server.model.Exchange.DURABLE, method.getDurable());
                    attributes.put(org.apache.qpid.server.model.Exchange.LIFETIME_POLICY,
                                   method.getAutoDelete() ? LifetimePolicy.DELETE_ON_NO_LINKS : LifetimePolicy.PERMANENT);
                    if (method.hasAlternateExchange() && !nameNullOrEmpty(alternateExchangeName))
                    {
                        validateAlternateExchangeIsNotQueue(addressSpace, alternateExchangeName);

                        attributes.put(org.apache.qpid.server.model.Exchange.ALTERNATE_BINDING,
                                       Collections.singletonMap(AlternateBinding.DESTINATION, alternateExchangeName));
                    }
                    addressSpace.createMessageDestination(Exchange.class, attributes);;
                }
                catch(ReservedExchangeNameException e)
                {
                    Exchange<?> existingExchange = getExchange(session, exchangeName);
                    if(existingExchange == null
                       || !existingExchange.getType().equals(method.getType())
                       || (method.hasAlternateExchange() && (existingExchange.getAlternateBinding() == null ||
                                                             !alternateExchangeName
                                                                    .equals(existingExchange.getAlternateBinding().getDestination()))) )
                    {
                        exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Attempt to declare exchange: "
                                                                                   + exchangeName + " which begins with reserved name or prefix.");
                    }
                }
                catch (UnknownAlternateBindingException e)
                {
                    exception(session,
                              method,
                              ExecutionErrorCode.NOT_FOUND,
                              String.format("Unknown alternate exchange '%s'", e.getAlternateBindingName()));
                }
                catch(NoFactoryForTypeException e)
                {
                    exception(session, method, ExecutionErrorCode.NOT_FOUND, "Unknown Exchange Type: " + method.getType());
                }
                catch(AbstractConfiguredObject.DuplicateNameException e)
                {
                    Exchange<?> exchange = (Exchange<?>) e.getExisting();
                    if(!exchange.getType().equals(method.getType()))
                    {
                        exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
                                "Attempt to redeclare exchange: " + exchangeName
                                        + " of type " + exchange.getType()
                                        + " to " + method.getType() +".");
                    }
                    else if(method.hasAlternateExchange()
                              && (exchange.getAlternateBinding() == null ||
                                  !alternateExchangeName.equals(exchange.getAlternateBinding().getDestination())))
                    {
                        exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
                                  "Attempt to change alternate exchange of: " + exchangeName
                                  + " from " + exchange.getAlternateBinding()
                                  + " to " + alternateExchangeName + ".");
                    }
                }
                catch (AccessControlException e)
                {
                    exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
                }
                catch (IllegalArgumentException | IllegalConfigurationException e)
                {
                    exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, e.getMessage());
                }
            }
        }
    }

    private void exception(ServerSession session, Method method, ExecutionErrorCode errorCode, String description)
    {
        ExecutionException ex = new ExecutionException();
        ex.setErrorCode(errorCode);
        ex.setCommandId(method.getId());
        ex.setDescription(description);

        session.invoke(ex);

        session.close(errorCode.getValue(), description);
    }

    private Exchange<?> getExchange(ServerSession session, String exchangeName)
    {
        return getExchange(getAddressSpace(session),exchangeName);
    }

    private Exchange<?> getExchange(NamedAddressSpace addressSpace, String exchangeName)
    {
        MessageDestination destination = addressSpace.getAttainedMessageDestination(exchangeName);
        return destination instanceof Exchange ? (Exchange<?>) destination : null;
    }


    private Queue<?> getQueue(NamedAddressSpace addressSpace, String name)
    {
        MessageSource source = addressSpace.getAttainedMessageSource(name);
        return source instanceof Queue ? (Queue<?>) source : null;
    }

    private MessageDestination getDestinationForMessage(ServerSession ssn, MessageTransfer xfr)
    {
        NamedAddressSpace addressSpace = getAddressSpace(ssn);

        MessageDestination destination;
        if(xfr.hasDestination())
        {
            destination = addressSpace.getAttainedMessageDestination(xfr.getDestination());
            if(destination == null)
            {
                destination = addressSpace.getDefaultDestination();
            }
            else
            {
                Header header = xfr.getHeader();
                DeliveryProperties delvProps;
                if(header == null)
                {
                    delvProps = new DeliveryProperties();
                    header = new Header(delvProps, null, null);
                    xfr.setHeader(header);
                }
                else if(header.getDeliveryProperties() == null)
                {
                    delvProps = new DeliveryProperties();
                    header = new Header(delvProps, header.getMessageProperties(), header.getNonStandardProperties());
                    xfr.setHeader(header);
                }
                else
                {
                    delvProps = header.getDeliveryProperties();
                }
                if(delvProps.getExchange() == null && !xfr.getDestination().equals(delvProps.getRoutingKey()))
                {
                    delvProps.setExchange(xfr.getDestination());
                }
            }
        }
        else if(xfr.getHeader() != null
                && xfr.getHeader().getDeliveryProperties() != null
                && xfr.getHeader().getDeliveryProperties().getExchange() != null)
        {
            destination = addressSpace.getAttainedMessageDestination(xfr.getHeader().getDeliveryProperties().getExchange());
        }
        else
        {
            destination = addressSpace.getDefaultDestination();
        }
        return destination;
    }

    private NamedAddressSpace getAddressSpace(ServerSession session)
    {
        ServerConnection conn = getServerConnection(session);
        return conn.getAddressSpace();
    }

    private ServerConnection getServerConnection(ServerSession session)
    {
        return (ServerConnection) session.getConnection();
    }

    private <T> T getContextValue(ServerSession session, Class<T> clazz, String name)
    {
        return getServerConnection(session).getAmqpConnection().getContextProvider().getContextValue(clazz, name);
    }

    private EventLogger getEventLogger(ServerSession session)
    {
        return getServerConnection(session).getAmqpConnection().getEventLogger();
    }

    @Override
    public void exchangeDelete(ServerSession session, ExchangeDelete method)
    {
        if (nameNullOrEmpty(method.getExchange()))
        {
            exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Delete not allowed for default exchange");
            return;
        }

        Exchange<?> exchange = getExchange(session, method.getExchange());

        if(exchange == null)
        {
            exception(session, method, ExecutionErrorCode.NOT_FOUND, "No such exchange '" + method.getExchange() + "'");
        }
        else
        {
            if (method.getIfUnused() && exchange.hasBindings())
            {
                exception(session, method, ExecutionErrorCode.PRECONDITION_FAILED, "Exchange has bindings");
            }
            else
            {
                try
                {
                    exchange.delete();
                }
                catch (MessageDestinationIsAlternateException e)
                {
                    exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Exchange in use as an alternate binding destination");
                }
                catch (RequiredExchangeException e)
                {
                    exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Exchange '"+method.getExchange()+"' cannot be deleted");
                }
                catch (AccessControlException e)
                {
                    exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
                }
            }
        }
    }

    private boolean nameNullOrEmpty(String name)
    {
        if(name == null || name.length() == 0)
        {
            return true;
        }

        return false;
    }

    @Override
    public void exchangeQuery(ServerSession session, ExchangeQuery method)
    {

        ExchangeQueryResult result = new ExchangeQueryResult();


        final String exchangeName = method.getName();

        if(nameNullOrEmpty(exchangeName))
        {
            // Fake the existence of the "default" exchange for 0-10
            result.setDurable(true);
            result.setType(ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
            result.setNotFound(false);
        }
        else
        {
            Exchange<?> exchange = getExchange(session, exchangeName);

            if(exchange != null)
            {
                result.setDurable(exchange.isDurable());
                result.setType(exchange.getType());
                result.setNotFound(false);
            }
            else
            {
                result.setNotFound(true);
            }
        }
        session.executionResult((int) method.getId(), result);
    }

    @Override
    public void exchangeBind(ServerSession session, ExchangeBind method)
    {

        NamedAddressSpace addressSpace = getAddressSpace(session);

        if (!method.hasQueue())
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "queue not set");
        }
        else
        {
            final String exchangeName = method.getExchange();
            if (nameNullOrEmpty(exchangeName))
            {
                exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Bind not allowed for default exchange");
            }
            else
            {
                //TODO - here because of non-compliant python tests
                // should raise exception ILLEGAL_ARGUMENT "binding-key not set"
                if (!method.hasBindingKey())
                {
                    method.setBindingKey(method.getQueue());
                }
                Queue<?> queue = getQueue(addressSpace, method.getQueue());
                Exchange<?> exchange = getExchange(addressSpace, exchangeName);
                if(queue == null)
                {
                    exception(session, method, ExecutionErrorCode.NOT_FOUND, "Queue: '" + method.getQueue() + "' not found");
                }
                else if(exchange == null)
                {
                    exception(session, method, ExecutionErrorCode.NOT_FOUND, "Exchange: '" + exchangeName + "' not found");
                }
                else if(exchange.getType().equals(ExchangeDefaults.HEADERS_EXCHANGE_CLASS) && (!method.hasArguments() || method.getArguments() == null || !method.getArguments().containsKey("x-match")))
                {
                    exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, "Bindings to an exchange of type " + ExchangeDefaults.HEADERS_EXCHANGE_CLASS + " require an x-match header");
                }
                else
                {
                    if (!exchange.isBound(method.getBindingKey(), method.getArguments(), queue))
                    {
                        try
                        {
                            exchange.addBinding(method.getBindingKey(), queue, method.getArguments());
                        }
                        catch (AccessControlException e)
                        {
                            exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
                        }
                    }
                    else
                    {
                        // todo
                    }
                }


            }
        }


    }

    @Override
    public void exchangeUnbind(ServerSession session, ExchangeUnbind method)
    {
        NamedAddressSpace addressSpace = getAddressSpace(session);

        if (!method.hasQueue())
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "queue not set");
        }
        else if (nameNullOrEmpty(method.getExchange()))
        {
            exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Unbind not allowed for default exchange");
        }
        else if (!method.hasBindingKey())
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "binding-key not set");
        }
        else
        {
            Queue<?> queue = getQueue(addressSpace, method.getQueue());
            Exchange<?> exchange = getExchange(addressSpace, method.getExchange());
            if(queue == null)
            {
                exception(session, method, ExecutionErrorCode.NOT_FOUND, "Queue: '" + method.getQueue() + "' not found");
            }
            else if(exchange == null)
            {
                exception(session, method, ExecutionErrorCode.NOT_FOUND, "Exchange: '" + method.getExchange() + "' not found");
            }
            else
            {
                try
                {
                    if(exchange.hasBinding(method.getBindingKey(), queue))
                    {
                        exchange.deleteBinding(method.getBindingKey(), queue);
                    }
                }
                catch (AccessControlException e)
                {
                    exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
                }
            }
        }
    }

    @Override
    public void exchangeBound(ServerSession session, ExchangeBound method)
    {

        ExchangeBoundResult result = new ExchangeBoundResult();
        NamedAddressSpace addressSpace = getAddressSpace(session);
        Exchange<?> exchange;
        MessageSource source;
        Queue<?> queue;
        boolean isDefaultExchange;
        if(!nameNullOrEmpty(method.getExchange()))
        {
            isDefaultExchange = false;
            exchange = getExchange(addressSpace, method.getExchange());

            if(exchange == null)
            {
                result.setExchangeNotFound(true);
            }
        }
        else
        {
            isDefaultExchange = true;
            exchange = null;
        }

        if(isDefaultExchange)
        {
            // fake the existence of the "default" exchange for 0-10
            if(method.hasQueue())
            {
                queue = getQueue(session, method.getQueue());

                if(queue == null)
                {
                    result.setQueueNotFound(true);
                }
                else
                {
                    if(method.hasBindingKey())
                    {
                        if(!method.getBindingKey().equals(method.getQueue()))
                        {
                            result.setKeyNotMatched(true);
                        }
                    }
                }
            }
            else if(method.hasBindingKey())
            {
                if(getQueue(session, method.getBindingKey()) == null)
                {
                    result.setKeyNotMatched(true);
                }
            }

            if(method.hasArguments() && !method.getArguments().isEmpty())
            {
                result.setArgsNotMatched(true);
            }


        }
        else if(method.hasQueue())
        {
            source = getMessageSource(session, method.getQueue());

            if(source == null)
            {
                result.setQueueNotFound(true);
            }
            if(source == null || source instanceof Queue)
            {
                queue = (Queue<?>) source;

                if (exchange != null && queue != null)
                {

                    boolean queueMatched = exchange.isBound(queue);

                    result.setQueueNotMatched(!queueMatched);


                    if (method.hasBindingKey())
                    {

                        if (queueMatched)
                        {
                            final boolean keyMatched = exchange.isBound(method.getBindingKey(), queue);
                            result.setKeyNotMatched(!keyMatched);
                            if (method.hasArguments())
                            {
                                if (keyMatched)
                                {
                                    result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(),
                                                                               method.getArguments(),
                                                                               queue));
                                }
                                else
                                {
                                    result.setArgsNotMatched(!exchange.isBound(method.getArguments(), queue));
                                }
                            }
                        }
                        else
                        {
                            boolean keyMatched = exchange.isBound(method.getBindingKey());
                            result.setKeyNotMatched(!keyMatched);
                            if (method.hasArguments())
                            {
                                if (keyMatched)
                                {
                                    result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(),
                                                                               method.getArguments()));
                                }
                                else
                                {
                                    result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
                                }
                            }
                        }

                    }
                    else if (method.hasArguments())
                    {
                        if (queueMatched)
                        {
                            result.setArgsNotMatched(!exchange.isBound(method.getArguments(), queue));
                        }
                        else
                        {
                            result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
                        }
                    }

                }
                else if (exchange != null && method.hasBindingKey())
                {
                    final boolean keyMatched = exchange.isBound(method.getBindingKey());
                    result.setKeyNotMatched(!keyMatched);

                    if (method.hasArguments())
                    {
                        if (keyMatched)
                        {
                            result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(), method.getArguments()));
                        }
                        else
                        {
                            result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
                        }
                    }


                }
            }

        }
        else if(exchange != null && method.hasBindingKey())
        {
            final boolean keyMatched = exchange.isBound(method.getBindingKey());
            result.setKeyNotMatched(!keyMatched);

            if(method.hasArguments())
            {
                if(keyMatched)
                {
                    result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(), method.getArguments()));
                }
                else
                {
                    result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
                }
            }

        }
        else if(exchange != null && method.hasArguments())
        {
            result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
        }


        session.executionResult((int) method.getId(), result);


    }

    private MessageSource getMessageSource(ServerSession session, String queue)
    {
        return getAddressSpace(session).getAttainedMessageSource(queue);
    }

    private Queue<?> getQueue(ServerSession session, String queue)
    {
        return getQueue(getAddressSpace(session), queue);
    }

    @Override
    public void queueDeclare(ServerSession session, final QueueDeclare method)
    {

        final NamedAddressSpace addressSpace = getAddressSpace(session);

        String queueName = method.getQueue();
        Queue<?> queue;
        //TODO: do we need to check that the queue already exists with exactly the same "configuration"?

        final boolean exclusive = method.getExclusive();
        final boolean autoDelete = method.getAutoDelete();

        if(method.getPassive())
        {
            queue = getQueue(addressSpace, queueName);

            if (queue == null)
            {
                String description = "Queue: " + queueName + " not found on VirtualHost(" + addressSpace + ").";
                ExecutionErrorCode errorCode = ExecutionErrorCode.NOT_FOUND;

                exception(session, method, errorCode, description);

            }
            else if (exclusive)
            {
                if (queue.getExclusive() == ExclusivityPolicy.NONE)
                {
                    String description = "Cannot passively declare queue ('" + queueName + "')"
                                         + " as exclusive as queue with same name is" +
                                         " already declared as non-exclusive";
                    ExecutionErrorCode errorCode = ExecutionErrorCode.RESOURCE_LOCKED;
                    exception(session, method, errorCode, description);

                }
                else if (!verifySessionAccess(session, queue))
                {
                    String description = "Cannot passively declare queue('" + queueName + "'),"
                                         + " as exclusive queue with same name "
                                         + "declared on another session";
                    ExecutionErrorCode errorCode = ExecutionErrorCode.RESOURCE_LOCKED;
                    exception(session, method, errorCode, description);
                }
            }
        }
        else
        {
            try
            {
                final Map<String, Object> arguments = QueueArgumentsConverter.convertWireArgsToModel(queueName,
                                                                                                     method.getArguments());
                final String alternateExchangeName = method.getAlternateExchange();
                if (method.hasAlternateExchange() && !nameNullOrEmpty(alternateExchangeName))
                {
                    validateAlternateExchangeIsNotQueue(addressSpace, alternateExchangeName);
                    arguments.put(Queue.ALTERNATE_BINDING, Collections.singletonMap(AlternateBinding.DESTINATION, alternateExchangeName));
                }

                arguments.put(Queue.NAME, queueName);

                if(!arguments.containsKey(Queue.LIFETIME_POLICY))
                {
                    LifetimePolicy lifetime;
                    if(autoDelete)
                    {
                        lifetime = exclusive ? LifetimePolicy.DELETE_ON_SESSION_END
                                : LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS;
                    }
                    else
                    {
                        lifetime = LifetimePolicy.PERMANENT;
                    }
                    arguments.put(Queue.LIFETIME_POLICY, lifetime);
                }

                if(!arguments.containsKey(Queue.EXCLUSIVE))
                {
                    ExclusivityPolicy exclusivityPolicy =
                            exclusive ? ExclusivityPolicy.SESSION : ExclusivityPolicy.NONE;

                    arguments.put(Queue.EXCLUSIVE, exclusivityPolicy);
                }

                arguments.put(Queue.DURABLE, method.getDurable());


                queue = addressSpace.createMessageSource(Queue.class, arguments);

            }
            catch(AbstractConfiguredObject.DuplicateNameException qe)
            {
                queue = (Queue<?>) qe.getExisting();
                if (!verifySessionAccess(session, queue))
                {
                    String description = "Cannot declare queue('" + queueName + "'),"
                                                                           + " as exclusive queue with same name "
                                                                           + "declared on another session";
                    ExecutionErrorCode errorCode = ExecutionErrorCode.RESOURCE_LOCKED;

                    exception(session, method, errorCode, description);
                }
            }
            catch (AccessControlException e)
            {
                exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
            }
            catch (UnknownAlternateBindingException e)
            {
                exception(session,
                          method,
                          ExecutionErrorCode.NOT_FOUND,
                          String.format("Unknown alternate exchange '%s'", e.getAlternateBindingName()));
            }
            catch (IllegalArgumentException | IllegalConfigurationException e)
            {
                exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, e.getMessage());
            }
        }
    }

    private void validateAlternateExchangeIsNotQueue(final NamedAddressSpace addressSpace, final String alternateExchangeName)
    {
        MessageDestination alternateMessageDestination = addressSpace.getAttainedMessageDestination(alternateExchangeName, false);
        if (alternateMessageDestination != null && !(alternateMessageDestination instanceof Exchange))
        {
            throw new IllegalConfigurationException(String.format(
                    "Alternate exchange '%s' is not a destination of type 'exchange'.", alternateExchangeName));
        }
    }

    @Override
    public void queueDelete(ServerSession session, QueueDelete method)
    {
        String queueName = method.getQueue();
        if(queueName == null || queueName.length()==0)
        {
            exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "No queue name supplied");

        }
        else
        {
            Queue<?> queue = getQueue(session, queueName);


            if (queue == null)
            {
                exception(session, method, ExecutionErrorCode.NOT_FOUND, "No queue " + queueName + " found");
            }
            else
            {
                if(!verifySessionAccess(session, queue))
                {
                    exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
                }
                else if (method.getIfEmpty() && !queue.isEmpty())
                {
                    exception(session, method, ExecutionErrorCode.PRECONDITION_FAILED, "Queue " + queueName + " not empty");
                }
                else if (method.getIfUnused() && !queue.isUnused())
                {
                    // TODO - Error code
                    exception(session, method, ExecutionErrorCode.PRECONDITION_FAILED, "Queue " + queueName + " in use");

                }
                else
                {
                    try
                    {
                        queue.delete();
                    }
                    catch (AccessControlException e)
                    {
                        exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
                    }
                }
            }
        }
    }

    @Override
    public void queuePurge(ServerSession session, QueuePurge method)
    {
        String queueName = method.getQueue();
        if(queueName == null || queueName.length()==0)
        {
            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "No queue name supplied");
        }
        else
        {
            Queue<?> queue = getQueue(session, queueName);

            if (queue == null)
            {
                exception(session, method, ExecutionErrorCode.NOT_FOUND, "No queue " + queueName + " found");
            }
            else
            {
                try
                {
                    queue.clearQueue();
                }
                catch (AccessControlException e)
                {
                    exception(session, method, ExecutionErrorCode.UNAUTHORIZED_ACCESS, e.getMessage());
                }
            }
        }
    }

    @Override
    public void queueQuery(ServerSession session, QueueQuery method)
    {
        QueueQueryResult result = new QueueQueryResult();

        MessageSource source = getMessageSource(session, method.getQueue());

        if(source != null)
        {
            result.setQueue(source.getName());

            if (source instanceof Queue)
            {
                final Queue<?> queue = (Queue<?>) source;
                result.setDurable(queue.isDurable());
                result.setExclusive(queue.isExclusive());
                result.setAutoDelete(queue.getLifetimePolicy() != LifetimePolicy.PERMANENT);
                Map<String, Object> arguments = new LinkedHashMap<>();
                Collection<String> availableAttrs = queue.getAvailableAttributes();
                for(String attrName : availableAttrs)
                {
                    arguments.put(attrName,  queue.getAttribute(attrName));
                }
                result.setArguments(QueueArgumentsConverter.convertModelArgsToWire(arguments));
                result.setMessageCount(queue.getQueueDepthMessages());
                result.setSubscriberCount(queue.getConsumerCount());
            }
            else
            {
                result.setDurable(true);
                result.setExclusive(false);
                result.setAutoDelete(false);
                result.setMessageCount(Integer.MAX_VALUE);
                result.setSubscriberCount(0);
            }
        }

        session.executionResult((int) method.getId(), result);

    }

    @Override
    public void messageSetFlowMode(ServerSession session, MessageSetFlowMode sfm)
    {
        String destination = sfm.getDestination();

        ConsumerTarget_0_10 sub = session.getSubscription(destination);

        if(sub == null)
        {
            exception(session, sfm, ExecutionErrorCode.NOT_FOUND, "not-found: destination '" + destination + "'");
        }
        else if(sub.isFlowModeChangeAllowed())
        {
            sub.setFlowMode(sfm.getFlowMode());
        }
        else
        {
            exception(session, sfm, ExecutionErrorCode.PRECONDITION_FAILED, "destination '" + destination + "' has credit");
        }
    }

    @Override
    public void messageStop(ServerSession session, MessageStop stop)
    {
        String destination = stop.getDestination();

        ConsumerTarget_0_10 sub = session.getSubscription(destination);

        if(sub == null)
        {
            exception(session, stop, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
        }
        else
        {
            sub.stop();
        }

    }

    @Override
    public void messageFlow(ServerSession session, MessageFlow flow)
    {
        String destination = flow.getDestination();

        ConsumerTarget_0_10 sub = session.getSubscription(destination);

        if(sub == null)
        {
            exception(session, flow, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
        }
        else
        {
            sub.addCredit(flow.getUnit(), flow.getValue());
        }

    }

    public void closed(ServerSession session)
    {
        if (_closed.compareAndSet(false, true))
        {
            session.stopSubscriptions();
            session.onClose();
            session.unregisterSubscriptions();
        }
    }

    public void detached(ServerSession session)
    {
        closed(session);
    }

    @Override
    public void init(ServerSession ssn, ProtocolHeader hdr)
    {
        LOGGER.warn("INIT: [{}] {}", ssn, hdr);
    }

    @Override
    public void control(ServerSession ssn, Method method)
    {
        method.dispatch(ssn, this);
    }

    public void command(ServerSession ssn, Method method, boolean processed)
    {
        ssn.identify(method);
        method.dispatch(ssn, this);
        if (processed)
        {
            ssn.processed(method);
        }
    }

    @Override
    public void error(ServerSession ssn, ProtocolError error)
    {
        LOGGER.warn("ERROR: [{}] {}", ssn, error);
    }

    @Override
    public void handle(ServerSession ssn, Method method)
    {
        LOGGER.warn("UNHANDLED: [{}] {}", ssn, method);
    }

    @Override public void sessionRequestTimeout(ServerSession ssn, SessionRequestTimeout t)
    {
        if (t.getTimeout() == 0)
        {
            ssn.setClose(true);
        }
        ssn.sessionTimeout(0); // Always report back an expiry of 0 until it is implemented
    }

    @Override public void sessionAttached(ServerSession ssn, SessionAttached atc)
    {
        ssn.setState(ServerSession.State.OPEN);
        synchronized (ssn.getStateLock())
        {
            ssn.getStateLock().notifyAll();
        }
    }

    @Override public void sessionTimeout(ServerSession ssn, SessionTimeout t)
    {
        // Setting of expiry is not implemented
    }

    @Override public void sessionCompleted(ServerSession ssn, SessionCompleted cmp)
    {
        RangeSet ranges = cmp.getCommands();
        RangeSet known = null;

        if (ranges != null)
        {
            if(ranges.size() == 1)
            {
                Range range = ranges.getFirst();
                boolean advanced = ssn.complete(range.getLower(), range.getUpper());

                if(advanced && cmp.getTimelyReply())
                {
                    known = range;
                }
            }
            else
            {
                if (cmp.getTimelyReply())
                {
                    known = RangeSetFactory.createRangeSet();
                }
                for (Range range : ranges)
                {
                    boolean advanced = ssn.complete(range.getLower(), range.getUpper());
                    if (advanced && known != null)
                    {
                        known.add(range);
                    }
                }
            }
        }
        else if (cmp.getTimelyReply())
        {
            known = RangeSetFactory.createRangeSet();
        }

        if (known != null)
        {
            ssn.sessionKnownCompleted(known);
        }
    }

    @Override public void sessionKnownCompleted(ServerSession ssn, SessionKnownCompleted kcmp)
    {
        RangeSet kc = kcmp.getCommands();
        if (kc != null)
        {
            ssn.knownComplete(kc);
        }
    }

    @Override public void sessionFlush(ServerSession ssn, SessionFlush flush)
    {
        if (flush.getCompleted())
        {
            ssn.flushProcessed();
        }
        if (flush.getConfirmed())
        {
           ssn.flushProcessed();
        }
        if (flush.getExpected())
        {
            ssn.flushExpected();
        }
    }

    @Override public void sessionCommandPoint(ServerSession ssn, SessionCommandPoint scp)
    {
        ssn.commandPoint(scp.getCommandId());
    }

    @Override public void executionResult(ServerSession ssn, ExecutionResult result)
    {
        ssn.result(result.getCommandId(), result.getValue());
    }

    @Override public void executionException(ServerSession ssn, ExecutionException exc)
    {
        ssn.setException(exc);
        LOGGER.error("session exception", exc);
        ssn.closed();
    }

    private static class CommandProcessedAction implements ServerTransaction.Action
    {
        private final ServerSession _serverSession;
        private final Method _method;

        public CommandProcessedAction(final ServerSession serverSession, final Method xfr)
        {
            _serverSession = serverSession;
            _method = xfr;
        }

        @Override
        public void postCommit()
        {
            _serverSession.processed(_method);
        }

        @Override
        public void onRollback()
        {
        }
    }

}