/*
 *
 * 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.
 *
 */
using System;
using System.Collections;
using System.Text.RegularExpressions;
using System.Threading;
using log4net;
using Apache.Qpid.Buffer;
using Apache.Qpid.Client.Message;
using Apache.Qpid.Client.Util;
using Apache.Qpid.Collections;
using Apache.Qpid.Framing;
using Apache.Qpid.Messaging;
using Apache.Qpid.Protocol;

namespace Apache.Qpid.Client
{
    /// <summary>
    /// <p/><table id="crc"><caption>CRC Card</caption>
    /// <tr><th> Responsibilities <th> Collaborations
    /// <tr><td> Declare queues.
    /// <tr><td> Declare exchanges.
    /// <tr><td> Bind queues to exchanges.
    /// <tr><td> Create messages.
    /// <tr><td> Set up message consumers on the channel.
    /// <tr><td> Set up message producers on the channel.
    /// <tr><td> Commit the current transaction.
    /// <tr><td> Roll-back the current transaction.
    /// <tr><td> Close the channel.
    /// </table>
    /// </summary>
    public class AmqChannel : Closeable, IChannel
    {
        private static readonly ILog _logger = LogManager.GetLogger(typeof(AmqChannel));

        internal const int BASIC_CONTENT_TYPE = 60;

        public const int DEFAULT_PREFETCH_HIGH_MARK = 5000;

        public const int DEFAULT_PREFETCH_LOW_MARK = 2500;

        private static int _nextSessionNumber = 0;

        private AMQConnection _connection;

        private int _sessionNumber;

        private bool _suspended;

        private object _suspensionLock = new object();
        
        // Used in the consume method. We generate the consume tag on the client so that we can use the nowait feature.
        private int _nextConsumerNumber = 1;

        private bool _transacted;

        private AcknowledgeMode _acknowledgeMode;

        private ushort _channelId;

        private int _defaultPrefetchHighMark = DEFAULT_PREFETCH_HIGH_MARK;

        private int _defaultPrefetchLowMark = DEFAULT_PREFETCH_LOW_MARK;

        private FlowControlQueue _queue;

        private Dispatcher _dispatcher;

        private MessageFactoryRegistry _messageFactoryRegistry;

        /// <summary> Holds all of the producers created by this channel. </summary>
        private Hashtable _producers = Hashtable.Synchronized(new Hashtable());

        /// <summary> Holds all of the consumers created by this channel. </summary>
        private Hashtable _consumers = Hashtable.Synchronized(new Hashtable());

        private ArrayList _replayFrames = new ArrayList();

        /// <summary>
        /// The counter of the _next producer id. This id is generated by the session and used only to allow the
        /// producer to identify itself to the session when deregistering itself.
        ///
        /// Access to this id does not require to be synchronized since according to the JMS specification only one
        /// thread of control is allowed to create producers for any given session instance.
        /// </summary>
        private long _nextProducerId;

        /// <summary>
        /// Initializes a new instance of the <see cref="AmqChannel"/> class.
        /// </summary>
        /// <param name="con">The connection.</param>
        /// <param name="channelId">The channel id.</param>
        /// <param name="transacted">if set to <c>true</c> [transacted].</param>
        /// <param name="acknowledgeMode">The acknowledge mode.</param>
        /// <param name="defaultPrefetchHigh">Default prefetch high value</param>
        /// <param name="defaultPrefetchLow">Default prefetch low value</param>
        internal AmqChannel(AMQConnection con, ushort channelId, bool transacted, AcknowledgeMode acknowledgeMode, 
                            int defaultPrefetchHigh, int defaultPrefetchLow)
            : this()
        {
            _sessionNumber = Interlocked.Increment(ref _nextSessionNumber);
            _connection = con;
            _transacted = transacted;
            
            if ( transacted )
            {
                _acknowledgeMode = AcknowledgeMode.SessionTransacted;
            } 
            else
            {
                _acknowledgeMode = acknowledgeMode;
            }

            _channelId = channelId;
            _defaultPrefetchHighMark = defaultPrefetchHigh;
            _defaultPrefetchLowMark = defaultPrefetchLow;

            if ( _acknowledgeMode == AcknowledgeMode.NoAcknowledge )
            {
                _queue = new FlowControlQueue(_defaultPrefetchLowMark, _defaultPrefetchHighMark,
                                              new ThresholdMethod(OnPrefetchLowMark),
                                              new ThresholdMethod(OnPrefetchHighMark));
            } 
            else
            {
                // low and upper are the same
                _queue = new FlowControlQueue(_defaultPrefetchHighMark, _defaultPrefetchHighMark,
                                              null, null);
            }
        }

        private AmqChannel()
        {
            _messageFactoryRegistry = MessageFactoryRegistry.NewDefaultRegistry();
        }

        /// <summary>
        /// Acknowledge mode for messages received.
        /// </summary>
        public AcknowledgeMode AcknowledgeMode
        {
            get
            {
                CheckNotClosed();
                return _acknowledgeMode;
            }
        }

        /// <summary>
        /// True if the channel should use transactions.
        /// </summary>
        public bool Transacted
        {
            get
            {
                CheckNotClosed();
                return _transacted;
            }
        }

