Merge pull request #30 from AntonyVorontsov/feature/stop-consuming

StopConsuming method for consumers.

Author: antonyvorontsov

docs/message-consumption.md

@@ -33,7 +33,7 @@ You can register `IHostedService` and inject an instance of `IQueueService` into
 services.AddSingleton<IHostedService, ConsumingService>();
 ```
 
-And then simply call `StartConsuming` so a consumer can work in the background.
+Then simply call `StartConsuming` so a consumer can work in the background. There is also an option which allows you to stop consuming messages - method `StopConsuming` which you can use any time you want to pause a message consumption for any reason.
 
 ```c#
 public class ConsumingService : IHostedService
@@ -59,6 +59,7 @@ public class ConsumingService : IHostedService
     public Task StopAsync(CancellationToken cancellationToken)
     {
         _logger.LogInformation("Stopping consuming.");
+        _queueService.StopConsuming();
         return Task.CompletedTask;
     }
 }

src/RabbitMQ.Client.Core.DependencyInjection/Services/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/Services/QueueService.cs

@@ -33,6 +33,7 @@ public sealed class QueueService : IQueueService, IDisposable
         readonly IEnumerable<RabbitMqExchange> _exchanges;
         readonly ILogger<QueueService> _logger;
 
+        IEnumerable<string> _consumerTags = new List<string>();
         bool _consumingStarted;
         readonly object _lock = new object();
 
@@ -131,16 +132,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);
             }
         }
 
@@ -479,5 +498,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);
+        }
     }
 }

tests/RabbitMQ.Client.Core.DependencyInjection.Tests/UnitTests/QueueServiceConsumerTests.cs

@@ -65,6 +65,58 @@ await consumer.HandleBasicDeliver(
             messageHandlingServiceMock.Verify(x => x.HandleMessageReceivingEvent(It.IsAny<BasicDeliverEventArgs>(), It.IsAny<IQueueService>()), Times.Exactly(numberOfMessages));
         }
 
+        [Theory]
+        [InlineData(1)]
+        [InlineData(5)]
+        [InlineData(10)]
+        [InlineData(15)]
+        [InlineData(20)]
+        [InlineData(25)]
+        public async Task ShouldProperlyStopConsumingMessages(int numberOfMessages)
+        {
+            var channelMock = new Mock<IModel>();
+            var connectionMock = new Mock<IConnection>();
+            connectionMock.Setup(x => x.CreateModel())
+                .Returns(channelMock.Object);
+            
+            var connectionFactoryMock = new Mock<IRabbitMqConnectionFactory>();
+            connectionFactoryMock.Setup(x => x.CreateRabbitMqConnection(It.IsAny<RabbitMqClientOptions>()))
+                .Returns(connectionMock.Object);
+
+            var consumer = new AsyncEventingBasicConsumer(channelMock.Object);
+            connectionFactoryMock.Setup(x => x.CreateConsumer(It.IsAny<IModel>()))
+                .Returns(consumer);
+
+            var messageHandlingServiceMock = new Mock<IMessageHandlingService>();
+            var queueService = CreateService(connectionFactoryMock.Object, messageHandlingServiceMock.Object);
+            queueService.StartConsuming();
+            for (var i = 1; i <= numberOfMessages; i++)
+            {
+                await consumer.HandleBasicDeliver(
+                    "1",
+                    (ulong)numberOfMessages,
+                    false,
+                    "exchange",
+                    "routing,key",
+                    null,
+                    new ReadOnlyMemory<byte>());
+            }
+
+            messageHandlingServiceMock.Verify(x => x.HandleMessageReceivingEvent(It.IsAny<BasicDeliverEventArgs>(), It.IsAny<IQueueService>()), Times.Exactly(numberOfMessages));
+            
+            queueService.StopConsuming();
+            await consumer.HandleBasicDeliver(
+                "1",
+                0,
+                false,
+                "exchange",
+                "routing,key",
+                null,
+                new ReadOnlyMemory<byte>());
+            
+            messageHandlingServiceMock.Verify(x => x.HandleMessageReceivingEvent(It.IsAny<BasicDeliverEventArgs>(), It.IsAny<IQueueService>()), Times.Exactly(numberOfMessages));
+        }
+
         static IConsumingService CreateService(
             IRabbitMqConnectionFactory connectionFactory,
             IMessageHandlingService messageHandlingService)