CommandProcessor Bulk Clear #1764

src/Paramore.Brighter.Extensions.Hosting/TimedOutboxSweeper.cs

@@ -40,14 +40,17 @@ private void DoWork(object state)
             IAmACommandProcessor commandProcessor = scope.ServiceProvider.GetService<IAmACommandProcessor>();
 
             var outBoxSweeper = new OutboxSweeper(
-                milliSecondsSinceSent:_options.MinimumMessageAge, 
-                outbox:outbox, 
-                commandProcessor:commandProcessor);
+                milliSecondsSinceSent: _options.MinimumMessageAge, 
+                outbox: outbox, 
+                commandProcessor: commandProcessor, 
+                _options.UseBulk);
+
+            if(_options.UseBulk)
+                outBoxSweeper.SweepAsync(CancellationToken.None).RunSynchronously();
+            else
+                outBoxSweeper.Sweep();
 
-            outBoxSweeper.Sweep();
-
             s_logger.LogInformation("Outbox Sweeper sleeping");
-
         }
 
         public Task StopAsync(CancellationToken cancellationToken)

src/Paramore.Brighter.Extensions.Hosting/TimedOutboxSweeperOptions.cs

@@ -13,5 +13,10 @@ public class TimedOutboxSweeperOptions
         /// The age a message to pickup by the sweeper in milliseconds.
         /// </summary>
         public int MinimumMessageAge { get; set; } = 5000;
+
+        /// <summary>
+        /// Use bulk operations to dispatch messages.
+        /// </summary>
+        public bool UseBulk { get; set; } = false;
     }
 }

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusMessageProducer.cs

@@ -1,5 +1,7 @@
 using System;
+using System.Collections.Generic;
 using System.Linq;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using Microsoft.Extensions.Logging;
 using Paramore.Brighter.Logging;
@@ -14,7 +16,7 @@ namespace Paramore.Brighter.MessagingGateway.AzureServiceBus
     /// <summary>
     /// A Sync and Async Message Producer for Azure Service Bus.
     /// </summary>
-    public class AzureServiceBusMessageProducer : IAmAMessageProducerSync, IAmAMessageProducerAsync
+    public class AzureServiceBusMessageProducer : IAmAMessageProducerSync, IAmAMessageProducerAsync, IAmABulkMessageProducerAsync
     {
         public int MaxOutStandingMessages { get; set; } = -1;
         public int MaxOutStandingCheckIntervalMilliSeconds { get; set; } = 0;
@@ -52,6 +54,53 @@ public async Task SendAsync(Message message)
             await SendWithDelayAsync(message);
         }
 
+        /// <summary>
+        /// Sends a Batch of Messages
+        /// </summary>
+        /// <param name="messages">The messages to send.</param>
+        /// <param name="batchSize">The size of batches to send messages in.</param>
+        /// <param name="cancellationToken">The Cancellation Token.</param>
+        /// <returns>List of Messages successfully sent.</returns>
+        /// <exception cref="NotImplementedException"></exception>
+        public async IAsyncEnumerable<Guid[]> SendAsync(IEnumerable<Message> messages, int batchSize,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            var topics = messages.Select(m => m.Header.Topic).Distinct();
+            if (topics.Count() != 1)
+            {
+                s_logger.LogError("Cannot Bulk send for Multiple Topics, {NumberOfTopics} Topics Requested", topics.Count());
+                throw new Exception($"Cannot Bulk send for Multiple Topics, {topics.Count()} Topics Requested");
+            }
+            var topic = topics.Single();
+
+            var batches = Enumerable.Range(0, (int)Math.Ceiling((messages.Count() / (decimal)batchSize)))
+                .Select(i => new List<Message>(messages
+                    .Skip(i * batchSize)
+                    .Take(batchSize)
+                    .ToArray()));
+
+            var serviceBusSenderWrapper = GetSender(topic);
+
+            s_logger.LogInformation("Sending Messages for {TopicName} split into {NumberOfBatches} Batches of {BatchSize}", topic, batches.Count(), batchSize);
+            try
+            {
+                foreach (var batch in batches)
+                {
+                    var asbMessages = batch.Select(ConvertToServiceBusMessage).ToArray();
+
+                    s_logger.LogDebug("Publishing {NumberOfMessages} messages to topic {Topic}.",
+                        asbMessages.Length, topic);
+
+                    await serviceBusSenderWrapper.SendAsync(asbMessages, cancellationToken);
+                    yield return batch.Select(m => m.Id).ToArray();
+                }
+            }
+            finally
+            {
+                await serviceBusSenderWrapper.CloseAsync();
+            }
+        }
+
         /// <summary>
         /// Send the specified message with specified delay
         /// </summary>