        /// <summary>
        /// Prefetch value to be used as the default for 
        /// consumers created on this channel.
        /// </summary>
        public int DefaultPrefetch
        {
            get { return DefaultPrefetchHigh; }
        }

        /// <summary>
        /// Prefetch low value to be used as the default for 
        /// consumers created on this channel.
        /// </summary>
        public int DefaultPrefetchLow
        {
            get { return _defaultPrefetchLowMark; }
        }

        /// <summary>
        /// Prefetch high value to be used as the default for 
        /// consumers created on this channel.
        /// </summary>
        public int DefaultPrefetchHigh
        {
            get { return _defaultPrefetchHighMark; }
        }

        /// <summary> Indicates whether or not this channel is currently suspended. </summary>
        public bool IsSuspended
        {
            get { return _suspended; }
        }

        /// <summary> Provides the channels number within the the connection. </summary>
        public ushort ChannelId
        {
            get { return _channelId; }
        }

        /// <summary> Provides the connection that this channel runs over. </summary>
        public AMQConnection Connection
        {
            get { return _connection; }
        }

        /// <summary>
        /// Declare a new exchange.
        /// </summary>
        /// <param name="exchangeName">Name of the exchange</param>
        /// <param name="exchangeClass">Class of the exchange, from <see cref="ExchangeClassConstants"/></param>
        public void DeclareExchange(String exchangeName, String exchangeClass)
        {
            _logger.Debug(string.Format("DeclareExchange vame={0} exchangeClass={1}", exchangeName, exchangeClass));

            DeclareExchange(_channelId, 0, exchangeName, exchangeClass, false, false, false, false, true, null);
        }

        /// <summary>
        /// Declare a new exchange using the default exchange class.
        /// </summary>
        /// <param name="exchangeName">Name of the exchange</param>
        public void DeleteExchange(string exchangeName)
        {
            throw new NotImplementedException();
        }

        /// <summary>
        /// Declare a new queue with the specified set of arguments.
        /// </summary>
        /// <param name="queueName">Name of the queue</param>
        /// <param name="isDurable">True if the queue should be durable</param>
        /// <param name="isExclusive">True if the queue should be exclusive to this channel</param>
        /// <param name="isAutoDelete">True if the queue should be deleted when the channel closes</param>
        public void DeclareQueue(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete)
        {
            DoQueueDeclare(queueName, isDurable, isExclusive, isAutoDelete, null);
        }

        /// <summary>
        /// Declare a new queue with the specified set of arguments.
        /// </summary>
        /// <param name="queueName">Name of the queue</param>
        /// <param name="isDurable">True if the queue should be durable</param>
        /// <param name="isExclusive">True if the queue should be exclusive to this channel</param>
        /// <param name="isAutoDelete">True if the queue should be deleted when the channel closes</param>
        /// <param name="args">Optional arguments to Queue.Declare</param>
        public void DeclareQueue(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete, IFieldTable args)
        {
            DoQueueDeclare(queueName, isDurable, isExclusive, isAutoDelete, args);
        }

        /// <summary>
        /// Delete a queue with the specifies arguments.
        /// </summary>
        /// <param name="queueName">Name of the queue to delete</param>
        /// <param name="ifUnused">If true, the queue will not deleted if it has no consumers</param>
        /// <param name="ifEmpty">If true, the queue will not deleted if it has no messages</param>
        /// <param name="noWait">If true, the server will not respond to the method</param>
        public void DeleteQueue(string queueName, bool ifUnused, bool ifEmpty, bool noWait)
        {
            DoDeleteQueue(queueName, ifUnused, ifEmpty, noWait);
        }

        /// <summary>
        /// Generate a new Unique name to use for a queue.
        /// </summary>
        /// <returns>A unique name to this channel</returns>
        public string GenerateUniqueName()
        {
            string result = _connection.ProtocolSession.GenerateQueueName();            
            return Regex.Replace(result, "[^a-z0-9_]", "_");
        }

        /// <summary>
        /// Removes all messages from a queue.
        /// </summary>
        /// <param name="queueName">Name of the queue to delete</param>
        /// <param name="noWait">If true, the server will not respond to the method</param>
        public void PurgeQueue(string queueName, bool noWait)
        {
            DoPurgeQueue(queueName, noWait);
        }
        
        /// <summary>
        /// Bind a queue to the specified exchange.
        /// </summary>
        /// <param name="queueName">Name of queue to bind</param>
        /// <param name="exchangeName">Name of exchange to bind to</param>
        /// <param name="routingKey">Routing key</param>
        public void Bind(string queueName, string exchangeName, string routingKey)
        {
            DoBind(queueName, exchangeName, routingKey, new FieldTable());
        }

        /// <summary>
        /// Bind a queue to the specified exchange.
        /// </summary>
        /// <param name="queueName">Name of queue to bind</param>
        /// <param name="exchangeName">Name of exchange to bind to</param>
        /// <param name="routingKey">Routing key</param>
        /// <param name="args">Table of arguments for the binding. Used to bind with a Headers Exchange</param>
        public void Bind(string queueName, string exchangeName, string routingKey, IFieldTable args)
        {
            DoBind(queueName, exchangeName, routingKey, (FieldTable)args);            
        }

