Created ConsumerCancelled event on ConsumerCancellation class to be fired when a consumer has been canced by the broker

Author: rtaylor01

Source/EasyNetQ/Consumer/ConsumerCancellation.cs

@@ -1,9 +1,16 @@
-using System;
+
+using System;
 
 namespace EasyNetQ.Consumer
 {
     public class ConsumerCancellation : IDisposable
     {
+       
+        public delegate void ConsumerCancel(object queue);
+        // This event will fire whenever this class is disposed. It is necessary in order to catch
+        // when the consumer is canceled by the broker as well as the user. 
+        public event ConsumerCancel ConsumerCancelled;
+
         private readonly Action onCancellation;
 
         public ConsumerCancellation(Action onCancellation)
@@ -17,5 +24,10 @@ public void Dispose()
         {
             onCancellation();
         }
+
+        public void OnCancel(object queue)
+        {
+            ConsumerCancelled?.Invoke(queue);
+        }
     }
 }

Source/EasyNetQ/Consumer/ExclusiveConsumer.cs

@@ -27,6 +27,8 @@ public class ExclusiveConsumer : IConsumer
 
         private readonly IList<IDisposable> disposables = new List<IDisposable>();
 
+        private ConsumerCancellation consumerCancellation;
+
         public ExclusiveConsumer(
             IQueue queue,
             Func<byte[], MessageProperties, MessageReceivedInfo, Task> onMessage,
@@ -58,8 +60,9 @@ public IDisposable StartConsuming()
             disposables.Add(Timers.Start(s => StartConsumingInternal(), RestartConsumingPeriod, RestartConsumingPeriod));
 
             StartConsumingInternal();
-            
-            return new ConsumerCancellation(Dispose);   
+
+            consumerCancellation = new ConsumerCancellation(Dispose);
+            return consumerCancellation;
         }
 
         private void StartConsumingInternal()
@@ -115,7 +118,9 @@ public void Dispose()
                 return;
 
             disposed = true;
-            
+
+            consumerCancellation.OnCancel(queue);
+
             eventBus.Publish(new StoppedConsumingEvent(this));
 
             foreach (var disposal in disposables)

Source/EasyNetQ/Consumer/PersistentConsumer.cs

@@ -22,6 +22,8 @@ public class PersistentConsumer : IConsumer
 
         private readonly IList<IDisposable> subscriptions = new List<IDisposable>();
 
+        private ConsumerCancellation consumerCancellation;
+
         public PersistentConsumer(
             IQueue queue, 
             Func<byte[], MessageProperties, MessageReceivedInfo, Task> onMessage, 
@@ -52,7 +54,8 @@ public IDisposable StartConsuming()
 
             StartConsumingInternal();
 
-            return new ConsumerCancellation(Dispose);
+            consumerCancellation = new ConsumerCancellation(Dispose);
+            return consumerCancellation;
         }
 
         private void StartConsumingInternal()
@@ -101,6 +104,8 @@ public void Dispose()
 
             disposed = true;
 
+            consumerCancellation.OnCancel(queue);
+
             eventBus.Publish(new StoppedConsumingEvent(this));
 
             foreach (var subscription in subscriptions)

Source/EasyNetQ/Consumer/PersistentMultipleConsumer.cs

@@ -23,6 +23,8 @@ public class PersistentMultipleConsumer : IConsumer
 
         private readonly IList<IDisposable> subscriptions = new List<IDisposable>();
 
+        private ConsumerCancellation consumerCancellation;
+
         public PersistentMultipleConsumer(
             ICollection<Tuple<IQueue, Func<byte[], MessageProperties, MessageReceivedInfo, Task>>> queueConsumerPairs,
             IPersistentConnection connection,
@@ -50,7 +52,8 @@ public IDisposable StartConsuming()
 
             StartConsumingInternal();
 
-            return new ConsumerCancellation(Dispose);
+            consumerCancellation = new ConsumerCancellation(Dispose);
+            return consumerCancellation;
         }
 
         private void StartConsumingInternal()
@@ -92,7 +95,9 @@ public void Dispose()
             if (disposed) return;
 
             disposed = true;
-            
+
+            consumerCancellation.OnCancel(queueConsumerPairs);
+
             eventBus.Publish(new StoppedConsumingEvent(this));
 
             foreach (var subscription in subscriptions)

Source/EasyNetQ/Consumer/TransientConsumer.cs

@@ -16,6 +16,8 @@ public class TransientConsumer : IConsumer
 
         private IInternalConsumer internalConsumer;
 
+        private ConsumerCancellation consumerCancellation;
+
         public TransientConsumer(
             IQueue queue, 
             Func<byte[], MessageProperties, MessageReceivedInfo, Task> onMessage, 
@@ -58,7 +60,8 @@ public IDisposable StartConsuming()
             else
                 eventBus.Publish(new StartConsumingFailedEvent(this, queue));
 
-            return new ConsumerCancellation(Dispose);
+            consumerCancellation = new ConsumerCancellation(Dispose);
+            return consumerCancellation;
         }
 
         private bool disposed;
@@ -68,6 +71,8 @@ public void Dispose()
             if (disposed) return;
             disposed = true;
 
+            consumerCancellation.OnCancel(queue);
+
             eventBus.Publish(new StoppedConsumingEvent(this));
 
             internalConsumer?.Dispose();