Fixed concurrency in getting messages while processing a batch.

Author: Anton Vorontsov

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

@@ -55,6 +55,7 @@ public abstract class BaseBatchMessageHandler : IHostedService, IDisposable
 
         readonly ConcurrentBag<BasicDeliverEventArgs> _messages = new ConcurrentBag<BasicDeliverEventArgs>();
         Timer _timer;
+        object _lock = new object();
         bool _disposed = false;
 
         protected BaseBatchMessageHandler(
@@ -104,18 +105,34 @@ public Task StartAsync(CancellationToken cancellationToken)
 
         async Task ProcessBatchOfMessages(CancellationToken cancellationToken)
         {
-            if (!_messages.Any())
+
+            var messages = GetMessages();
+            if (!messages.Any())
             {
                 return;
             }
 
-            var byteMessages = _messages.Select(x => x.Body).ToList();
+            var byteMessages = messages.Select(x => x.Body).ToList();
             await HandleMessages(byteMessages, cancellationToken).ConfigureAwait(false);
-            var latestDeliveryTag = _messages.Max(x => x.DeliveryTag);
-            _messages.Clear();
+            var latestDeliveryTag = messages.Max(x => x.DeliveryTag);
             Channel.BasicAck(latestDeliveryTag, true);
         }
 
+        IList<BasicDeliverEventArgs> GetMessages()
+        {
+            lock (_lock)
+            {
+                if (!_messages.Any())
+                {
+                    return new List<BasicDeliverEventArgs>();
+                }
+                
+                var messages = _messages.ToList();
+                _messages.Clear();
+                return messages;
+            }
+        }
+
         void ValidateProperties()
         {
             if (string.IsNullOrEmpty(QueueName))

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

@@ -123,10 +123,11 @@ await consumer.HandleBasicDeliver(
                 }
 
                 waitHandle.WaitOne(_globalTestsTimeout);
-                callerMock.Verify(x => x.EmptyCall(), Times.Exactly(b + 1));
-                callerMock.Verify(x => x.Call(It.IsAny<ReadOnlyMemory<byte>>()), Times.Exactly(upperBound));
             }
 
+            callerMock.Verify(x => x.EmptyCall(), Times.Exactly(numberOfSmallBatches));
+            callerMock.Verify(x => x.Call(It.IsAny<ReadOnlyMemory<byte>>()), Times.Exactly(numberOfMessages));
+            
             await messageHandler.StopAsync(CancellationToken.None);
         }
 

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

@@ -123,10 +123,11 @@ await consumer.HandleBasicDeliver(
                 }
 
                 waitHandle.WaitOne(_globalTestsTimeout);
-                callerMock.Verify(x => x.EmptyCall(), Times.Exactly(b + 1));
-                callerMock.Verify(x => x.Call(It.IsAny<string>()), Times.Exactly(upperBound));
             }
 
+            callerMock.Verify(x => x.EmptyCall(), Times.Exactly(numberOfSmallBatches));
+            callerMock.Verify(x => x.Call(It.IsAny<string>()), Times.Exactly(numberOfMessages));
+            
             await messageHandler.StopAsync(CancellationToken.None);
         }