        /// <summary>
        /// Create a new empty message with no body.
        /// </summary>
        /// <returns>The new message</returns>
        public IMessage CreateMessage()
        {
            return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
        }

        /// <summary>
        /// Create a new message of the specified MIME type.
        /// </summary>
        /// <param name="mimeType">The mime type to create</param>
        /// <returns>The new message</returns>
        public IMessage CreateMessage(string mimeType)
        {
            return _messageFactoryRegistry.CreateMessage(mimeType);
        }

        /// <summary>
        /// Creates a new message for bytes (application/octet-stream).
        /// </summary>
        /// <returns>The new message</returns>
        public IBytesMessage CreateBytesMessage()
        {
            return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
        }

        /// <summary>
        /// Creates a new text message (text/plain) with empty content.
        /// </summary>
        /// <returns>The new message</returns>
        public ITextMessage CreateTextMessage()
        {
            return CreateTextMessage(String.Empty);
        }

        /// <summary>
        /// Creates a new text message (text/plain) with a body.
        /// </summary>
        /// <param name="text">Initial body of the message</param>
        /// <returns>The new message</returns>
        public ITextMessage CreateTextMessage(string text)
        {
            ITextMessage msg = (ITextMessage)_messageFactoryRegistry.CreateMessage("text/plain");
            msg.Text = text;
            return msg;
        }

        /// <summary>
        /// Creates a new Consumer using the builder pattern.
        /// </summary>
        /// <param name="queueName">Name of queue to receive messages from</param>
        /// <returns>The builder object</returns>
        public MessageConsumerBuilder CreateConsumerBuilder(string queueName)
        {
            return new MessageConsumerBuilder(this, queueName);
        }

        /// <summary>
        /// Creates a new consumer.
        /// </summary>
        /// <param name="queueName">Name of queue to receive messages from</param>
        /// <param name="prefetchLow">Low prefetch value</param>
        /// <param name="prefetchHigh">High prefetch value</param>
        /// <param name="noLocal">If true, messages sent on this channel will not be received by this consumer</param>
        /// <param name="exclusive">If true, the consumer opens the queue in exclusive mode</param>
        /// <returns>The new consumer</returns>
        public IMessageConsumer CreateConsumer(string queueName,
                                               int prefetchLow,
                                               int prefetchHigh,
                                               bool noLocal,
                                               bool exclusive)
        {
            _logger.Debug(String.Format("CreateConsumer queueName={0} prefetchLow={1} prefetchHigh={2} noLocal={3} exclusive={4} ",
                                        queueName, prefetchLow, prefetchHigh, noLocal, exclusive));

            return CreateConsumerImpl(queueName, prefetchLow, prefetchHigh, noLocal, exclusive, false);
        }
        
                /// <summary>
        /// Creates a new consumer.
        /// </summary>
        /// <param name="queueName">Name of queue to receive messages from</param>
        /// <param name="prefetchLow">Low prefetch value</param>
        /// <param name="prefetchHigh">High prefetch value</param>
        /// <param name="noLocal">If true, messages sent on this channel will not be received by this consumer</param>
        /// <param name="exclusive">If true, the consumer opens the queue in exclusive mode</param>
        /// <param name="browse">If true, the consumer only browses and does not consume messages</param>
        /// <returns>The new consumer</returns>
        public IMessageConsumer CreateConsumer(string queueName,
                                               int prefetchLow,
                                               int prefetchHigh,
                                               bool noLocal,
                                               bool exclusive,
                                               bool browse)
        {
            _logger.Debug(String.Format("CreateConsumer queueName={0} prefetchLow={1} prefetchHigh={2} noLocal={3} exclusive={4} browse={5}",
                                        queueName, prefetchLow, prefetchHigh, noLocal, exclusive, browse));

            return CreateConsumerImpl(queueName, prefetchLow, prefetchHigh, noLocal, exclusive, browse);
        }
        
        
        /// <summary>
        /// Unsubscribe from a queue.
        /// </summary>
        /// <param name="subscriptionName">Subscription name</param>
        public void Unsubscribe(String name)
        {
            throw new NotImplementedException();
        }

        /// <summary>
        /// Create a new message publisher using the builder pattern.
        /// </summary>
        /// <returns>The builder object</returns>
        public MessagePublisherBuilder CreatePublisherBuilder()
        {
            return new MessagePublisherBuilder(this);
        }
        
        /// <summary>
        /// Create a new message publisher.
        /// </summary>
        /// <param name="exchangeName">Name of exchange to publish to</param>
        /// <param name="routingKey">Routing key</param>
        /// <param name="deliveryMode">Default delivery mode</param>
        /// <param name="timeToLive">Default TTL time of messages</param>
        /// <param name="immediate">If true, sent immediately</param>
        /// <param name="mandatory">If true, the broker will return an error 
        /// (as a connection exception) if the message cannot be delivered</param>
        /// <param name="priority">Default message priority</param>
        /// <returns>The new message publisher</returns>
        public IMessagePublisher CreatePublisher(string exchangeName, string routingKey, DeliveryMode deliveryMode,
                                                 long timeToLive, bool immediate, bool mandatory, int priority)
        {
            _logger.Debug(string.Format("Using new CreatePublisher exchangeName={0}, exchangeClass={1} routingKey={2}",
                                        exchangeName, "none", routingKey));
            
            return CreateProducerImpl(exchangeName, routingKey, deliveryMode,
                                      timeToLive, immediate, mandatory, priority);
        }

