1+ using System ;
2+ using System . Threading . Tasks ;
3+ using RabbitMQ . Client . Events ;
4+ using TaskExtensions = RabbitMQ . Client . Impl . TaskExtensions ;
5+
6+ namespace RabbitMQ . Client
7+ {
8+ public class AsyncDefaultBasicConsumer : IAsyncBasicConsumer
9+ {
10+ public readonly object m_eventLock = new object ( ) ;
11+ public AsyncEventHandler < ConsumerEventArgs > m_consumerCancelled ;
12+
13+ /// <summary>
14+ /// Creates a new instance of an <see cref="DefaultBasicConsumer"/>.
15+ /// </summary>
16+ public AsyncDefaultBasicConsumer ( )
17+ {
18+ ShutdownReason = null ;
19+ Model = null ;
20+ IsRunning = false ;
21+ ConsumerTag = null ;
22+ }
23+
24+ /// <summary>
25+ /// Constructor which sets the Model property to the given value.
26+ /// </summary>
27+ /// <param name="model">Common AMQP model.</param>
28+ public AsyncDefaultBasicConsumer ( IModel model )
29+ {
30+ ShutdownReason = null ;
31+ IsRunning = false ;
32+ ConsumerTag = null ;
33+ Model = model ;
34+ }
35+
36+ /// <summary>
37+ /// Retrieve the consumer tag this consumer is registered as; to be used when discussing this consumer
38+ /// with the server, for instance with <see cref="IModel.BasicCancel"/>.
39+ /// </summary>
40+ public string ConsumerTag { get ; set ; }
41+
42+ /// <summary>
43+ /// Returns true while the consumer is registered and expecting deliveries from the broker.
44+ /// </summary>
45+ public bool IsRunning { get ; protected set ; }
46+
47+ /// <summary>
48+ /// If our <see cref="IModel"/> shuts down, this property will contain a description of the reason for the
49+ /// shutdown. Otherwise it will contain null. See <see cref="ShutdownEventArgs"/>.
50+ /// </summary>
51+ public ShutdownEventArgs ShutdownReason { get ; protected set ; }
52+
53+ /// <summary>
54+ /// Signalled when the consumer gets cancelled.
55+ /// </summary>
56+ public event AsyncEventHandler < ConsumerEventArgs > ConsumerCancelled
57+ {
58+ add
59+ {
60+ lock ( m_eventLock )
61+ {
62+ m_consumerCancelled += value ;
63+ }
64+ }
65+ remove
66+ {
67+ lock ( m_eventLock )
68+ {
69+ m_consumerCancelled -= value ;
70+ }
71+ }
72+ }
73+
74+ /// <summary>
75+ /// Retrieve the <see cref="IModel"/> this consumer is associated with,
76+ /// for use in acknowledging received messages, for instance.
77+ /// </summary>
78+ public IModel Model { get ; set ; }
79+
80+ /// <summary>
81+ /// Called when the consumer is cancelled for reasons other than by a basicCancel:
82+ /// e.g. the queue has been deleted (either by this channel or by any other channel).
83+ /// See <see cref="HandleBasicCancelOk"/> for notification of consumer cancellation due to basicCancel
84+ /// </summary>
85+ /// <param name="consumerTag">Consumer tag this consumer is registered.</param>
86+ public virtual async Task HandleBasicCancel ( string consumerTag )
87+ {
88+ await OnCancel ( ) . ConfigureAwait ( false ) ;
89+ }
90+
91+ /// <summary>
92+ /// Called upon successful deregistration of the consumer from the broker.
93+ /// </summary>
94+ /// <param name="consumerTag">Consumer tag this consumer is registered.</param>
95+ public virtual async Task HandleBasicCancelOk ( string consumerTag )
96+ {
97+ await OnCancel ( ) . ConfigureAwait ( false ) ;
98+ }
99+
100+ /// <summary>
101+ /// Called upon successful registration of the consumer with the broker.
102+ /// </summary>
103+ /// <param name="consumerTag">Consumer tag this consumer is registered.</param>
104+ public virtual Task HandleBasicConsumeOk ( string consumerTag )
105+ {
106+ ConsumerTag = consumerTag ;
107+ IsRunning = true ;
108+ return TaskExtensions . CompletedTask ;
109+ }
110+
111+ /// <summary>
112+ /// Called each time a message arrives for this consumer.
113+ /// </summary>
114+ /// <remarks>
115+ /// Does nothing with the passed in information.
116+ /// Note that in particular, some delivered messages may require acknowledgement via <see cref="IModel.BasicAck"/>.
117+ /// The implementation of this method in this class does NOT acknowledge such messages.
118+ /// </remarks>
119+ public virtual Task HandleBasicDeliver ( string consumerTag ,
120+ ulong deliveryTag ,
121+ bool redelivered ,
122+ string exchange ,
123+ string routingKey ,
124+ IBasicProperties properties ,
125+ byte [ ] body )
126+ {
127+ // Nothing to do here.
128+ return TaskExtensions . CompletedTask ;
129+ }
130+
131+ /// <summary>
132+ /// Called when the model shuts down.
133+ /// </summary>
134+ /// <param name="model"> Common AMQP model.</param>
135+ /// <param name="reason"> Information about the reason why a particular model, session, or connection was destroyed.</param>
136+ public virtual async Task HandleModelShutdown ( object model , ShutdownEventArgs reason )
137+ {
138+ ShutdownReason = reason ;
139+ await OnCancel ( ) . ConfigureAwait ( false ) ;
140+ }
141+
142+ /// <summary>
143+ /// Default implementation - overridable in subclasses.</summary>
144+ /// <remarks>
145+ /// This default implementation simply sets the <see cref="IsRunning"/>
146+ /// property to false, and takes no further action.
147+ /// </remarks>
148+ public virtual async Task OnCancel ( )
149+ {
150+ IsRunning = false ;
151+ AsyncEventHandler < ConsumerEventArgs > handler ;
152+ lock ( m_eventLock )
153+ {
154+ handler = m_consumerCancelled ;
155+ }
156+ if ( handler != null )
157+ {
158+ foreach ( AsyncEventHandler < ConsumerEventArgs > h in handler . GetInvocationList ( ) )
159+ {
160+ await h ( this , new ConsumerEventArgs ( ConsumerTag ) ) . ConfigureAwait ( false ) ;
161+ }
162+ }
163+ }
164+ }
165+ }
0 commit comments