/* Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.qpid.client;

import static org.apache.qpid.transport.Option.BATCH;
import static org.apache.qpid.transport.Option.NONE;
import static org.apache.qpid.transport.Option.SYNC;
import static org.apache.qpid.transport.Option.UNRELIABLE;

import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import javax.jms.Destination;
import javax.jms.JMSException;

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

import org.apache.qpid.QpidException;
import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQDestination.Binding;
import org.apache.qpid.client.AMQDestination.DestSyntax;
import org.apache.qpid.client.failover.FailoverException;
import org.apache.qpid.client.failover.FailoverNoopSupport;
import org.apache.qpid.client.failover.FailoverProtectedOperation;
import org.apache.qpid.client.message.AMQMessageDelegateFactory;
import org.apache.qpid.client.message.UnprocessedMessage_0_10;
import org.apache.qpid.client.messaging.address.AddressHelper;
import org.apache.qpid.client.messaging.address.Link;
import org.apache.qpid.client.messaging.address.Link.SubscriptionQueue;
import org.apache.qpid.client.messaging.address.Node;
import org.apache.qpid.common.AMQPFilterTypes;
import org.apache.qpid.protocol.ErrorCodes;
import org.apache.qpid.transport.*;
import org.apache.qpid.util.Serial;
import org.apache.qpid.util.Strings;

/**
 * This is a 0.10 Session
 */