        /// <summary>
        /// Recover after transaction failure.
        /// </summary>
        /// <remarks>The 0-8 protocol does not support this, not implemented exception will always be thrown.</remarks>
        public void Recover()
        {
            CheckNotClosed();
            CheckNotTransacted();

            throw new NotImplementedException();
        }

        /// <summary>
        /// Commit the transaction.
        /// </summary>
        public void Commit()
        {
            // FIXME: Fail over safety. Needs FailoverSupport?
            CheckNotClosed();
            CheckTransacted(); // throws IllegalOperationException if not a transacted session

            try
            {
                // Acknowledge up to message last delivered (if any) for each consumer.
                // Need to send ack for messages delivered to consumers so far.
                foreach (BasicMessageConsumer consumer  in _consumers.Values)
                {
                    // Sends acknowledgement to server.
                    consumer.AcknowledgeDelivered();
                }

                // Commits outstanding messages sent and outstanding acknowledgements.
                _connection.ConvenientProtocolWriter.SyncWrite(TxCommitBody.CreateAMQFrame(_channelId), typeof(TxCommitOkBody));
            }
            catch (AMQException e)
            {
                throw new QpidException("Failed to commit", e);
            }
        }

        /// <summary>
        /// Rollback the transaction.
        /// </summary>
        public void Rollback()
        {
            lock (_suspensionLock)
            {
                CheckTransacted(); // throws IllegalOperationException if not a transacted session

                try
                {
                    bool suspended = IsSuspended;
                    if (!suspended)
                    {
                        Suspend(true);
                    }

                    // Reject up to message last delivered (if any) for each consumer.
                    // Need to send reject for messages delivered to consumers so far.
                    foreach (BasicMessageConsumer consumer  in _consumers.Values)
                    {
                        // Sends acknowledgement to server.
                        consumer.RejectUnacked();
                    }

                    _connection.ConvenientProtocolWriter.SyncWrite(TxRollbackBody.CreateAMQFrame(_channelId), typeof(TxRollbackOkBody));

                    if ( !suspended )
                    {
                        Suspend(false);
                    }
                } 
                catch (AMQException e)
                {
                    throw new QpidException("Failed to rollback", e);
                }
            }
        }

        /// <summary>
        /// Create a disconnected channel that will fault
        /// for most things, but is useful for testing
        /// </summary>
        /// <returns>A new disconnected channel</returns>
        public static IChannel CreateDisconnectedChannel()
        {
            return new AmqChannel();
        }

        public override void Close()
        {
            lock (_connection.FailoverMutex)
            {
                // We must close down all producers and consumers in an orderly fashion. This is the only method
                // that can be called from a different thread of control from the one controlling the session

                lock (_closingLock)
                {
                    SetClosed();

                    // we pass null since this is not an error case
                    CloseProducersAndConsumers(null);

                    try
                    {
                        _connection.CloseSession(this);
                    }
                    catch (AMQException e)
                    {
                        throw new QpidException("Error closing session: " + e);
                    }
                    finally
                    {
                        _connection.DeregisterSession(_channelId);
                    }
                }
            }
        }

        /**
         * Called when the server initiates the closure of the session
         * unilaterally.
         * @param e the exception that caused this session to be closed. Null causes the
         */
        public void ClosedWithException(Exception e)
        {
            lock (_connection.FailoverMutex)
            {
                // An AMQException has an error code and message already and will be passed in when closure occurs as a
                // result of a channel close request
                SetClosed();
                AMQException amqe;

                if (e is AMQException)
                {
                    amqe = (AMQException) e;
                }
                else
                {
                    amqe = new AMQException("Closing session forcibly", e);
                }

                _connection.DeregisterSession(_channelId);
                CloseProducersAndConsumers(amqe);
            }
        }

        public void MessageReceived(UnprocessedMessage message)
        {
            if (_logger.IsDebugEnabled)
            {
                _logger.Debug("Message received in session with channel id " + _channelId);
            }

            if ( message.DeliverBody == null )
            {
                ReturnBouncedMessage(message);
            } 
            else
            {
                _queue.Enqueue(message);
            }
        }

        public void Dispose()
        {
            Close();
        }

        private void SetClosed()
        {
            Interlocked.Exchange(ref _closed, CLOSED);
        }

        /// <summary>
        /// Close all producers or consumers. This is called either in the error case or when closing the session normally.
        /// <param name="amqe">the exception, may be null to indicate no error has occurred</param>
        ///
        private void CloseProducersAndConsumers(AMQException amqe)
        {
            try
            {
                CloseProducers();
            }
            catch (QpidException e)
            {
                _logger.Error("Error closing session: " + e, e);
            }
            try
            {
                CloseConsumers(amqe);
            }
            catch (QpidException e)
            {
                _logger.Error("Error closing session: " + e, e);
            }
        }

