/*
 *
 * 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
{
    public class AmqChannel : Closeable, IChannel
    {
        private static readonly ILog _logger = LogManager.GetLogger(typeof(AmqChannel));

        internal const int BASIC_CONTENT_TYPE = 60;

        private static int _nextSessionNumber = 0;

        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;

        public const int DEFAULT_PREFETCH_HIGH_MARK = 5000;
        public const int DEFAULT_PREFETCH_LOW_MARK = 2500;

        private AMQConnection _connection;

        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>
        /// Set of all producers created by this session
        /// </summary>
        private Hashtable _producers = Hashtable.Synchronized(new Hashtable());

        /// <summary>
        /// Maps from consumer tag to JMSMessageConsumer instance
        /// </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>
        /// Responsible for decoding a message fragment and passing it to the appropriate message consumer.
        /// </summary>
        private class Dispatcher
        {
            private int _stopped = 0;

            private AmqChannel _containingChannel;
            
            public Dispatcher(AmqChannel containingChannel)
            {
                _containingChannel = containingChannel;
            }
            
            /// <summary>
            /// Runs the dispatcher. This is intended to be Run in a separate thread.
            /// </summary>
            public void RunDispatcher()
            {
                UnprocessedMessage message;

                while (_stopped == 0 && (message = (UnprocessedMessage)_containingChannel._queue.Dequeue()) != null)
                {
                    //_queue.size()
                    DispatchMessage(message);
                }                

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

            private void DispatchMessage(UnprocessedMessage message)
            {
                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 mina 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);
                    }
                }
            }

            public void StopDispatcher()
            {
                Interlocked.Exchange(ref _stopped, 1);                
            }
        }

        /// <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>
        /// 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 IBytesMessage CreateBytesMessage()
        {
            return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
        }

        public IMessage CreateMessage()
        {
            // TODO: this is supposed to create a message consisting only of message headers
            return (IBytesMessage)_messageFactoryRegistry.CreateMessage("application/octet-stream");
        }
        
        public IMessage CreateMessage(string mimeType)
        {
           return _messageFactoryRegistry.CreateMessage(mimeType);
        }

        public ITextMessage CreateTextMessage()
        {
            return CreateTextMessage(String.Empty);
        }

        public ITextMessage CreateTextMessage(string text)
        {
            ITextMessage msg = (ITextMessage)_messageFactoryRegistry.CreateMessage("text/plain");
            msg.Text = text;
            return msg;
        }

        public bool Transacted
        {
            get
            {
                CheckNotClosed();
                return _transacted;
            }
        }

        public AcknowledgeMode AcknowledgeMode
        {
            get
            {
                CheckNotClosed();
                return _acknowledgeMode;
            }
        }

        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.AcknowledgeLastDelivered();
                }

                // 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);
            }
        }

        public void Rollback()
        {
           lock ( _suspensionLock )
           {
              CheckTransacted(); // throws IllegalOperationException if not a transacted session

              try
              {
                 bool suspended = IsSuspended;
                 if ( !suspended )
                    Suspend(true);
                 
                 // todo: rollback dispatcher when TX support is added
                 //if ( _dispatcher != null )
                 //   _dispatcher.Rollback();

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

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

        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);
                    }
                }
            }
        }

        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);
            }
        }

        /**
         * 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);
            }
        }

        /// <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
        }

        public void Recover()
        {
            CheckNotClosed();
            CheckNotTransacted(); // throws IllegalOperationException if not a transacted session

            // TODO: This cannot be implemented using 0.8 semantics
            throw new NotImplementedException();
        }                

        public void Run()
        {
            throw new NotImplementedException();
        }

        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);
        }

        public 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);
                }
            }
        }

        public IMessageConsumer CreateConsumer(string queueName,
                                               int prefetchLow,
                                               int prefetchHigh,
                                               bool noLocal,
                                               bool exclusive,
                                               bool durable,
                                               string subscriptionName)
        {
            _logger.Debug(String.Format("CreateConsumer queueName={0} prefetchLow={1} prefetchHigh={2} noLocal={3} exclusive={4} durable={5} subscriptionName={6}",
                                  queueName, prefetchLow, prefetchHigh, noLocal, exclusive, durable, subscriptionName));
            return CreateConsumerImpl(queueName, prefetchLow, prefetchHigh, noLocal, exclusive, durable, subscriptionName);
        }

        private IMessageConsumer CreateConsumerImpl(string queueName,
                                                    int prefetchLow,
                                                    int prefetchHigh,
                                                    bool noLocal,
                                                    bool exclusive,
                                                    bool durable,
                                                    string subscriptionName)
        {
            if (durable || subscriptionName != null)
            {
                throw new NotImplementedException(); // TODO: durable subscriptions.
            }

            lock (_closingLock)
            {
                CheckNotClosed();
               
                BasicMessageConsumer consumer = new BasicMessageConsumer(_channelId, queueName, noLocal,
                                                                         _messageFactoryRegistry, this,
                                                                         prefetchHigh, prefetchLow, exclusive);
                try
                {
                    RegisterConsumer(consumer);
                }
                catch (AMQException e)
                {
                    throw new QpidException("Error registering consumer: " + e, e);
                }

                return consumer;
            }
        }

        public void Unsubscribe(String name)
        {
            throw new NotImplementedException(); // FIXME
        }

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

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

        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 int DefaultPrefetch
        {
           get { return DefaultPrefetchHigh; }
        }
        public int DefaultPrefetchLow
        {
            get { return _defaultPrefetchLowMark; }
        }
        public int DefaultPrefetchHigh
        {
           get { return _defaultPrefetchHighMark; }
        }
        public bool IsSuspended
        {
           get { return _suspended; }
        }

        public ushort ChannelId
        {
            get
            {
                return _channelId;
            }
        }

        public AMQConnection Connection
        {
            get
            {
                return _connection;
            }
        }
        
        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;
        }
        
        public void Dispose()
        {
            Close();
        }

        /**
         * 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
        }

        public void PurgeQueue(string queueName, bool noWait)
        {
            DoPurgeQueue(queueName, noWait);
        }

        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>
        void RegisterConsumer(BasicMessageConsumer consumer)
        {
            String consumerTag = ConsumeFromQueue(consumer.QueueName, consumer.NoLocal,
                                           consumer.Exclusive, consumer.AcknowledgeMode);
            consumer.ConsumerTag = consumerTag;
            _consumers.Add(consumerTag, consumer);
        }

        public void Bind(string queueName, string exchangeName, string routingKey, IFieldTable args)
        {
            DoBind(queueName, exchangeName, routingKey, (FieldTable)args);            
        }

        public void Bind(string queueName, string exchangeName, string routingKey)
        {
            DoBind(queueName, exchangeName, routingKey, new FieldTable());
        }

        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, true, args);
            _replayFrames.Add(queueBind);

            lock (_connection.FailoverMutex)
            {
                _connection.ProtocolWriter.Write(queueBind);
            }
        }

        private String ConsumeFromQueue(String queueName, bool noLocal, bool exclusive, AcknowledgeMode acknowledgeMode)
        {
            // Need to generate a consumer tag on the client so we can exploit the nowait flag.
            String tag = string.Format("{0}-{1}", _sessionNumber, _nextConsumerNumber++);
            
            AMQFrame basicConsume = BasicConsumeBody.CreateAMQFrame(_channelId, 0,
                                                                  queueName, tag, noLocal,
                                                                  acknowledgeMode == AcknowledgeMode.NoAcknowledge,
                                                                  exclusive, true, new FieldTable());

            _replayFrames.Add(basicConsume);

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

        public void DeleteExchange(string exchangeName)
        {
            throw new NotImplementedException(); // FIXME
        }

        public void DeleteQueue(string queueName, bool ifUnused, bool ifEmpty, bool noWait)
        {
            DoDeleteQueue(queueName, ifUnused, ifEmpty, noWait);
        }

        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, // queueName
                    ifUnused, // IfUnUsed
                    ifEmpty, // IfEmpty
                    noWait); // NoWait

                _replayFrames.Add(queueDelete);

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

        public MessageConsumerBuilder CreateConsumerBuilder(string queueName)
        {
            return new MessageConsumerBuilder(this, queueName);
        }

        public MessagePublisherBuilder CreatePublisherBuilder()
        {
            return new MessagePublisherBuilder(this);
        }

        public string GenerateUniqueName()
        {
            string result = _connection.ProtocolSession.GenerateQueueName();            
            return Regex.Replace(result, "[^a-z0-9_]", "_");
        }

        public void DeclareQueue(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete)
        {
            DoQueueDeclare(queueName, isDurable, isExclusive, isAutoDelete);
        }

        private void DoQueueDeclare(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete)
        {
            _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, true, null);

            _replayFrames.Add(queueDeclare);

            lock (_connection.FailoverMutex)
            {
                _connection.ProtocolWriter.Write(queueDeclare);
            }
        }

        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);
        }

        // 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);

            _replayFrames.Add(declareExchange);
            
            if (noWait)
            {
                lock (_connection.FailoverMutex)
                {
                    _connection.ProtocolWriter.Write(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);
        }

       /// <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));
          }
       }

    }
}