@@ -71,46 +120,15 @@ public async Task SendWithDelayAsync(Message message, int delayMilliseconds = 0)
         {
             s_logger.LogDebug("Preparing  to send message on topic {Topic}", message.Header.Topic);
 
-            EnsureTopicExists(message.Header.Topic);
-
-            IServiceBusSenderWrapper serviceBusSenderWrapper;
-
-            try
-            {
-                RetryPolicy policy = Policy
-                    .Handle<Exception>()
-                    .Retry(TopicConnectionRetryCount, (exception, retryNumber) =>
-                    {
-                        s_logger.LogError(exception, "Failed to connect to topic {Topic}, retrying...", message.Header.Topic);
-
-                        Thread.Sleep(TimeSpan.FromMilliseconds(TopicConnectionSleepBetweenRetriesInMilliseconds));
-                    }
-                    );
-
-                serviceBusSenderWrapper = policy.Execute(() => _serviceBusSenderProvider.Get(message.Header.Topic));
-            }
-            catch (Exception e)
-            {
-                s_logger.LogError(e, "Failed to connect to topic {Topic}, aborting.", message.Header.Topic);
-                throw;
-            }
+            var serviceBusSenderWrapper = GetSender(message.Header.Topic);
 
             try
             {
                 s_logger.LogDebug(
                     "Publishing message to topic {Topic} with a delay of {Delay} and body {Request} and id {Id}.",
                     message.Header.Topic, delayMilliseconds, message.Body.Value, message.Id);
 
-                var azureServiceBusMessage = new ServiceBusMessage(message.Body.Bytes);
-                azureServiceBusMessage.ApplicationProperties.Add(ASBConstants.MessageTypeHeaderBagKey, message.Header.MessageType.ToString());
-                azureServiceBusMessage.ApplicationProperties.Add(ASBConstants.HandledCountHeaderBagKey, message.Header.HandledCount);
-                foreach (var header in message.Header.Bag.Where(h => !ASBConstants.ReservedHeaders.Contains(h.Key)))
-                {
-                    azureServiceBusMessage.ApplicationProperties.Add(header.Key, header.Value);
-                }
-                azureServiceBusMessage.CorrelationId = message.Header.CorrelationId.ToString();
-                azureServiceBusMessage.ContentType = message.Header.ContentType;
-                azureServiceBusMessage.MessageId = message.Header.Id.ToString();
+                var azureServiceBusMessage = ConvertToServiceBusMessage(message);
                 if (delayMilliseconds == 0)
                 {
                     await serviceBusSenderWrapper.SendAsync(azureServiceBusMessage);
@@ -140,6 +158,48 @@ public void Dispose()
         {
         }
 
+        private IServiceBusSenderWrapper GetSender(string topic)
+        {
+            EnsureTopicExists(topic);
+
+            try
+            {
+                RetryPolicy policy = Policy
+                    .Handle<Exception>()
+                    .Retry(TopicConnectionRetryCount, (exception, retryNumber) =>
+                        {
+                            s_logger.LogError(exception, "Failed to connect to topic {Topic}, retrying...",
+                                topic);
+
+                            Thread.Sleep(TimeSpan.FromMilliseconds(TopicConnectionSleepBetweenRetriesInMilliseconds));
+                        }
+                    );
+
+                return policy.Execute(() => _serviceBusSenderProvider.Get(topic));
+            }
+            catch (Exception e)
+            {
+                s_logger.LogError(e, "Failed to connect to topic {Topic}, aborting.", topic);
+                throw;
+            }
+        }
+
+        private ServiceBusMessage ConvertToServiceBusMessage(Message message)
+        {
+            var azureServiceBusMessage = new ServiceBusMessage(message.Body.Bytes);
+            azureServiceBusMessage.ApplicationProperties.Add(ASBConstants.MessageTypeHeaderBagKey, message.Header.MessageType.ToString());
+            azureServiceBusMessage.ApplicationProperties.Add(ASBConstants.HandledCountHeaderBagKey, message.Header.HandledCount);
+            foreach (var header in message.Header.Bag.Where(h => !ASBConstants.ReservedHeaders.Contains(h.Key)))
+            {
+                azureServiceBusMessage.ApplicationProperties.Add(header.Key, header.Value);
+            }
+            azureServiceBusMessage.CorrelationId = message.Header.CorrelationId.ToString();
+            azureServiceBusMessage.ContentType = message.Header.ContentType;
+            azureServiceBusMessage.MessageId = message.Header.Id.ToString();
+
+            return azureServiceBusMessage;
+        }
+
         private void EnsureTopicExists(string topic)
         {
             if (_topicCreated || _makeChannel.Equals(OnMissingChannel.Assume))

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusWrappers/IServiceBusSenderWrapper.cs

@@ -35,6 +35,13 @@ public interface IServiceBusSenderWrapper
         /// <param name="cancellationToken">Cancellation Token.</param>
         Task SendAsync(ServiceBusMessage message, CancellationToken cancellationToken = default(CancellationToken));
 
+        /// <summary>
+        /// Send Messages
+        /// </summary>
+        /// <param name="messages">The messages to send.</param>
+        /// <param name="cancellationToken">Cancellation Token.</param>
+        Task SendAsync(ServiceBusMessage[] messages, CancellationToken cancellationToken = default(CancellationToken));
+
         /// <summary>
         /// Schedule a message to be sent.
         /// </summary>

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusWrappers/ServiceBusSenderWrapper.cs

@@ -48,6 +48,11 @@ public async Task SendAsync(ServiceBusMessage message,
             }
         }
 
+        public Task SendAsync(ServiceBusMessage[] messages, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            return _serviceBusSender.SendMessagesAsync(messages, cancellationToken);
+        }
+
         public void ScheduleMessage(ServiceBusMessage message, DateTimeOffset scheduleEnqueueTime)
         {
             _serviceBusSender.ScheduleMessageAsync(message, scheduleEnqueueTime).Wait();

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/Paramore.Brighter.MessagingGateway.AzureServiceBus.csproj

@@ -4,6 +4,7 @@
     <Authors>Yiannis Triantafyllopoulos</Authors>
     <TargetFrameworks>netstandard2.0</TargetFrameworks>
     <PackageTags>awssqs;AMQP;Command;Event;Service Activator;Decoupled;Invocation;Messaging;Remote;Command Dispatcher;Command Processor;Request;Service;Task Queue;Work Queue;Retry;Circuit Breaker;Availability</PackageTags>
+    <LangVersion>8.0</LangVersion>
   </PropertyGroup>
   <ItemGroup>
     <PackageReference Include="Azure.Identity" Version="1.5.0" />

src/Paramore.Brighter.Outbox.DynamoDB/DynamoDbOutboxSync.cs

@@ -153,6 +153,18 @@ public Message Get(Guid messageId, int outBoxTimeout = -1)
                 .ConfigureAwait(ContinueOnCapturedContext);
         }
 
+        public async Task<IEnumerable<Message>> GetAsync(IEnumerable<Guid> messageIds, int outBoxTimeout = -1,
+            CancellationToken cancellationToken = default(CancellationToken))
+        {
+            var messages = new List<Message>();
+            foreach (var messageId in messageIds)
+            {
+                messages.Add(await GetAsync(messageId, -1, cancellationToken));
+            }
+
+            return messages;
+        }
+
         /// <summary>
         /// Get paginated list of Messages.
         /// </summary>
@@ -200,7 +212,16 @@ await _context.SaveAsync(
                 new DynamoDBOperationConfig{OverrideTableName = _configuration.TableName},
                 cancellationToken);
        }
-
+
+        public async Task MarkDispatchedAsync(IEnumerable<Guid> ids, DateTime? dispatchedAt = null, Dictionary<string, object> args = null,
+            CancellationToken cancellationToken = default(CancellationToken))
+        {
+            foreach(var messageId in ids)
+            {
+                await MarkDispatchedAsync(messageId, dispatchedAt, args, cancellationToken);
+            }
+        }
+
         /// <summary>
         /// Update a message to show it is dispatched
         /// </summary>

src/Paramore.Brighter.Outbox.EventStore/EventStoreOutboxSync.cs

@@ -201,6 +201,12 @@ public Task<Message> GetAsync(
             throw new NotImplementedException();
         }
 
+        public Task<IEnumerable<Message>> GetAsync(IEnumerable<Guid> messageIds, int outBoxTimeout = -1,
+            CancellationToken cancellationToken = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
         /// <summary>
         ///     Returns multiple events from a given stream.
         ///     If all the events do not exist, as many as can be found will be returned.
@@ -258,6 +264,12 @@ public async Task MarkDispatchedAsync(Guid id, DateTime? dispatchedAt = null, Di
             await _eventStore.AppendToStreamAsync(stream, nextEventNumber.Value, eventData);
         }
 
+        public Task MarkDispatchedAsync(IEnumerable<Guid> ids, DateTime? dispatchedAt = null, Dictionary<string, object> args = null,
+            CancellationToken cancellationToken = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
         /// <summary>
         /// Update a message to show it is dispatched
         /// </summary>