        /// <summary>
        /// Called to close message producers cleanly. This may or may <b>not</b> be as a result of an error. There is
        /// currently no way of propagating errors to message producers (this is a JMS limitation).
        /// </summary>
        private void CloseProducers()
        {
            _logger.Debug("Closing producers on session " + this);

            // we need to clone the list of producers since the close() method updates the _producers collection
            // which would result in a concurrent modification exception
            ArrayList clonedProducers = new ArrayList(_producers.Values);
            
            foreach (BasicMessageProducer prod in clonedProducers)
            {
                _logger.Debug("Closing producer " + prod);
                prod.Close();
            }

            // at this point the _producers map is empty
        }

        /// <summary>
        /// Called to close message consumers cleanly. This may or may <b>not</b> be as a result of an error.
        /// <param name="error">not null if this is a result of an error occurring at the connection level</param>
        private void CloseConsumers(Exception error)
        {
            if (_dispatcher != null)
            {
                _dispatcher.StopDispatcher();
            }

            // we need to clone the list of consumers since the close() method updates the _consumers collection
            // which would result in a concurrent modification exception
            ArrayList clonedConsumers = new ArrayList(_consumers.Values);

            foreach (BasicMessageConsumer con in clonedConsumers)
            {             
                if (error != null)
                {
                    con.NotifyError(error);
                }
                else
                {
                    con.Close();
                }
            }

            // at this point the _consumers map will be empty
        }

        private IMessagePublisher CreateProducerImpl(string exchangeName, string routingKey,
                                                    DeliveryMode deliveryMode,
                                                    long timeToLive, bool immediate, bool mandatory, int priority)
        {
            lock (_closingLock)
            {
                CheckNotClosed();

                try
                {
                    return new BasicMessageProducer(exchangeName, routingKey, _transacted, _channelId,
                                                    this, GetNextProducerId(),
                                                    deliveryMode, timeToLive, immediate, mandatory, priority);
                }
                catch (AMQException e)
                {
                    _logger.Error("Error creating message producer: " + e, e);
                    throw new QpidException("Error creating message producer", e);
                }
            }
        }

        /// <summary> Creates a message consumer on this channel.</summary>
        ///
        /// <param name="queueName">The name of the queue to attach the consumer to.</param>
        /// <param name="prefetchLow">The pre-fetch buffer low-water mark.</param>
        /// <param name="prefetchHigh">The pre-fetch buffer high-water mark.</param>
        /// <param name="noLocal">The no-local flag, <tt>true</tt> means that the consumer does not receive messages sent on this channel.</param>
        /// <param name="exclusive">The exclusive flag, <tt>true</tt> gives this consumer exclusive receive access to the queue.</param>
        ///
        /// <return>The message consumer.</return>
        private IMessageConsumer CreateConsumerImpl(string queueName,
                                                    int prefetchLow,
                                                    int prefetchHigh,
                                                    bool noLocal,
                                                    bool exclusive,
                                                    bool browse)
        {
            lock (_closingLock)
            {
                CheckNotClosed();
               
                BasicMessageConsumer consumer = new BasicMessageConsumer(_channelId, queueName, noLocal,
                                                                         _messageFactoryRegistry, this,
                                                                         prefetchHigh, prefetchLow, exclusive,
                                                                         browse);
                try
                {
                    RegisterConsumer(consumer);
                }
                catch (AMQException e)
                {
                    throw new QpidException("Error registering consumer: " + e, e);
                }

                return consumer;
            }
        }

        private void CheckTransacted()
        {
            if (!Transacted)
            {
                throw new InvalidOperationException("Channel is not transacted");
            }
        }

        private void CheckNotTransacted()
        {
            if (Transacted)
            {
                throw new InvalidOperationException("Channel is transacted");
            }
        }
     
        internal void Start()
        {
            _dispatcher = new Dispatcher(this);
            Thread dispatcherThread = new Thread(new ThreadStart(_dispatcher.RunDispatcher));
            dispatcherThread.IsBackground = true;
            dispatcherThread.Start();
        }

        internal void Stop()
        {
            Suspend(true);
            if (_dispatcher != null)
            {
                _dispatcher.StopDispatcher();
            }
        }

        internal void RegisterConsumer(string consumerTag, IMessageConsumer consumer)
        {
            _consumers[consumerTag] =  consumer;
        }

        /// <summary>
        /// Called by the MessageConsumer when closing, to deregister the consumer from the
        /// map from consumerTag to consumer instance.
        /// </summary>
        /// <param name="consumerTag">the consumer tag, that was broker-generated</param>        
        internal void DeregisterConsumer(string consumerTag)
        {
            _consumers.Remove(consumerTag);
        }

        internal void RegisterProducer(long producerId, IMessagePublisher publisher)
        {
            _producers[producerId] = publisher;
        }

        internal void DeregisterProducer(long producerId)
        {
            _producers.Remove(producerId);
        }

