diff options
Diffstat (limited to 'trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs')
| -rw-r--r-- | trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs | 1198 |
1 files changed, 0 insertions, 1198 deletions
diff --git a/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs b/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs deleted file mode 100644 index 86dc9a4681..0000000000 --- a/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs +++ /dev/null @@ -1,1198 +0,0 @@ -/* - * - * 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); - } - - /// <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); - } - - /// <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) - { - 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; - } - } - - 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); - - } - - 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) - { - - 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; - } - - 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) - { - _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, null); - - - 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); - } - } - } -} |