Finished making batch message handlers.

Author: Anton Vorontsov

examples/Examples.BatchMessageHandler/AnotherCustomBatchMessageHandler.cs

@@ -1,13 +1,12 @@
-using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
-using RabbitMQ.Client.Core.DependencyInjection.BatchMessageHandlers;
 using RabbitMQ.Client.Core.DependencyInjection.Models;
 
 namespace Examples.BatchMessageHandler
 {
-    public class AnotherCustomBatchMessageHandler : BaseBatchMessageHandler
+    public class AnotherCustomBatchMessageHandler : RabbitMQ.Client.Core.DependencyInjection.BatchMessageHandlers.BatchMessageHandler
     {
         readonly ILogger<AnotherCustomBatchMessageHandler> _logger;
 
@@ -19,15 +18,17 @@ public AnotherCustomBatchMessageHandler(
             _logger = logger;
         }
 
-        protected override TimeSpan Period { get; set; } = TimeSpan.FromSeconds(10);
-        
         protected override ushort PrefetchCount { get; set; } = 5;
 
         protected override string QueueName { get; set; } = "another.queue.name";
 
-        protected override Task HandleMessages()
+        protected override Task HandleMessage(IEnumerable<string> messages, CancellationToken cancellationToken)
         {
-            _logger.LogInformation($"Handling from {typeof(AnotherCustomBatchMessageHandler)} messages");
+            _logger.LogInformation("Handling a batch of messages.");
+            foreach (var message in messages)
+            {
+                _logger.LogInformation(message);
+            }
             return Task.CompletedTask;
         }
     }

examples/Examples.BatchMessageHandler/CustomBatchMessageHandler.cs

@@ -1,5 +1,7 @@
 using System;
 using System.Collections.Generic;
+using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using RabbitMQ.Client.Core.DependencyInjection.BatchMessageHandlers;
@@ -18,16 +20,19 @@ public CustomBatchMessageHandler(
         {
             _logger = logger;
         }
-
-        protected override TimeSpan Period { get; set; } = TimeSpan.FromMinutes(10);
 
-        protected override ushort PrefetchCount { get; set; } = 5;
+        protected override ushort PrefetchCount { get; set; } = 3;
 
         protected override string QueueName { get; set; } = "queue.name";
 
-        protected override Task HandleMessages()
+        protected override Task HandleMessages(IEnumerable<ReadOnlyMemory<byte>> messages, CancellationToken cancellationToken)
         {
-            _logger.LogInformation("Handling messages");
+            _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;
         }
     }

examples/Examples.BatchMessageHandler/Program.cs

@@ -23,14 +23,14 @@ public static async Task Main()
                     services.AddBatchMessageHandler<CustomBatchMessageHandler>(hostContext.Configuration.GetSection("RabbitMq"));
 
                     // Second one - passing configuration instance.
-                    //  var rabbitMqConfiguration = new RabbitMqClientOptions
-                    //  {
-                    //      HostName = "127.0.0.1",
-                    //      Port = 5672,
-                    //      UserName = "guest",
-                    //      Password = "guest"
-                    //  };
-                    // services.AddBatchMessageHandler<AnotherCustomBatchMessageHandler>(rabbitMqConfiguration);
+                     var rabbitMqConfiguration = new RabbitMqClientOptions
+                     {
+                         HostName = "127.0.0.1",
+                         Port = 5672,
+                         UserName = "guest",
+                         Password = "guest"
+                     };
+                    services.AddBatchMessageHandler<AnotherCustomBatchMessageHandler>(rabbitMqConfiguration);
                 })
                 .ConfigureLogging((hostingContext, logging) => 
                 {

src/RabbitMQ.Client.Core.DependencyInjection/BatchMessageHandlers/BaseBatchMessageHandler.cs

@@ -1,7 +1,7 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Hosting;
@@ -14,21 +14,38 @@
 
 namespace RabbitMQ.Client.Core.DependencyInjection.BatchMessageHandlers
 {
+    /// <summary>
+    /// A message handler that handles messages in batches.
+    /// </summary>
     public abstract class BaseBatchMessageHandler : IHostedService, IDisposable
     {
-        protected virtual TimeSpan DueTo { get; set; } = TimeSpan.Zero;
+        /// <summary>
+        /// A connection which is in use by batch message handler.
+        /// </summary>
+        protected IConnection Connection { get; private set; }
 
-        protected abstract TimeSpan Period { get; set; }
-
+        /// <summary>
+        /// A channel that has been created using the connection.
+        /// </summary>
+        protected IModel Channel { get;  private set; }
+        
+        /// <summary>
+        /// Prefetch size value that can be overriden.
+        /// </summary>
         protected virtual uint PrefetchSize { get; set; } = 0;
 
+        /// <summary>
+        /// Queue name which will be read by that batch message handler.
+        /// </summary>
         protected abstract string QueueName { get; set; }
 
+        /// <summary>
+        /// Prefetch count value (batch size).
+        /// </summary>
         protected abstract ushort PrefetchCount { get; set; }
 
         readonly RabbitMqClientOptions _clientOptions;
         readonly ILogger<BaseBatchMessageHandler> _logger;
-        Timer _timer;
 
         protected BaseBatchMessageHandler(
             IEnumerable<BatchConsumerConnectionOptions> batchConsumerConnectionOptions,
@@ -47,12 +64,28 @@ protected BaseBatchMessageHandler(
         public Task StartAsync(CancellationToken cancellationToken)
         {
             ValidateProperties();
-            _logger.LogInformation("BatchMessageHandler has been started.");
-            _timer = new Timer(
-                async state => await StartPeriodicJob(cancellationToken),
-                null,
-                DueTo,
-                Period);
+            _logger.LogInformation($"Batch message handler {GetType()} has been started.");
+            Connection = RabbitMqFactoryExtensions.CreateRabbitMqConnection(_clientOptions);
+            Channel = Connection.CreateModel();
+            Channel.BasicQos(PrefetchSize, PrefetchCount, false);
+            
+            var messages = new ConcurrentBag<BasicDeliverEventArgs>();
+            var consumer = new AsyncEventingBasicConsumer(Channel);
+            consumer.Received += async (sender, eventArgs) =>
+            {
+                messages.Add(eventArgs);
+                if (messages.Count < PrefetchCount)
+                {
+                    return;
+                }
+                
+                var byteMessages = messages.Select(x => x.Body).ToList();
+                await HandleMessages(byteMessages, cancellationToken);
+                var latestDeliveryTag = messages.Max(x => x.DeliveryTag);
+                messages.Clear();
+                Channel.BasicAck(latestDeliveryTag, true);
+            };
+            Channel.BasicConsume(queue: QueueName, autoAck: false, consumer: consumer);
             return Task.CompletedTask;
         }
 
@@ -69,40 +102,24 @@ void ValidateProperties()
             }
         }
 
-        async Task StartPeriodicJob(CancellationToken cancellationToken)
-        {
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                using var connection = RabbitMqFactoryExtensions.CreateRabbitMqConnection(_clientOptions);
-                using var channel = connection.CreateModel();
-            
-                channel.BasicQos(PrefetchSize, PrefetchCount, false);
-            
-                var consumer = new AsyncEventingBasicConsumer(channel);
-                consumer.Received += async (sender, eventArgs) => await HandleMessageReceivingEvent(eventArgs);
-                channel.BasicConsume(queue: QueueName, autoAck: false, consumer: consumer);
-            }
-        }
-        
-        async Task HandleMessageReceivingEvent(BasicDeliverEventArgs eventArgs)
-        {
-            var message = Encoding.UTF8.GetString(eventArgs.Body.ToArray());
-            _logger.LogInformation($"A new message was received with deliveryTag {eventArgs.DeliveryTag}.");
-            _logger.LogInformation(message);
-        }
-
-        protected abstract Task HandleMessages();
+        /// <summary>
+        /// Handle a batch of messages.
+        /// </summary>
+        /// <param name="messages">A collection of messages as bytes.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        /// <returns></returns>
+        protected abstract Task HandleMessages(IEnumerable<ReadOnlyMemory<byte>> messages, CancellationToken cancellationToken);
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
-            _logger.LogInformation("BatchMessageHandler has been stopped.");
-            _timer?.Change(Timeout.Infinite, 0);
+            _logger.LogInformation($"Batch message handler {GetType()} has been stopped.");
             return Task.CompletedTask;
         }
 
         public void Dispose()
         {
-            _timer?.Dispose();
+            Connection?.Dispose();
+            Channel?.Dispose();
         }
     }
 }