        private long GetNextProducerId()
        {
            return ++_nextProducerId;
        }
        
        /**
         * Called to mark the session as being closed. Useful when the session needs to be made invalid, e.g. after
         * failover when the client has veoted resubscription.
         *
         * The caller of this method must already hold the failover mutex.
         */
        internal void MarkClosed()
        {
            SetClosed();
            _connection.DeregisterSession(_channelId);
            MarkClosedProducersAndConsumers();
        }

        private void MarkClosedProducersAndConsumers()
        {
            try
            {
                // no need for a markClosed* method in this case since there is no protocol traffic closing a producer
                CloseProducers();
            }
            catch (QpidException e)
            {
                _logger.Error("Error closing session: " + e, e);
            }
            try
            {
                MarkClosedConsumers();
            }
            catch (QpidException e)
            {
                _logger.Error("Error closing session: " + e, e);
            }
        }

        private void MarkClosedConsumers()
        {
            if (_dispatcher != null)
            {
                _dispatcher.StopDispatcher();
            }
            // we need to clone the list of consumers since the close() method updates the _consumers collection
            // which would result in a concurrent modification exception
            ArrayList clonedConsumers = new ArrayList(_consumers.Values);

            foreach (BasicMessageConsumer consumer in clonedConsumers)
            {
                consumer.MarkClosed();
            }
            // at this point the _consumers map will be empty
        }

        private void DoPurgeQueue(string queueName, bool noWait)
        {
            try
            {
                _logger.DebugFormat("PurgeQueue {0}", queueName);

                AMQFrame purgeQueue = QueuePurgeBody.CreateAMQFrame(_channelId, 0, queueName, noWait);

                if (noWait)
                    _connection.ProtocolWriter.Write(purgeQueue);
                else
                    _connection.ConvenientProtocolWriter.SyncWrite(purgeQueue, typeof(QueuePurgeOkBody));
            }
            catch (AMQException)
            {
                throw;
            }
        }

        /**
         * Replays frame on fail over.
         * 
         * @throws AMQException
         */
        internal void ReplayOnFailOver()
        {
            _logger.Debug(string.Format("Replaying frames for channel {0}", _channelId));
            foreach (AMQFrame frame in _replayFrames)
            {
                _logger.Debug(string.Format("Replaying frame=[{0}]", frame));
                _connection.ProtocolWriter.Write(frame);
            }
        }

        /// <summary>
        /// Callers must hold the failover mutex before calling this method.
        /// </summary>
        /// <param name="consumer"></param>
        private void RegisterConsumer(BasicMessageConsumer consumer)
        {
            // Need to generate a consumer tag on the client so we can exploit the nowait flag.
            String tag = string.Format("{0}-{1}", _sessionNumber, _nextConsumerNumber++);
            consumer.ConsumerTag = tag;
            _consumers.Add(tag, consumer);
            
            String consumerTag = ConsumeFromQueue(consumer.QueueName, consumer.NoLocal,
                                                  consumer.Exclusive, consumer.AcknowledgeMode, tag, consumer.Browse);
            
        }

        internal void DoBind(string queueName, string exchangeName, string routingKey, FieldTable args)
        {

            _logger.Debug(string.Format("QueueBind queueName={0} exchangeName={1} routingKey={2}, arg={3}",
                                        queueName, exchangeName, routingKey, args));

            AMQFrame queueBind = QueueBindBody.CreateAMQFrame(_channelId, 0,
                                                              queueName, exchangeName,
                                                              routingKey, false, args);
            

            lock (_connection.FailoverMutex)
            {
                _connection.ConvenientProtocolWriter.SyncWrite(queueBind, typeof(QueueBindOkBody));
            }
            // AS FIXME: wasnae me
            _replayFrames.Add(QueueBindBody.CreateAMQFrame(_channelId, 0,
                                                           queueName, exchangeName,
                                                           routingKey, true, args));
        }

        private String ConsumeFromQueue(String queueName, bool noLocal, bool exclusive, AcknowledgeMode acknowledgeMode, String tag, bool browse)
        {
            FieldTable args = new FieldTable();   
            if(browse)
            {
               args["x-filter-no-consume"] = true;
            }
            AMQFrame basicConsume = BasicConsumeBody.CreateAMQFrame(_channelId, 0,
                                                                    queueName, tag, noLocal,
                                                                    acknowledgeMode == AcknowledgeMode.NoAcknowledge,
                                                                    exclusive, true, args);

            _replayFrames.Add(basicConsume);

            _connection.ProtocolWriter.Write(basicConsume);
            return tag;
        }

        private void DoDeleteQueue(string queueName, bool ifUnused, bool ifEmpty, bool noWait)
        {
            try
            {
                _logger.Debug(string.Format("DeleteQueue name={0}", queueName));
                
                AMQFrame queueDelete = QueueDeleteBody.CreateAMQFrame(_channelId, 0, queueName, ifUnused, ifEmpty, noWait);
                
                if (noWait)
                {
                    _connection.ProtocolWriter.Write(queueDelete);
                }
                else
                {
                    _connection.ConvenientProtocolWriter.SyncWrite(queueDelete, typeof(QueueDeleteOkBody));
                }
                // AS FIXME: wasnae me
                _replayFrames.Add(QueueDeleteBody.CreateAMQFrame(_channelId, 0, queueName, ifUnused, ifEmpty, true));
            }
            catch (AMQException)
            {
                throw;
            }
        }

