Added StopConsuming method for consumers.

Author: Anton Vorontsov

src/RabbitMQ.Client.Core.DependencyInjection/IConsumingService.cs

@@ -19,5 +19,10 @@ public interface IConsumingService
         /// Start consuming (getting messages).
         /// </summary>
         void StartConsuming();
+
+        /// <summary>
+        /// Stop consuming (getting messages).
+        /// </summary>
+        void StopConsuming();
     }
 }

src/RabbitMQ.Client.Core.DependencyInjection/QueueService.cs

@@ -33,6 +33,7 @@ internal sealed class QueueService : IQueueService, IDisposable
         readonly IEnumerable<RabbitMqExchange> _exchanges;
         readonly ILogger<QueueService> _logger;
 
+        IEnumerable<string> _consumerTags;
         bool _consumingStarted;
         readonly object _lock = new object();
 
@@ -123,16 +124,34 @@ public void StartConsuming()
                 return;
             }
 
-            _consumer.Received += async (sender, eventArgs) => await _messageHandlingService.HandleMessageReceivingEvent(eventArgs, this);
+            _consumer.Received += ConsumerOnReceived;
             _consumingStarted = true;
 
             var consumptionExchanges = _exchanges.Where(x => x.IsConsuming);
-            foreach (var exchange in consumptionExchanges)
+            _consumerTags = consumptionExchanges.SelectMany(
+                    exchange => exchange.Options.Queues.Select(
+                        queue => ConsumingChannel.BasicConsume(queue: queue.Name, autoAck: false, consumer: _consumer)))
+                .Distinct()
+                .ToList();
+        }
+
+        public void StopConsuming()
+        {
+            if (ConsumingChannel is null)
             {
-                foreach (var queue in exchange.Options.Queues)
-                {
-                    ConsumingChannel.BasicConsume(queue: queue.Name, autoAck: false, consumer: _consumer);
-                }
+                throw new ConsumingChannelIsNullException($"Consuming channel is null. Configure {nameof(IConsumingService)} or full functional {nameof(IQueueService)} for consuming messages.");
+            }
+
+            if (!_consumingStarted)
+            {
+                return;
+            }
+
+            _consumer.Received -= ConsumerOnReceived;
+            _consumingStarted = false;
+            foreach (var tag in _consumerTags)
+            {
+                ConsumingChannel.BasicCancel(tag);
             }
         }
 
@@ -465,5 +484,10 @@ static Dictionary<string, object> CreateArguments(string exchangeName, string ro
                 { "x-message-ttl", secondsDelay * 1000 },
                 { "x-expires", secondsDelay * 1000 + QueueExpirationTime }
             };
+        
+        async Task ConsumerOnReceived(object sender, BasicDeliverEventArgs eventArgs)
+        {
+            await _messageHandlingService.HandleMessageReceivingEvent(eventArgs, this);
+        }
     }
 }