Merge pull request #37 from AntonyVorontsov/feature/prefetch-handlers

Batch message handlers

Author: antonyvorontsov

RabbitMQ.Client.Core.DependencyInjection.sln

@@ -45,6 +45,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "RabbitMQ.Client.Core.Depend
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Examples.AdvancedConfiguration", "examples\Examples.AdvancedConfiguration\Examples.AdvancedConfiguration.csproj", "{0C713913-8D54-49E7-AADD-45497216E2EF}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Examples.BatchMessageHandler", "examples\Examples.BatchMessageHandler\Examples.BatchMessageHandler.csproj", "{BC9BE326-AEFE-42F2-B592-80126188514D}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -75,6 +77,10 @@ Global
 		{0C713913-8D54-49E7-AADD-45497216E2EF}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{0C713913-8D54-49E7-AADD-45497216E2EF}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{0C713913-8D54-49E7-AADD-45497216E2EF}.Release|Any CPU.Build.0 = Release|Any CPU
+		{BC9BE326-AEFE-42F2-B592-80126188514D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BC9BE326-AEFE-42F2-B592-80126188514D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{BC9BE326-AEFE-42F2-B592-80126188514D}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BC9BE326-AEFE-42F2-B592-80126188514D}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -87,6 +93,7 @@ Global
 		{93D59B0E-856C-4260-B50E-57FE5C0F5073} = {BC4CDBDE-4AEE-44B3-B00B-380EB1AC5E62}
 		{85907F19-00B0-4BA6-9B7C-0452A174903D} = {9D0289B2-C566-46CC-A53A-471BCBA0F277}
 		{0C713913-8D54-49E7-AADD-45497216E2EF} = {04EFD8F7-5120-4072-9728-64203F6F4873}
+		{BC9BE326-AEFE-42F2-B592-80126188514D} = {04EFD8F7-5120-4072-9728-64203F6F4873}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {0EBBD182-65B2-47F9-ABBE-64B5B8C9652F}

docs/changelog.md

@@ -2,6 +2,19 @@
 
 All notable changes to this library will be documented in this file.
 
+## [4.0.0] - 2020-05-05
+
+### Added
+
+- `BaseBatchMessageHandler` and `BatchMessageHandler` for handling messages in batches via prefetch count feature.
+- Example of basic usages of batch message handlers.
+
+### Updated
+
+- Updated RabbitMQ.Client to the newest version 6.0.0. Made some changes according to the breaking changes that come with the newest version of RabbitMQ.Client.
+- Moved message handlers to the different namespace `RabbitMQ.Client.Core.DependencyInjection.MessageHandlers`.
+- Moved internal DI extensions to the different namespace `RabbitMQ.Client.Core.DependencyInjection.InternalExtensions`.
+
 ## [3.2.1] - 2020-03-29
 
 ### Fixed

docs/message-consumption.md

@@ -311,6 +311,84 @@ The message handling process is organized as follows:
 - If any exception occurs `IMessageHandlingService` acknowledges the message anyway and checks if the message has to be re-send. If exchange option `RequeueFailedMessages` is set `true` then `IMessageHandlingService` adds a header `"requeued"` to the message and sends it again with delay in 60 seconds. Mechanism of sending delayed messages covered in the message production [documentation](message-production.md).
 - If any exception occurs within handling the message that has been already re-sent that message will not be re-send again (re-send happens only once).
 
+### Batch message handlers
+
+There are also a feature that you can use in case of necessity of handling messages in batches.
+First of all you have to create a class that inherits a `BatchMessageHandler` class.
+You have to set up values for `QueueName` and `PrefetchCount` properties. These values are responsible for the queue that will be read by the message handler, and the size of batches of messages.
+
+```c#
+public class CustomBatchMessageHandler : BatchMessageHandler
+{
+    readonly ILogger<CustomBatchMessageHandler> _logger;
+    
+    public CustomBatchMessageHandler(
+        IEnumerable<BatchConsumerConnectionOptions> batchConsumerConnectionOptions,
+        ILogger<CustomBatchMessageHandler> logger)
+        : base(batchConsumerConnectionOptions, logger)
+    {
+        _logger = logger;
+    }
+
+    protected override ushort PrefetchCount { get; set; } = 50;
+    
+    protected override string QueueName { get; set; } = "another.queue.name";
+    
+    protected override Task HandleMessage(IEnumerable<string> messages, CancellationToken cancellationToken)
+    {
+        _logger.LogInformation("Handling a batch of messages.");
+        foreach (var message in messages)
+        {
+            _logger.LogInformation(message);
+        }
+        return Task.CompletedTask;
+    }
+}
+```
+
+If you want to get raw messages as `ReadOnlyMemory<byte>` you can inherit base message handler class.
+
+```c#
+public class CustomBatchMessageHandler : BaseBatchMessageHandler
+{
+    readonly ILogger<CustomBatchMessageHandler> _logger;
+    
+    public CustomBatchMessageHandler(
+        IEnumerable<BatchConsumerConnectionOptions> batchConsumerConnectionOptions,
+        ILogger<CustomBatchMessageHandler> logger)
+        : base(batchConsumerConnectionOptions, logger)
+    {
+        _logger = logger;
+    }
+    
+    protected override ushort PrefetchCount { get; set; } = 3;
+
+    protected override string QueueName { get; set; } = "queue.name";
+    
+    protected override Task HandleMessages(IEnumerable<ReadOnlyMemory<byte>> messages, CancellationToken cancellationToken)
+    {
+        _logger.LogInformation("Handling a batch of messages.");
+        foreach (var message in messages)
+        {
+            var stringifiedMessage = Encoding.UTF8.GetString(message.ToArray());
+            _logger.LogInformation(stringifiedMessage);
+        }
+        return Task.CompletedTask;
+    }
+}
+```
+
+After all you have to register that batch message handler via DI.
+```c#
+services.AddBatchMessageHandler<CustomBatchMessageHandler>(Configuration.GetSection("RabbitMq"));
+```
+
+The message handler will create a separate connection and use it for reading messages.
+When the message collection is full to the size of `PrefetchCount` they are passed to the `HandleMessage` method.
+Both `BaseBatchMessageHandler` and `BatchMessageHandler` implement `IDisposable` interface, so you can use it for release of resources.
+
+Use this method of getting messages only when you sure that the number of messages that pass through this queue is really huge. Otherwise, messages could stack in the temporary collection of messages waiting to get in full.
+
 For message production features see the [Previous page](message-production.md)
 
 For more information about advanced usage of the RabbitMQ client see the [Next page](advanced-usage.md)