        private void DoQueueDeclare(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete, IFieldTable args)
        {
            _logger.Debug(string.Format("DeclareQueue name={0} durable={1} exclusive={2}, auto-delete={3}",
                                        queueName, isDurable, isExclusive, isAutoDelete));

            AMQFrame queueDeclare = QueueDeclareBody.CreateAMQFrame(_channelId, 0, queueName, false, isDurable, isExclusive,
                                                                    isAutoDelete, false, (FieldTable) args);


            lock (_connection.FailoverMutex)
            {
                _connection.ConvenientProtocolWriter.SyncWrite(queueDeclare, typeof(QueueDeclareOkBody));
            }
            // AS FIXME: wasnae me
            _replayFrames.Add(QueueDeclareBody.CreateAMQFrame(_channelId, 0, queueName, false, isDurable, isExclusive,
                                                                    isAutoDelete, true, null));
        }

        // AMQP-level method.
        private void DeclareExchange(ushort channelId, ushort ticket, string exchangeName, 
                                     string exchangeClass, bool passive, bool durable, 
                                     bool autoDelete, bool xinternal, bool noWait, FieldTable args)
        {
            _logger.Debug(String.Format("DeclareExchange channelId={0} exchangeName={1} exchangeClass={2}",
                                        _channelId, exchangeName, exchangeClass));

            AMQFrame declareExchange = ExchangeDeclareBody.CreateAMQFrame(channelId, ticket, exchangeName, exchangeClass, passive, 
                                                                          durable, autoDelete, xinternal, noWait, args);
            
            if (noWait)
            {
                lock (_connection.FailoverMutex)
                {
                    _connection.ProtocolWriter.Write(declareExchange);
                }
                // AS FIXME: wasnae me
            	_replayFrames.Add(declareExchange);
            }
            else
            {
                throw new NotImplementedException("Don't use nowait=false with DeclareExchange");
                //_connection.ConvenientProtocolWriter.SyncWrite(declareExchange, typeof (ExchangeDeclareOkBody));
            }
        }

        /**
         * Acknowledge a message or several messages. This method can be called via AbstractJMSMessage or from
         * a BasicConsumer. The former where the mode is CLIENT_ACK and the latter where the mode is
         * AUTO_ACK or similar.
         *
         * @param deliveryTag the tag of the last message to be acknowledged
         * @param multiple    if true will acknowledge all messages up to and including the one specified by the
         *                    delivery tag
         */
        internal void AcknowledgeMessage(ulong deliveryTag, bool multiple)
        {
            AMQFrame ackFrame = BasicAckBody.CreateAMQFrame(_channelId, deliveryTag, multiple);
            if (_logger.IsDebugEnabled)
            {
                _logger.Debug("Sending ack for delivery tag " + deliveryTag + " on channel " + _channelId);
            }
            // FIXME: lock FailoverMutex here?
            _connection.ProtocolWriter.Write(ackFrame);
        }

        public void RejectMessage(ulong deliveryTag, bool requeue)
        {
            if ((_acknowledgeMode == AcknowledgeMode.ClientAcknowledge) || (_acknowledgeMode == AcknowledgeMode.SessionTransacted))
            {
                AMQFrame rejectFrame = BasicRejectBody.CreateAMQFrame(_channelId, deliveryTag, requeue);
                _connection.ProtocolWriter.Write(rejectFrame);
            }
        }
        
        /// <summary>
        /// Handle a message that bounced from the server, creating
        /// the corresponding exception and notifying the connection about it
        /// </summary>
        /// <param name="message">Unprocessed message</param>
        private void ReturnBouncedMessage(UnprocessedMessage message)
        {
            try
            {
                AbstractQmsMessage bouncedMessage =
                    _messageFactoryRegistry.CreateMessage(0, false, message.ContentHeader, message.Bodies);

                int errorCode = message.BounceBody.ReplyCode;
                string reason = message.BounceBody.ReplyText;
                _logger.Debug("Message returned with error code " + errorCode + " (" + reason + ")");
                AMQException exception;

                if (errorCode == AMQConstant.NO_CONSUMERS.Code)
                {
                    exception = new AMQNoConsumersException(reason, bouncedMessage);
                } 
                else if (errorCode == AMQConstant.NO_ROUTE.Code)
                {
                    exception = new AMQNoRouteException(reason, bouncedMessage);
                }
                else
                {
                    exception = new AMQUndeliveredException(errorCode, reason, bouncedMessage);
                }
                
                _connection.ExceptionReceived(exception);
            } 
            catch (Exception ex)
            {
                _logger.Error("Caught exception trying to raise undelivered message exception (dump follows) - ignoring...", ex);
            }
        }

        private void OnPrefetchLowMark(int count)
        {
            if (_acknowledgeMode == AcknowledgeMode.NoAcknowledge)
            {
                _logger.Warn("Below threshold(" + _defaultPrefetchLowMark + ") so unsuspending channel. Current value is " + count);
                Suspend(false);
            }
        }