public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, BasicMessageProducer_0_10>
    implements SessionListener
{

    /**
     * This class logger
     */
    private static final Logger _logger = LoggerFactory.getLogger(AMQSession_0_10.class);

    private final String _name;

    private static class Flusher implements Runnable
    {

        private WeakReference<AMQSession_0_10> session;
        private ScheduledFuture<?> _future;

        public Flusher(AMQSession_0_10 session)
        {
            this.session = new WeakReference<AMQSession_0_10>(session);
        }

        public void setFuture(final ScheduledFuture<?> future)
        {
            _future = future;
        }

        public void run()
        {
            AMQSession_0_10 ssn = session.get();
            if (ssn == null)
            {
                if(_future != null)
                {
                    _future.cancel(false);
                }
            }
            else
            {
                try
                {
                    ssn.flushAcknowledgments(true);
                }
                catch (Exception t)
                {
                    _logger.error("error flushing acks", t);
                }
            }
        }
    }


    /**
     * The underlying QpidSession
     */
    private Session _qpidSession;

    /**
     * The latest qpid Exception that has been raised.
     */
    private Object _currentExceptionLock = new Object();
    private QpidException _currentException;

    // a ref on the qpid connection
    private org.apache.qpid.transport.Connection _qpidConnection;

    private long maxAckDelay = Long.getLong("qpid.session.max_ack_delay", 1000);
    private ScheduledFuture<?> _flushTaskFuture = null;
    private RangeSet unacked = RangeSetFactory.createRangeSet();
    private int unackedCount = 0;

    /**
     * Used to store the range of in tx messages
     */
    private final RangeSet _txRangeSet = RangeSetFactory.createRangeSet();
    private int _txSize = 0;
    private boolean _isHardError = Boolean.getBoolean("qpid.session.legacy_exception_behaviour");
    //--- constructors


    /**
     * Creates a new session on a connection.
     *
     * @param con                     The connection on which to create the session.
     * @param channelId               The unique identifier for the session.
     * @param transacted              Indicates whether or not the session is transactional.
     * @param acknowledgeMode         The acknowledgement mode for the session.
     * @param defaultPrefetchHighMark The maximum number of messages to prefetched before suspending the session.
     * @param defaultPrefetchLowMark  The number of prefetched messages at which to resume the session.
     * @param qpidConnection          The qpid connection
     */
    AMQSession_0_10(org.apache.qpid.transport.Connection qpidConnection, AMQConnection con, int channelId,
                    boolean transacted, int acknowledgeMode, int defaultPrefetchHighMark, int defaultPrefetchLowMark,String name)
    {

        super(con, channelId, transacted, acknowledgeMode, defaultPrefetchHighMark,
              defaultPrefetchLowMark);
        _qpidConnection = qpidConnection;
        _name = name;
        _qpidSession = createSession();

        if (maxAckDelay > 0)
        {
            Flusher flusher = new Flusher(this);
            _flushTaskFuture = con.scheduleTask(flusher, 0, maxAckDelay, TimeUnit.MILLISECONDS);
            flusher.setFuture(_flushTaskFuture);
        }
    }

    protected Session createSession()
    {
        Session qpidSession;
        if (_name == null)
        {
            qpidSession = _qpidConnection.createSession(1);
        }
        else
        {
            qpidSession = _qpidConnection.createSession(_name,1);
        }
        if (isTransacted())
        {
            qpidSession.txSelect();
            qpidSession.setTransacted(true);
        }
        qpidSession.setSessionListener(this);

        return qpidSession;
    }

    private void addUnacked(int id)
    {
        synchronized (unacked)
        {
            unacked.add(id);
            unackedCount++;
        }
    }

    private void clearUnacked()
    {
        synchronized (unacked)
        {
            unacked.clear();
            unackedCount = 0;
        }
    }

    protected Connection getQpidConnection()
    {
        return _qpidConnection;
    }

    //------- overwritten methods of class AMQSession

    void failoverPrep()
    {
        syncDispatchQueue(true);
        clearUnacked();
    }

    /**
     * Acknowledge one or many messages.
     *
     * @param deliveryTag The tag of the last message to be acknowledged.
     * @param multiple    <tt>true</tt> to acknowledge all messages up to and including the one specified by the
     *                    delivery tag, <tt>false</tt> to just acknowledge that message.
     */

    public void acknowledgeMessage(long deliveryTag, boolean multiple)
    {
        if (_logger.isDebugEnabled())
        {
            _logger.debug("Sending ack for delivery tag " + deliveryTag + " on session " + getChannelId());
        }
        // acknowledge this message
        if (multiple)
        {
            for (Long messageTag : getUnacknowledgedMessageTags())
            {
                if( messageTag <= deliveryTag )
                {
                    addUnacked(messageTag.intValue());
                    getUnacknowledgedMessageTags().remove(messageTag);
                }
            }
            //empty the list of unack messages

        }
        else
        {
            addUnacked((int) deliveryTag);
            getUnacknowledgedMessageTags().remove(deliveryTag);
        }

        long prefetch = getAMQConnection().getMaxPrefetch();

        if (unackedCount >= prefetch/2 || maxAckDelay <= 0 || getAcknowledgeMode() == javax.jms.Session.AUTO_ACKNOWLEDGE)
        {
            flushAcknowledgments();
        }
    }

    protected void flushAcknowledgments()
    {
        flushAcknowledgments(false);
    }

    void flushAcknowledgments(boolean setSyncBit)
    {
        synchronized (unacked)
        {
            if (unackedCount > 0)
            {
                messageAcknowledge
                    (unacked, getAcknowledgeMode() != org.apache.qpid.jms.Session.NO_ACKNOWLEDGE,setSyncBit);
                clearUnacked();
            }
        }
    }

    void messageAcknowledge(final RangeSet ranges, final boolean accept)
    {
        messageAcknowledge(ranges,accept,false);
    }

    void messageAcknowledge(final RangeSet ranges, final boolean accept, final boolean setSyncBit)
    {
        final Session ssn = getQpidSession();
        flushProcessed(ranges,accept);
        if (accept)
        {
            ssn.messageAccept(ranges, UNRELIABLE, setSyncBit ? SYNC : NONE);
        }
    }

    /**
     * Flush any outstanding commands. This causes session complete to be sent.
     * @param ranges the range of command ids.
     * @param batch true if batched.
     */
    void flushProcessed(final RangeSet ranges, final boolean batch)
    {
        final Session ssn = getQpidSession();
        for (final Range range : ranges)
        {
            ssn.processed(range);
        }
        ssn.flushProcessed(batch ? BATCH : NONE);
    }

    /**
     * Bind a queue with an exchange.
     *
     * @param queueName    Specifies the name of the queue to bind. If the queue name is empty,
     *                     refers to the current
     *                     queue for the session, which is the last declared queue.
     * @param exchangeName The exchange name.
     * @param routingKey   Specifies the routing key for the binding.
     * @param arguments    0_8 specific
     */
    @Override
    public void sendQueueBind(final String queueName, final String routingKey,
                              final Map<String,Object> arguments, final String exchangeName,
                              final AMQDestination destination, final boolean nowait)
            throws QpidException
    {
        if (destination == null || destination.getDestSyntax() == DestSyntax.BURL)
        {
            if(destination != null)
            {
                for (String rk: destination.getBindingKeys())
                {
                    doSendQueueBind(queueName, exchangeName, arguments, rk);
                }
                if(!Arrays.asList(destination.getBindingKeys()).contains(routingKey))
                {
                    doSendQueueBind(queueName, exchangeName, arguments, routingKey);
                }
            }
            else
            {
                doSendQueueBind(queueName, exchangeName, arguments, routingKey);
            }
        }
        else
        {
            // Leaving this here to ensure the public method bindQueue in AMQSession.java works as expected.
            List<Binding> bindings = new ArrayList<Binding>();
            bindings.addAll(destination.getNode().getBindings());

            String defaultExchange = destination.getAddressType() == AMQDestination.TOPIC_TYPE ?
                                     destination.getAddressName(): "amq.topic";

            for (Binding binding: bindings)
            {
                // Currently there is a bug (QPID-3317) with setting up and tearing down x-bindings for link.
                // The null check below is a way to side step that issue while fixing QPID-4146
                // Note this issue only affects producers.
                if (binding.getQueue() == null && queueName == null)
                {
                    continue;
                }
                String queue = binding.getQueue() == null ? queueName : binding.getQueue();

                String exchange = binding.getExchange() == null ? defaultExchange : binding.getExchange();

                _logger.debug("Binding queue : " + queue +
                              " exchange: " + exchange +
                              " using binding key " + binding.getBindingKey() +
                              " with args " + Strings.printMap(binding.getArgs()));
                doBind(destination, binding, queue, exchange);
            }
        }

        if (!nowait)
        {
            // We need to sync so that we get notify of an error.
            sync();
        }
    }

    private void doSendQueueBind(final String queueName,
                                 final String exchangeName,
                                 final Map args,
                                 final String rk)
    {
        _logger.debug("Binding queue : " + queueName +
                      " exchange: " + exchangeName +
                      " using binding key " + rk);
        getQpidSession().exchangeBind(queueName,
                                      exchangeName,
                                      rk,
                                      args);
    }


    /**
     * Close this session.
     *
     * @param timeout no used / 0_8 specific
     * @throws QpidException
     * @throws FailoverException
     */
    public void sendClose(long timeout) throws QpidException, FailoverException
    {
        cancelTimerTask();
        flushAcknowledgments();
        try
        {
	        getQpidSession().sync();
	        getQpidSession().close();
        }
        catch (SessionException se)
        {
            setCurrentException(se);
        }

        QpidException amqe = getCurrentException();
        if (amqe != null)
        {
            throw amqe;
        }
    }

    /**
     * Create a queue with a given name.
     *
     * @param name       The queue name
     * @param autoDelete If this field is set and the exclusive field is also set,
     *                   then the queue is deleted when the connection closes.
     * @param durable    If set when creating a new queue,
     *                   the queue will be marked as durable.
     * @param exclusive  Exclusive queues can only be used from one connection at a time.
     * @param arguments  Exclusive queues can only be used from one connection at a time.
     * @throws QpidException
     * @throws FailoverException
     */
    public void sendCreateQueue(String name, final boolean autoDelete, final boolean durable,
                                final boolean exclusive, Map<String, Object> arguments) throws QpidException, FailoverException
    {
        getQpidSession().queueDeclare(name, null, arguments, durable ? Option.DURABLE : Option.NONE,
                                      autoDelete ? Option.AUTO_DELETE : Option.NONE,
                                      exclusive ? Option.EXCLUSIVE : Option.NONE);
        // We need to sync so that we get notify of an error.
        sync();
    }

    /**
     * This method asks the broker to redeliver all unacknowledged messages
     *
     * @throws QpidException
     * @throws FailoverException
     */
    public void sendRecover() throws QpidException, FailoverException
    {
        // release all unacked messages
        RangeSet all = RangeSetFactory.createRangeSet();
        RangeSet delivered = gatherRangeSet(getUnacknowledgedMessageTags());
        RangeSet prefetched = gatherRangeSet(getPrefetchedMessageTags());
        for (Iterator<Range> deliveredIter = delivered.iterator(); deliveredIter.hasNext();)
        {
            Range range = deliveredIter.next();
            all.add(range);
        }
        for (Iterator<Range> prefetchedIter = prefetched.iterator(); prefetchedIter.hasNext();)
        {
            Range range = prefetchedIter.next();
            all.add(range);
        }
        flushProcessed(all, false);
        getQpidSession().messageRelease(delivered, Option.SET_REDELIVERED);
        getQpidSession().messageRelease(prefetched);

        // We need to sync so that we get notify of an error.
        sync();
    }

    private RangeSet gatherRangeSet(ConcurrentLinkedQueue<Long> messageTags)
    {
        RangeSet ranges = RangeSetFactory.createRangeSet();
        while (true)
        {
            Long tag = messageTags.poll();
            if (tag == null)
            {
                break;
            }

            ranges.add(tag.intValue());
        }

        return ranges;
    }

    public void releaseForRollback()
    {
        if (_txSize > 0)
        {
            flushProcessed(_txRangeSet, false);
            getQpidSession().messageRelease(_txRangeSet, Option.SET_REDELIVERED);
            _txRangeSet.clear();
            _txSize = 0;
        }
    }

    /**
     * Release (0_8 notion of Reject) an acquired message
     *
     * @param deliveryTag the message ID
     * @param requeue     always true
     */
    public void rejectMessage(long deliveryTag, boolean requeue)
    {
        // The value of requeue is always true
        RangeSet ranges = RangeSetFactory.createRangeSet();
        ranges.add((int) deliveryTag);
        flushProcessed(ranges, false);
        if (requeue)
        {
            getQpidSession().messageRelease(ranges);
        }
        else
        {
            getQpidSession().messageRelease(ranges, Option.SET_REDELIVERED);
        }
        //I don't think we need to sync
    }

    /**
     * Create an 0_10 message consumer
     */
    public BasicMessageConsumer_0_10 createMessageConsumer(final AMQDestination destination, final int prefetchHigh,
                                                      final int prefetchLow, final boolean noLocal,
                                                      final boolean exclusive, String messageSelector,
                                                      final Map<String,Object> rawSelector, final boolean noConsume,
                                                      final boolean autoClose) throws JMSException
    {
        return new BasicMessageConsumer_0_10(getChannelId(), getAMQConnection(), destination, messageSelector, noLocal,
                getMessageFactoryRegistry(), this, rawSelector, prefetchHigh, prefetchLow,
                                             exclusive, getAcknowledgeMode(), noConsume, autoClose);
    }

    /**
     * Bind a queue with an exchange.
     */

    public boolean isQueueBound(final String exchangeName, final String queueName, final String routingKey)
    {
        return isQueueBound(exchangeName,queueName,routingKey,(Map<String,Object>)null);
    }

    public boolean isQueueBound(final AMQDestination destination)
    {
        return isQueueBound(destination.getExchangeName(),destination.getAMQQueueName(),destination.getRoutingKey(),destination.getBindingKeys());
    }

    public boolean isQueueBound(final String exchangeName, final String queueName, final String routingKey, String[] bindingKeys)
    {
        String rk = null;
        if (bindingKeys != null && bindingKeys.length>0)
        {
            rk = bindingKeys[0];
        }
        else if (routingKey != null)
        {
            rk = routingKey;
        }

        return isQueueBound(exchangeName, queueName, rk, (Map<String,Object>)null);
    }

    public boolean isQueueBound(final String exchangeName, final String queueName, final String bindingKey,Map<String,Object> args)
    {
        boolean res;
        ExchangeBoundResult bindingQueryResult =
            getQpidSession().exchangeBound(exchangeName,queueName, bindingKey, args).get();

        if (bindingKey == null)
        {
            res = !(bindingQueryResult.getExchangeNotFound() || bindingQueryResult.getQueueNotFound());
        }
        else
        {
            if (args == null)
            {
                res = !(bindingQueryResult.getExchangeNotFound() || bindingQueryResult.getKeyNotMatched() || bindingQueryResult.getQueueNotFound() || bindingQueryResult
                        .getQueueNotMatched());
            }
            else
            {
                res = !(bindingQueryResult.getExchangeNotFound() || bindingQueryResult.getKeyNotMatched() || bindingQueryResult.getQueueNotFound() || bindingQueryResult
                        .getQueueNotMatched() || bindingQueryResult.getArgsNotMatched());
            }
        }
        return res;
    }

    @Override
    protected boolean isBound(String exchangeName, String amqQueueName, String routingKey)
    {
        return isQueueBound(exchangeName, amqQueueName, routingKey);
    }

    /**
     * This method is invoked when a consumer is created
     * Registers the consumer with the broker
     */
    public void sendConsume(BasicMessageConsumer_0_10 consumer, String queueName,
                            boolean nowait)
            throws QpidException, FailoverException
    {
        queueName = preprocessAddressTopic(consumer, queueName);
        boolean preAcquire = consumer.isPreAcquire();

        AMQDestination destination = consumer.getDestination();
        long capacity = consumer.getCapacity();

        Map<String, Object> arguments = consumer.getArguments();

        Link link = destination.getLink();
        if (link != null && link.getSubscription() != null && link.getSubscription().getArgs() != null)
        {
            arguments.putAll(link.getSubscription().getArgs());
        }

        boolean acceptModeNone = getAcknowledgeMode() == NO_ACKNOWLEDGE;

        String queue = queueName == null ? destination.getAddressName() : queueName;
        String consumerTag = consumer.getConsumerTag();
        getQpidSession().messageSubscribe
            (queue, consumerTag,
             acceptModeNone ? MessageAcceptMode.NONE : MessageAcceptMode.EXPLICIT,
             preAcquire ? MessageAcquireMode.PRE_ACQUIRED : MessageAcquireMode.NOT_ACQUIRED, null, 0, arguments,
             consumer.isExclusive() ? Option.EXCLUSIVE : Option.NONE);


        if (capacity == 0)
        {
            getQpidSession().messageSetFlowMode(consumerTag, MessageFlowMode.CREDIT);
        }
        else
        {
            getQpidSession().messageSetFlowMode(consumerTag, MessageFlowMode.WINDOW);
        }
        getQpidSession().messageFlow(consumerTag, MessageCreditUnit.BYTE, 0xFFFFFFFF,
                                     Option.UNRELIABLE);

        if(capacity > 0 && getDispatcher() != null && (isStarted() || isImmediatePrefetch()))
        {
            // set the flow
            getQpidSession().messageFlow(consumerTag,
                                         MessageCreditUnit.MESSAGE,
                                         capacity,
                                         Option.UNRELIABLE);
        }
        sync();
    }

    /**
     * Create an 0_10 message producer
     */
    public BasicMessageProducer_0_10 createMessageProducer(final Destination destination, final Boolean mandatory,
                                                           final Boolean immediate, final long producerId) throws JMSException
    {
        try
        {
            return new BasicMessageProducer_0_10(getAMQConnection(), (AMQDestination) destination, isTransacted(), getChannelId(), this,
                                             producerId, immediate, mandatory);
        }
        catch (QpidException e)
        {
            throw toJMSException("Error creating producer",e);
        }
        catch(TransportException e)
        {
            throw toJMSException("Exception while creating message producer:" + e.getMessage(), e);
        }

    }

    /**
     * creates an exchange if it does not already exist
     */
    public void sendExchangeDeclare(final String name, final String type, final boolean nowait,
                                    boolean durable, boolean autoDelete, boolean internal) throws QpidException, FailoverException
    {
        //The 'internal' parameter is ignored on the 0-10 path, the protocol does not support it
        sendExchangeDeclare(name, type, null, null, nowait, durable, autoDelete);
    }

    public void sendExchangeDeclare(final String name, final String type, final boolean nowait,
                                    boolean durable, boolean autoDelete, Map<String,Object> arguments, final boolean passive) throws
                                                                                                                              QpidException, FailoverException
    {
        sendExchangeDeclare(name, type, null,
                            arguments,
                            nowait, durable, autoDelete);
    }


    public void sendExchangeDeclare(final String name, final String type,
            final String alternateExchange, final Map<String, Object> args,
            final boolean nowait, boolean durable, boolean autoDelete) throws QpidException
    {
        getQpidSession().exchangeDeclare(
                name,
                type,
                alternateExchange,
                args,
                name.startsWith("amq.") ? Option.PASSIVE : Option.NONE,
                durable ? Option.DURABLE : Option.NONE,
                autoDelete ? Option.AUTO_DELETE : Option.NONE);
        // We need to sync so that we get notify of an error.
        if (!nowait)
        {
            sync();
        }
    }

    /**
     * deletes an exchange
     */
    public void sendExchangeDelete(final String name, final boolean nowait)
                throws QpidException, FailoverException
    {
        getQpidSession().exchangeDelete(name);
        // We need to sync so that we get notify of an error.
        if (!nowait)
        {
            sync();
        }
    }

    /**
     * Declare a queue with the given queueName
     */
    public String send0_10QueueDeclare(final AMQDestination amqd, final boolean noLocal,
                                       final boolean nowait, boolean passive)
            throws QpidException
    {
        String queueName;
        if (amqd.getAMQQueueName() == null)
        {
            // generate a name for this queue
            queueName = createTemporaryQueueName();
            amqd.setQueueName(queueName);
        }
        else
        {
            queueName = amqd.getAMQQueueName();
        }

        if (amqd.getDestSyntax() == DestSyntax.BURL)
        {
            Map<String,Object> arguments = new HashMap<String,Object>();
            if (noLocal)
            {
                arguments.put(AddressHelper.NO_LOCAL, true);
            }

            getQpidSession().queueDeclare(queueName, "" , arguments,
                                          amqd.isAutoDelete() ? Option.AUTO_DELETE : Option.NONE,
                                          amqd.isDurable() ? Option.DURABLE : Option.NONE,
                                          amqd.isExclusive() ? Option.EXCLUSIVE : Option.NONE,
                                          passive ? Option.PASSIVE : Option.NONE);
        }
        else
        {
            // This code is here to ensure address based destination work with the declareQueue public method in AMQSession.java
            Node node = amqd.getNode();
            Map<String,Object> arguments = new HashMap<String,Object>();
            arguments.putAll(node.getDeclareArgs());
            if (arguments.get(AddressHelper.NO_LOCAL) == null)
            {
                arguments.put(AddressHelper.NO_LOCAL, noLocal);
            }
            getQpidSession().queueDeclare(queueName, node.getAlternateExchange() ,
                    arguments,
                    node.isAutoDelete() ? Option.AUTO_DELETE : Option.NONE,
                    node.isDurable() ? Option.DURABLE : Option.NONE,
                    node.isExclusive() ? Option.EXCLUSIVE : Option.NONE);
        }

        // passive --> false
        if (!nowait)
        {
            // We need to sync so that we get notify of an error.
            sync();
        }
        return queueName;
    }

    /**
     * deletes a queue
     */
    public void sendQueueDelete(final String queueName) throws QpidException, FailoverException
    {
        getQpidSession().queueDelete(queueName);
        // We need to sync so that we get notify of an error.
        sync();
    }

    /**
     * Activate/deactivate the message flow for all the consumers of this session.
     */
    public void sendSuspendChannel(boolean suspend) throws QpidException, FailoverException
    {
        if (suspend)
        {
            for (BasicMessageConsumer consumer : getConsumers())
            {
                getQpidSession().messageStop(consumer.getConsumerTag(),
                                             Option.UNRELIABLE);
            }
            sync();
        }
        else
        {
            for (BasicMessageConsumer_0_10 consumer : getConsumers())
            {
                String consumerTag = consumer.getConsumerTag();
                //only set if msg list is null
                try
                {
                    long capacity = consumer.getCapacity();

                    if (capacity == 0)
                    {
                        if (consumer.getMessageListener() != null)
                        {
                            getQpidSession().messageFlow(consumerTag,
                                                         MessageCreditUnit.MESSAGE, 1,
                                                         Option.UNRELIABLE);
                        }
                    }
                    else
                    {
                        getQpidSession()
                            .messageFlow(consumerTag, MessageCreditUnit.MESSAGE,
                                         capacity,
                                         Option.UNRELIABLE);
                    }
                    getQpidSession()
                        .messageFlow(consumerTag, MessageCreditUnit.BYTE, 0xFFFFFFFF,
                                     Option.UNRELIABLE);
                }
                catch (Exception e)
                {
                    throw new AMQException(ErrorCodes.INTERNAL_ERROR, "Error while trying to get the listener", e);
                }
            }
        }
        // We need to sync so that we get notify of an error.
        sync();
    }


    public void sendRollback() throws QpidException, FailoverException
    {
        getQpidSession().txRollback();
        // We need to sync so that we get notify of an error.
        sync();
    }

    //------ Private methods
    /**
     * Access to the underlying Qpid Session
     *
     * @return The associated Qpid Session.
     */
    protected Session getQpidSession()
    {
        return _qpidSession;
    }


    /**
     * Get the latest thrown exception.
     *
     * @throws SessionException get the latest thrown error.
     */
    public QpidException getCurrentException()
    {
        QpidException amqe = null;
        synchronized (_currentExceptionLock)
        {
            if (_currentException != null)
            {
                amqe = _currentException;
                _currentException = null;
            }
        }
        return amqe;
    }

    public void opened(Session ssn) {}

    public void resumed(Session ssn)
    {
        _qpidConnection = ssn.getConnection();
    }

    public void message(Session ssn, MessageTransfer xfr)
    {
        messageReceived(new UnprocessedMessage_0_10(xfr));
    }

    public void exception(Session ssn, SessionException exc)
    {
        setCurrentException(exc);
    }

    public void closed(Session ssn)
    {
        try
        {
            super.closed(null);
            if (_flushTaskFuture != null)
            {
                _flushTaskFuture.cancel(false);
                _flushTaskFuture = null;
            }
        } catch (Exception e)
        {
            _logger.error("Error closing JMS session", e);
        }
    }

    public QpidException getLastException()
    {
        return getCurrentException();
    }

    @Override
    protected String declareQueue(final AMQDestination amqd,
                                          final boolean noLocal, final boolean nowait, final boolean passive)
            throws QpidException
    {
        return new FailoverNoopSupport<String, QpidException>(
                new FailoverProtectedOperation<String, QpidException>()
                {
                    public String execute() throws QpidException, FailoverException
                    {
                        // Generate the queue name if the destination indicates that a client generated name is to be used.
                        if (amqd.isNameRequired())
                        {
                            String bindingKey = "";
                            for(String key : amqd.getBindingKeys())
                            {
                               bindingKey = bindingKey + "_" + key;
                            }
                            amqd.setQueueName(bindingKey + "@"
                                    + amqd.getExchangeName() + "_" + UUID.randomUUID());
                        }
                        return send0_10QueueDeclare(amqd, noLocal, nowait, passive);
                    }
                }, getAMQConnection()).execute();
    }

    protected Long requestQueueDepth(AMQDestination amqd, boolean sync)
    {
        flushAcknowledgments();
        if (sync)
        {
            getQpidSession().sync();
        }
        return getQpidSession().queueQuery(amqd.getQueueName()).get().getMessageCount();
    }


    /**
     * Store non committed messages for this session
     * @param id
     */
    @Override protected void addDeliveredMessage(long id)
    {
        _txRangeSet.add((int) id);
        _txSize++;
    }

    /**
     * With 0.10 messages are consumed with window mode, we must send a completion
     * before the window size is reached so credits don't dry up.
     */
    protected void sendTxCompletionsIfNecessary()
    {
        // this is a heuristic, we may want to have that configurable
        if (_txSize > 0 && (getAMQConnection().getMaxPrefetch() == 1 ||
                getAMQConnection().getMaxPrefetch() != 0 && _txSize % (getAMQConnection().getMaxPrefetch() / 2) == 0))
        {
            // send completed so consumer credits don't dry up
            messageAcknowledge(_txRangeSet, false);
        }
    }

    public void commitImpl() throws QpidException, FailoverException, TransportException
    {
        if( _txSize > 0 )
        {
            messageAcknowledge(_txRangeSet, true);
            _txRangeSet.clear();
            _txSize = 0;
        }

        getQpidSession().setAutoSync(true);
        try
        {
            getQpidSession().txCommit();
        }
        finally
        {
            getQpidSession().setAutoSync(false);
        }
        // We need to sync so that we get notify of an error.
        sync();
    }

    protected final boolean tagLE(long tag1, long tag2)
    {
        return Serial.le((int) tag1, (int) tag2);
    }

    protected final boolean updateRollbackMark(long currentMark, long deliveryTag)
    {
        return Serial.lt((int) currentMark, (int) deliveryTag);
    }

    public void sync() throws QpidException
    {
        try
        {
            getQpidSession().sync();
        }
        catch (SessionException se)
        {
            setCurrentException(se);
        }

        QpidException amqe = getCurrentException();
        if (amqe != null)
        {
            throw amqe;
        }
    }

    public void setCurrentException(SessionException se)
    {
        QpidException amqe;
        synchronized (_currentExceptionLock)
        {
            ExecutionException ee = se.getException();
            int code = ErrorCodes.INTERNAL_ERROR;
            if (ee != null)
            {
                code = ee.getErrorCode().getValue();
            }
            amqe = new AMQException(code, _isHardError, se.getMessage(), se.getCause());
            _currentException = amqe;
        }
        if (!_isHardError)
        {
            cancelTimerTask();
            stopDispatcherThread();
            try
            {
                closed(amqe);
            }
            catch(Exception e)
            {
                _logger.warn("Error closing session", e);
            }
            getAMQConnection().exceptionReceived(amqe);
        }
        else
        {
            getAMQConnection().closed(amqe);
        }
    }

    public AMQMessageDelegateFactory getMessageDelegateFactory()
    {
        return AMQMessageDelegateFactory.FACTORY_0_10;
    }

    @Override
    public boolean isExchangeExist(AMQDestination dest,boolean assertNode) throws QpidException
    {
        boolean match = true;
        ExchangeQueryResult result = getQpidSession().exchangeQuery(dest.getAddressName(), Option.NONE).get();
        match = !result.getNotFound();
        Node node = dest.getNode();

        if (match)
        {
            if (assertNode)
            {
                match =  (result.getDurable() == node.isDurable()) &&
                         (node.getExchangeType() != null &&
                          node.getExchangeType().equals(result.getType())) &&
                         (matchProps(result.getArguments(),node.getDeclareArgs()));
            }
            else
            {
                _logger.debug("Setting Exchange type " + result.getType());
                node.setExchangeType(result.getType());
                dest.setExchangeClass(result.getType());
            }
        }

        if (assertNode)
        {
            if (!match)
            {
                throw new QpidException("Assert failed for address : " + dest  +", Result was : " + result);
            }
        }

        return match;
    }

    @Override
    public boolean isQueueExist(AMQDestination dest, boolean assertNode) throws QpidException
    {
        Node node = dest.getNode();
        return isQueueExist(dest.getAddressName(), assertNode,
                            node.isDurable(), node.isAutoDelete(),
                            node.isExclusive(), node.getDeclareArgs());
    }
    
    public boolean isQueueExist(String queueName, boolean assertNode,
                                boolean durable, boolean autoDelete,
                                boolean exclusive, Map<String, Object> args) throws QpidException
    {    
        boolean match = true;
        try
        {
            QueueQueryResult result = getQpidSession().queueQuery(queueName, Option.NONE).get();
            match = queueName.equals(result.getQueue());

            if (match && assertNode)
            {
                match = (result.getDurable() == durable) &&
                         (result.getAutoDelete() == autoDelete) &&
                         (result.getExclusive() == exclusive) &&
                         (matchProps(result.getArguments(),args));
            }

            if (assertNode)
            {
                if (!match)
                {
                    throw new QpidException("Assert failed for queue : " + queueName  +", Result was : " + result);
                }
            }
        }
        catch(SessionException e)
        {
            final ExecutionException underlying = e.getException();
            if (underlying == null)
            {
                throw new QpidException("Error querying queue", e);
            }
            else
            {
                if (underlying.getErrorCode() == ExecutionErrorCode.RESOURCE_DELETED
                    || underlying.getErrorCode() == ExecutionErrorCode.NOT_FOUND)
                {
                    match = false;
                }
                else
                {
                    throw new AMQException(underlying.getErrorCode().getValue(),
                                           "Error querying queue", e);
                }
            }
        }
        return match;
    }

    private boolean matchProps(Map<String,Object> target,Map<String,Object> source)
    {
        boolean match = true;
        for (String key: source.keySet())
        {
            match = target.containsKey(key) &&
                    (target.get(key).equals(source.get(key))
                    || (target.get(key) instanceof Number)
                        && source.get(key) instanceof Number &&
                        (((Number) target.get(key)).longValue()) == (((Number) source.get(key)).longValue()));

            if (!match)
            {
                StringBuffer buf = new StringBuffer();
                buf.append("Property given in address did not match with the args sent by the broker.");
                buf.append(" Expected { ").append(key).append(" : ").append(source.get(key)).append(" }, ");
                buf.append(" Actual { ").append(key).append(" : ").append(target.get(key)).append(" }");
                _logger.debug(buf.toString());
                return match;
            }
        }

        return match;
    }

    @Override
    public int resolveAddressType(AMQDestination dest) throws QpidException
    {
       int type = dest.getAddressType();
       String name = dest.getAddressName();
       if (type != AMQDestination.UNKNOWN_TYPE)
       {
           return type;
       }
       else
       {
            ExchangeBoundResult result = getQpidSession().exchangeBound(name,name,null,null).get();
            if (result.getQueueNotFound() && result.getExchangeNotFound()) {
                //neither a queue nor an exchange exists with that name; treat it as a queue
                type = AMQDestination.QUEUE_TYPE;
            } else if (result.getExchangeNotFound()) {
                //name refers to a queue
                type = AMQDestination.QUEUE_TYPE;
            } else if (result.getQueueNotFound()) {
                //name refers to an exchange
                type = AMQDestination.TOPIC_TYPE;
            } else {
                //both a queue and exchange exist for that name
                throw new QpidException("Ambiguous address, please specify queue or topic as node type");
            }
            dest.setAddressType(type);
            return type;
        }
    }

    @Override
    void createSubscriptionQueue(AMQDestination dest, boolean noLocal, String messageSelector) throws QpidException
    {
        Link link = dest.getLink();
        String queueName = dest.getQueueName();

        if (queueName == null)
        {
            queueName = link.getName() == null ? "TempQueue" + UUID.randomUUID() : link.getName();
            dest.setQueueName(queueName);
        }

        SubscriptionQueue queueProps = link.getSubscriptionQueue();
        Map<String,Object> arguments = queueProps.getDeclareArgs();
        if (!arguments.containsKey((AddressHelper.NO_LOCAL)))
        {
            arguments.put(AddressHelper.NO_LOCAL, noLocal);
        }

        if (link.isDurable() && queueName.startsWith("TempQueue"))
        {
            throw new QpidException("You cannot mark a subscription queue as durable without providing a name for the link.");
        }

        getQpidSession().queueDeclare(queueName,
                queueProps.getAlternateExchange(), arguments,
                queueProps.isAutoDelete() ? Option.AUTO_DELETE : Option.NONE,
                link.isDurable() ? Option.DURABLE : Option.NONE,
                queueProps.isExclusive() ? Option.EXCLUSIVE : Option.NONE);

        Map<String,Object> bindingArguments = new HashMap<String, Object>();
        bindingArguments.put(AMQPFilterTypes.JMS_SELECTOR.getValue(), messageSelector == null ? "" : messageSelector);
        getQpidSession().exchangeBind(queueName,
                              dest.getAddressName(),
                              dest.getSubject(),
                              bindingArguments);
    }

    protected void acknowledgeImpl()
    {
        RangeSet ranges = gatherRangeSet(getUnacknowledgedMessageTags());

        if(ranges.size() > 0 )
        {
            messageAcknowledge(ranges, true);
            if (getAMQConnection().getSyncClientAck())
            {
                getQpidSession().sync();
            }
        }
    }

    @Override
    void resubscribe() throws QpidException
    {
        // Clear txRangeSet/unacknowledgedMessageTags so we don't complete commands corresponding to
        //messages that came from the old broker.
        _txRangeSet.clear();
        _txSize = 0;

        super.resubscribe();
        getQpidSession().sync();
    }

    @Override
    void stop() throws QpidException
    {
        // Stop the server delivering messages to this session.
        suspendChannelIfNotClosing();
        drainDispatchQueueWithDispatcher();
        stopExistingDispatcher();

        for (BasicMessageConsumer consumer : getConsumers())
        {
            List<Long> tags = consumer.drainReceiverQueueAndRetrieveDeliveryTags();
            getPrefetchedMessageTags().addAll(tags);
        }

        RangeSet delivered = gatherRangeSet(getUnacknowledgedMessageTags());
		RangeSet prefetched = gatherRangeSet(getPrefetchedMessageTags());
		RangeSet all = RangeSetFactory.createRangeSet(delivered.size()
					+ prefetched.size());

		for (Iterator<Range> deliveredIter = delivered.iterator(); deliveredIter.hasNext();)
		{
			Range range = deliveredIter.next();
			all.add(range);
		}

		for (Iterator<Range> prefetchedIter = prefetched.iterator(); prefetchedIter.hasNext();)
		{
			Range range = prefetchedIter.next();
			all.add(range);
		}

		flushProcessed(all, false);
		getQpidSession().messageRelease(delivered,Option.SET_REDELIVERED);
		getQpidSession().messageRelease(prefetched);
		sync();
    }

    @Override
    public boolean isFlowBlocked()
    {
        return _qpidSession.isFlowBlocked();
    }

    @Override
    public void setFlowControl(boolean active)
    {
        // Supported by 0-8..0-9-1 only
        throw new UnsupportedOperationException("Operation not supported by this protocol");
    }

    private void cancelTimerTask()
    {
        if (_flushTaskFuture != null)
        {
            _flushTaskFuture.cancel(false);
            _flushTaskFuture = null;
        }
    }

    @Override
    protected void handleQueueNodeCreation(AMQDestination dest, boolean noLocal) throws QpidException
    {
        Node node = dest.getNode();
        Map<String,Object> arguments = node.getDeclareArgs();
        if (!arguments.containsKey((AddressHelper.NO_LOCAL)))
        {
            arguments.put(AddressHelper.NO_LOCAL, noLocal);
        }
        getQpidSession().queueDeclare(dest.getAddressName(),
                node.getAlternateExchange(), arguments,
                node.isAutoDelete() ? Option.AUTO_DELETE : Option.NONE,
                node.isDurable() ? Option.DURABLE : Option.NONE,
                node.isExclusive() ? Option.EXCLUSIVE : Option.NONE);

        createBindings(dest, dest.getNode().getBindings());
        sync();
    }

    @Override
    void handleExchangeNodeCreation(AMQDestination dest) throws QpidException
    {
        Node node = dest.getNode();
        sendExchangeDeclare(dest.getAddressName(),
                node.getExchangeType(),
                node.getAlternateExchange(),
                node.getDeclareArgs(),
                false,
                node.isDurable(),
                node.isAutoDelete());

        // If bindings are specified without a queue name and is called by the producer,
        // the broker will send an exception as expected.
        createBindings(dest, dest.getNode().getBindings());
        sync();
    }

    protected void doBind(final AMQDestination dest, final Binding binding, final String queue, final String exchange)
    {
        getQpidSession().exchangeBind(queue,
                                      exchange,
                                      binding.getBindingKey(),
                                      binding.getArgs());
    }

    void handleLinkDelete(AMQDestination dest) throws QpidException
    {
        // We need to destroy link bindings
        String defaultExchangeForBinding = dest.getAddressType() == AMQDestination.TOPIC_TYPE ? dest
                .getAddressName() : "amq.topic";

        String defaultQueueName = null;
        if (AMQDestination.QUEUE_TYPE == dest.getAddressType())
        {
            defaultQueueName = dest.getQueueName();
        }
        else
        {
            defaultQueueName = dest.getLink().getName() != null ? dest.getLink().getName() : dest.getQueueName();
        }

        for (Binding binding: dest.getLink().getBindings())
        {
            String queue = binding.getQueue() == null?
                    defaultQueueName: binding.getQueue();

            String exchange = binding.getExchange() == null ?
                        defaultExchangeForBinding :
                        binding.getExchange();

            if (_logger.isDebugEnabled())
            {
                _logger.debug("Unbinding queue : " + queue +
                         " exchange: " + exchange +
                         " using binding key " + binding.getBindingKey() +
                         " with args " + Strings.printMap(binding.getArgs()));
            }
            getQpidSession().exchangeUnbind(queue, exchange,
                                            binding.getBindingKey());
        }
    }

    void deleteSubscriptionQueue(AMQDestination dest) throws QpidException
    {
        // We need to delete the subscription queue.
        if (dest.getAddressType() == AMQDestination.TOPIC_TYPE &&
            dest.getLink().getSubscriptionQueue().isExclusive() &&
            isQueueExist(dest.getQueueName(), false, false, false, false, null))
        {
            getQpidSession().queueDelete(dest.getQueueName());
        }
    }

    @Override
    void handleNodeDelete(AMQDestination dest) throws QpidException
    {
        if (AMQDestination.TOPIC_TYPE == dest.getAddressType())
        {
            if (isExchangeExist(dest,false))
            {
                getQpidSession().exchangeDelete(dest.getAddressName());
                setUnresolved(dest);
            }
        }
        else
        {
            if (isQueueExist(dest,false))
            {
                getQpidSession().queueDelete(dest.getAddressName());
                setUnresolved(dest);
            }
        }
    }
}