examples/Examples.AdvancedConfiguration/Controllers/ExampleController.cs

@@ -2,6 +2,7 @@
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.Extensions.Logging;
 using RabbitMQ.Client.Core.DependencyInjection;
+using RabbitMQ.Client.Core.DependencyInjection.Services;
 
 namespace Examples.AdvancedConfiguration.Controllers
 {

examples/Examples.AdvancedConfiguration/MessageHandlers/CustomAsyncMessageHandler.cs

@@ -1,5 +1,6 @@
 using System.Threading.Tasks;
 using RabbitMQ.Client.Core.DependencyInjection;
+using RabbitMQ.Client.Core.DependencyInjection.MessageHandlers;
 
 namespace Examples.AdvancedConfiguration.MessageHandlers
 {

examples/Examples.AdvancedConfiguration/MessageHandlers/CustomAsyncNonCyclicMessageHandler.cs

@@ -1,5 +1,7 @@
 using System.Threading.Tasks;
 using RabbitMQ.Client.Core.DependencyInjection;
+using RabbitMQ.Client.Core.DependencyInjection.MessageHandlers;
+using RabbitMQ.Client.Core.DependencyInjection.Services;
 
 namespace Examples.AdvancedConfiguration.MessageHandlers
 {

examples/Examples.AdvancedConfiguration/MessageHandlers/CustomMessageHandler.cs

@@ -1,4 +1,5 @@
 using RabbitMQ.Client.Core.DependencyInjection;
+using RabbitMQ.Client.Core.DependencyInjection.MessageHandlers;
 
 namespace Examples.AdvancedConfiguration.MessageHandlers
 {

examples/Examples.AdvancedConfiguration/MessageHandlers/CustomNonCyclicMessageHandler.cs

@@ -1,4 +1,6 @@
 using RabbitMQ.Client.Core.DependencyInjection;
+using RabbitMQ.Client.Core.DependencyInjection.MessageHandlers;
+using RabbitMQ.Client.Core.DependencyInjection.Services;
 
 namespace Examples.AdvancedConfiguration.MessageHandlers
 {

examples/Examples.AdvancedConfiguration/Services/ConsumingHostedService.cs

@@ -2,6 +2,7 @@
 using System.Threading.Tasks;
 using Microsoft.Extensions.Hosting;
 using RabbitMQ.Client.Core.DependencyInjection;
+using RabbitMQ.Client.Core.DependencyInjection.Services;
 
 namespace Examples.AdvancedConfiguration.Services
 {

examples/Examples.BatchMessageHandler/AnotherCustomBatchMessageHandler.cs

@@ -0,0 +1,35 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using RabbitMQ.Client.Core.DependencyInjection.Models;
+
+namespace Examples.BatchMessageHandler
+{
+    public class AnotherCustomBatchMessageHandler : RabbitMQ.Client.Core.DependencyInjection.BatchMessageHandlers.BatchMessageHandler
+    {
+        readonly ILogger<AnotherCustomBatchMessageHandler> _logger;
+        
+        public AnotherCustomBatchMessageHandler(
+            IEnumerable<BatchConsumerConnectionOptions> batchConsumerConnectionOptions,
+            ILogger<AnotherCustomBatchMessageHandler> logger)
+            : base(batchConsumerConnectionOptions, logger)
+        {
+            _logger = logger;
+        }
+
+        protected override ushort PrefetchCount { get; set; } = 5;
+        
+        protected override string QueueName { get; set; } = "another.queue.name";
+        
+        protected override Task HandleMessage(IEnumerable<string> messages, CancellationToken cancellationToken)
+        {
+            _logger.LogInformation("Handling a batch of messages.");
+            foreach (var message in messages)
+            {
+                _logger.LogInformation(message);
+            }
+            return Task.CompletedTask;
+        }
+    }
+}