        private void OnPrefetchHighMark(int count)
        {
            if (_acknowledgeMode == AcknowledgeMode.NoAcknowledge)
            {
                _logger.Warn("Above threshold(" + _defaultPrefetchHighMark + ") so suspending channel. Current value is " + count);
                Suspend(true);
            }
        }

        private void Suspend(bool suspend)
        {
            lock (_suspensionLock)
            {
                if (_logger.IsDebugEnabled)
                {
                    _logger.Debug("Setting channel flow : " + (suspend ? "suspended" : "unsuspended"));
                }

                _suspended = suspend;
                AMQFrame frame = ChannelFlowBody.CreateAMQFrame(_channelId, !suspend);

                Connection.ConvenientProtocolWriter.SyncWrite(frame, typeof(ChannelFlowOkBody));
            }
        }

        /// <summary>A Dispatcher turns the consumption of incoming messages from an arrival queue, into event notifications on consumers.
        /// The arrival queue is typically a blocking queue, on which a dispatcher waits for messages to consume. Upon receipt of a message
        /// the dispatcher finds the consumer that is listening to the queue to which the message has been send and notifies it of the new
        /// message.
        ///
        /// <p/>The Dispatcher also contains logic to recognize bounced messages. Bounced messages returned from the broker can be
        /// told apart from regular deliveries because they do not have a delivery queue set on them. When the dispatcher receives a
        /// bounced message it creates an exception and notifies the connection, to which its containing channel belongs, of the condition.
        ///
        /// <p/><table id="crc"><caption>CRC Card</caption>
        /// <tr><th> Responsibilities <th> Collaborations
        /// <tr><td> Notify consumers of message arrivals on their queues. <td> <see cref="BasicMessageConsumer"/>
        /// <tr><td> Notify the containing connection of bounced message arrivals. <td> <see cref="AMQConnection"/>
        /// </table>
        /// </summary>
        ///
        /// <remarks>Stop mechanism seems wrong, as queue consume is evaluated after stop flag, so could consume and notify one more message.
        /// Placing stop check after consume may also be wrong as it may cause a message to be thrown away. Seems more correct to use interupt on 
        /// the block thread to cause it to prematurely return from its wait, whereupon it can be made to re-check the stop flag.</remarks>
        ///
        /// <remarks>Exception swallowed, if there is an exception whilst notifying the connection on bounced messages. Unhandled excetpion should
        /// fall through and terminate the loop, as it is a bug if it occurrs.</remarks>
        private class Dispatcher
        {            
            /// <summary> Flag used to indicate when this dispatcher is to be stopped (0=go, 1=stop). </summary>
            private int _stopped = 0;

            /// <summary> The channel for which this is a dispatcher. </summary>
            private AmqChannel _containingChannel;
            
            /// <summary> Creates a dispatcher on the specified channel. </summary>
            ///
            /// <param name="containingChannel"> The channel on which this is a dispatcher. </param>
            public Dispatcher(AmqChannel containingChannel)
            {
                _containingChannel = containingChannel;
            }
            
            /// <summary>The message dispatch loop. Consumes messages from the channels queue, notifying consumers of regular deliveries, and
            /// the connection of bounced messages.</summary>
            public void RunDispatcher()
            {
                UnprocessedMessage message;

                while (_stopped == 0 && (message = (UnprocessedMessage)_containingChannel._queue.Dequeue()) != null)
                {
                    if (message.DeliverBody != null)
                    {
                        BasicMessageConsumer consumer = (BasicMessageConsumer) _containingChannel._consumers[message.DeliverBody.ConsumerTag];
                        
                        if (consumer == null)
                        {
                            _logger.Warn("Received a message from queue " + message.DeliverBody.ConsumerTag + " without a f - ignoring...");
                        }
                        else
                        {
                            consumer.NotifyMessage(message, _containingChannel.ChannelId);
                        }
                    }
                    else
                    {
                        try
                        {
                            // Bounced message is processed here, away from the transport thread
                            AbstractQmsMessage bouncedMessage = _containingChannel._messageFactoryRegistry.
                                CreateMessage(0, false, message.ContentHeader, message.Bodies);
                            
                            int errorCode = message.BounceBody.ReplyCode;
                            string reason = message.BounceBody.ReplyText;

                            _logger.Debug("Message returned with error code " + errorCode + " (" + reason + ")");
                            
                            _containingChannel._connection.ExceptionReceived(new AMQUndeliveredException(errorCode, "Error: " + reason, bouncedMessage));
                        }
                        catch (Exception e)
                        {
                            _logger.Error("Caught exception trying to raise undelivered message exception (dump follows) - ignoring...", e);
                        }
                    }
                }                

                _logger.Debug("Dispatcher thread terminating for channel: " + _containingChannel._channelId + ".");
            }

            /// <summary> Sets a stop flag on this dispatcher, which causes its dispatch loop to exit at the next available opportunity. </summary>
            public void StopDispatcher()
            {
                Interlocked.Exchange(ref _stopped, 1);                
            }
        }
    }
}