src/RabbitMQ.Client.Core.DependencyInjection/BatchMessageHandlers/BatchMessageHandler.cs

@@ -0,0 +1,44 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using RabbitMQ.Client.Core.DependencyInjection.Models;
+
+namespace RabbitMQ.Client.Core.DependencyInjection.BatchMessageHandlers
+{
+    /// <summary>
+    /// Batch message handler that handles messages and passes them into HandleMessage method as strings.
+    /// </summary>
+    public abstract class BatchMessageHandler : BaseBatchMessageHandler
+    {
+        protected BatchMessageHandler(
+            IEnumerable<BatchConsumerConnectionOptions> batchConsumerConnectionOptions,
+            ILogger<BatchMessageHandler> logger) 
+            : base(batchConsumerConnectionOptions, logger)
+        {
+        }
+        
+        /// <summary>
+        /// Handle a batch of messages.
+        /// </summary>
+        /// <param name="messages">A collection of messages as bytes.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        /// <returns></returns>
+        protected override async Task HandleMessages(IEnumerable<ReadOnlyMemory<byte>> messages, CancellationToken cancellationToken)
+        {
+            var decodedMessages = messages.Select(x => Encoding.UTF8.GetString(x.ToArray()));
+            await HandleMessage(decodedMessages, cancellationToken);
+        }
+
+        /// <summary>
+        /// Handle a batch of decoded messages.
+        /// </summary>
+        /// <param name="messages">A collection of messages.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        /// <returns></returns>
+        protected abstract Task HandleMessage(IEnumerable<string> messages, CancellationToken cancellationToken);
+    }
+}