Refactor

zarusz · zarusz · commit 11c7c487371b · 2024-06-23T17:41:00.000+02:00
Signed-off-by: Tomasz Maruszak &lt;maruszaktomasz@gmail.com&gt;
diff --git a/src/Host.Plugin.Properties.xml b/src/Host.Plugin.Properties.xml
@@ -4,7 +4,7 @@
   <Import Project="Common.NuGet.Properties.xml" />
 
   <PropertyGroup>
-    <Version>2.4.0-rc7</Version>
+    <Version>2.4.0-rc8</Version>
   </PropertyGroup>
 
 </Project>
diff --git a/src/SlimMessageBus.Host.AzureEventHub/EventHubMessageBus.cs b/src/SlimMessageBus.Host.AzureEventHub/EventHubMessageBus.cs
@@ -104,43 +104,42 @@ protected override async Task OnStart()
         }
     }
 
-    protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
+    protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
         AssertActive();
 
         var dispatched = new List<T>(envelopes.Count);
         try
         {
-            var messages = envelopes
+            var messagesByPartition = envelopes
                 .Where(x => x.Message != null)
-                .Select(
-                    envelope =>
-                    {
-                        var messageType = envelope.Message?.GetType();
-                        var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);
+                .Select(envelope =>
+                {
+                    var messageType = envelope.Message?.GetType();
+                    var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);
 
-                        _logger.LogDebug("Producing message {Message} of Type {MessageType} on Path {Path} with Size {MessageSize}", envelope.Message, messageType?.Name, path, messagePayload?.Length ?? 0);
+                    _logger.LogDebug("Producing message {Message} of Type {MessageType} on Path {Path} with Size {MessageSize}", envelope.Message, messageType?.Name, path, messagePayload?.Length ?? 0);
 
-                        var ev = envelope.Message != null ? new EventData(messagePayload) : new EventData();
+                    var ev = envelope.Message != null ? new EventData(messagePayload) : new EventData();
 
-                        if (envelope.Headers != null)
+                    if (envelope.Headers != null)
+                    {
+                        foreach (var header in envelope.Headers)
                         {
-                            foreach (var header in envelope.Headers)
-                            {
-                                ev.Properties.Add(header.Key, header.Value);
-                            }
+                            ev.Properties.Add(header.Key, header.Value);
                         }
+                    }
 
-                        var partitionKey = messageType != null
-                            ? GetPartitionKey(messageType, envelope.Message)
-                            : null;
+                    var partitionKey = messageType != null
+                        ? GetPartitionKey(messageType, envelope.Message)
+                        : null;
 
-                        return (Envelope: envelope, Message: ev, PartitionKey: partitionKey);
-                    })
-                    .GroupBy(x => x.PartitionKey);
+                    return (Envelope: envelope, Message: ev, PartitionKey: partitionKey);
+                })
+                .GroupBy(x => x.PartitionKey);
 
             var producer = _producerByPath[path];
-            foreach (var partition in messages)
+            foreach (var partition in messagesByPartition)
             {
                 EventDataBatch batch = null;
                 try
@@ -184,7 +183,7 @@ protected override async Task OnStart()
                         batch = null;
                     }
 
-                    return (dispatched, null);
+                    return new(dispatched, null);
                 }
                 finally
                 {
@@ -194,10 +193,10 @@ protected override async Task OnStart()
         }
         catch (Exception ex)
         {
-            return (dispatched, ex);
+            return new(dispatched, ex);
         }
 
-        return (dispatched, null);
+        return new(dispatched, null);
     }
 
     #endregion
diff --git a/src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs b/src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs
@@ -116,15 +116,14 @@ void AddConsumerFrom(TopicSubscriptionParams topicSubscription, IMessageProcesso
         }
     }
 
-    protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
+    protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
-        async Task SendBatchAsync(ServiceBusSender senderClient, ServiceBusMessageBatch batch, CancellationToken cancellationToken)
-        {
-            await Retry.WithDelay(
+        Task SendBatchAsync(ServiceBusSender senderClient, ServiceBusMessageBatch batch, CancellationToken cancellationToken) =>
+            Retry.WithDelay(
                 async cancellationToken =>
                 {
                     await senderClient.SendMessagesAsync(batch, cancellationToken).ConfigureAwait(false);
-                    _logger.LogDebug("Batch of {BatchSize} message(s) dispatched to {path} ({SizeInBytes} bytes)", batch.Count, path, batch.SizeInBytes);
+                    _logger.LogDebug("Batch of {BatchSize} message(s) dispatched to {Path} ({SizeInBytes} bytes)", batch.Count, path, batch.SizeInBytes);
                 },
                 (exception, attempt) =>
                 {
@@ -135,18 +134,16 @@ await Retry.WithDelay(
                         _logger.LogWarning("Service bus throttled. Backing off (Attempt: {Attempt}).", attempt);
                         return true;
                     }
-
                     return false;
                 },
-                TimeSpan.FromSeconds(2),
-                TimeSpan.FromSeconds(1),
+                delay: TimeSpan.FromSeconds(2),
+                jitter: TimeSpan.FromSeconds(1),
                 cancellationToken);
-        }
 
         AssertActive();
 
-        var messages = envelopes.Select(
-            envelope =>
+        var messages = envelopes
+            .Select(envelope =>
             {
                 var messageType = envelope.Message?.GetType();
                 var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);
@@ -190,12 +187,12 @@ await Retry.WithDelay(
                 await senderClient.SendMessageAsync(item.ServiceBusMessage, cancellationToken: cancellationToken).ConfigureAwait(false);
                 _logger.LogDebug("Delivered item {Message} of type {MessageType} to {Path}", item.Envelope.Message, item.Envelope.MessageType?.Name, path);
 
-                return ([item.Envelope], null);
+                return new([item.Envelope], null);
             }
             catch (Exception ex)
             {
                 _logger.LogDebug(ex, "Producing message {Message} of type {MessageType} to path {Path} resulted in error {Error}", item.Envelope.Message, item.Envelope.MessageType?.Name, path, ex.Message);
-                return ([], ex);
+                return new([], ex);
             }
         }
 
@@ -233,12 +230,12 @@ await Retry.WithDelay(
                 batch = null;
             }
 
-            return (dispatched, null);
+            return new(dispatched, null);
         }
         catch (Exception ex)
         {
             _logger.LogError(ex, "Producing message batch to path {Path} resulted in error {Error}", path, ex.Message);
-            return (dispatched, ex);
+            return new(dispatched, ex);
         }
         finally
         {
diff --git a/src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs b/src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs
@@ -114,7 +114,7 @@ protected override async ValueTask DisposeAsyncCore()
         }
     }
 
-    protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
+    protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
         AssertActive();
 
@@ -172,10 +172,10 @@ protected override async ValueTask DisposeAsyncCore()
         }
         catch (Exception ex)
         {
-            return (dispatched, ex);
+            return new(dispatched, ex);
         }
 
-        return (dispatched, null);
+        return new(dispatched, null);
     }
 
     protected byte[] GetMessageKey(ProducerSettings producerSettings, Type messageType, object message, string topic)
diff --git a/src/SlimMessageBus.Host.Memory/MemoryMessageBus.cs b/src/SlimMessageBus.Host.Memory/MemoryMessageBus.cs
@@ -123,8 +123,8 @@ private IMessageProcessorQueue CreateMessageProcessorQueue(IMessageProcessor<obj
             : new MessageProcessorQueue(messageProcessor, LoggerFactory.CreateLogger<MessageProcessorQueue>(), CancellationToken);
     }
 
-    protected override Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
-        => Task.FromResult<(IReadOnlyCollection<T> Dispatched, Exception Exception)>(([], null)); // Not used
+    protected override Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
+        => Task.FromResult<ProduceToTransportBulkResult<T>>(new([], null)); // Not used
 
     public override Task ProduceResponse(string requestId, object request, IReadOnlyDictionary<string, object> requestHeaders, object response, Exception responseException, IMessageTypeConsumerInvokerSettings consumerInvoker)
         => Task.CompletedTask; // Not used to responses
diff --git a/src/SlimMessageBus.Host.Mqtt/MqttMessageBus.cs b/src/SlimMessageBus.Host.Mqtt/MqttMessageBus.cs
@@ -1,7 +1,5 @@
 ﻿namespace SlimMessageBus.Host.Mqtt;
 
-using System.Security.Cryptography;
-
 using MQTTnet.Extensions.ManagedClient;
 
 public class MqttMessageBus : MessageBusBase<MqttMessageBusSettings>
@@ -103,10 +101,10 @@ private Task OnMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
         return Task.CompletedTask;
     }
 
-    protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
+    protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
-        var messages = envelopes.Select(
-            envelope =>
+        var messages = envelopes
+            .Select(envelope =>
             {
                 var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);
 
@@ -156,10 +154,10 @@ private Task OnMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
             }
             catch (Exception ex)
             {
-                return (dispatched, ex);
+                return new(dispatched, ex);
             }
         }
 
-        return (dispatched, null);
+        return new(dispatched, null);
     }
 }
diff --git a/src/SlimMessageBus.Host.Outbox/Interceptors/OutboxSendingTask.cs b/src/SlimMessageBus.Host.Outbox/Interceptors/OutboxSendingTask.cs
@@ -252,7 +252,7 @@ internal async Task<int> SendMessages(IServiceProvider serviceProvider, IOutboxR
                         outboxMessage =>
                         {
                             var message = bus.Serializer.Deserialize(outboxMessage.MessageType, outboxMessage.MessagePayload);
-                            return new EnvelopeWithId(outboxMessage.Id, message, outboxMessage.MessageType, outboxMessage.Headers ?? new Dictionary<string, object>());
+                            return new OutboxBulkMessage(outboxMessage.Id, message, outboxMessage.MessageType, outboxMessage.Headers ?? new Dictionary<string, object>());
                         })
                         .Batch(bulkProducer.MaxMessagesPerTransaction ?? defaultBatchSize);
 
@@ -271,12 +271,12 @@ internal async Task<int> SendMessages(IServiceProvider serviceProvider, IOutboxR
         return count;
     }
 
-    internal async Task<(bool Success, int Published)> DispatchBatchAsync(IOutboxRepository outboxRepository, IMessageBusBulkProducer producer, IMessageBusTarget messageBusTarget, IReadOnlyCollection<EnvelopeWithId> batch, string busName, string path, CancellationToken cancellationToken)
+    internal async Task<(bool Success, int Published)> DispatchBatchAsync(IOutboxRepository outboxRepository, IMessageBusBulkProducer producer, IMessageBusTarget messageBusTarget, IReadOnlyCollection<OutboxBulkMessage> batch, string busName, string path, CancellationToken cancellationToken)
     {
         _logger.LogDebug("Publishing batch of {MessageCount} messages to pathGroup {Path} on {BusName} bus", batch.Count, path, busName);
 
         // TOOD: Enclose in a transaction
-        var results = await producer.ProduceToTransport(batch, path, messageBusTarget, cancellationToken).ConfigureAwait(false);
+        var results = await producer.ProduceToTransportBulk(batch, path, messageBusTarget, cancellationToken).ConfigureAwait(false);
         if (cancellationToken.IsCancellationRequested && results.Dispatched.Count == 0)
         {
             // if cancellation has been requested, only return if no messages were published
@@ -311,14 +311,14 @@ private static IMasterMessageBus GetBus(ICompositeMessageBus compositeMessageBus
         return null;
     }
 
-    public record EnvelopeWithId : Envelope
+    public record OutboxBulkMessage : BulkMessageEnvelope
     {
-        public EnvelopeWithId(Guid id, object Message, Type MessageType, IDictionary<string, object> Headers)
-            : base(Message, MessageType, Headers)
+        public Guid Id { get; }
+
+        public OutboxBulkMessage(Guid id, object message, Type messageType, IDictionary<string, object> headers)
+            : base(message, messageType, headers)
         {
             Id = id;
         }
-
-        public Guid Id { get; }
     }
 }
diff --git a/src/SlimMessageBus.Host.RabbitMQ/RabbitMqMessageBus.cs b/src/SlimMessageBus.Host.RabbitMQ/RabbitMqMessageBus.cs
@@ -128,7 +128,7 @@ protected override async ValueTask DisposeAsyncCore()
         }
     }
 
-    protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
+    protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
         await EnsureInitFinished();
 
@@ -178,9 +178,9 @@ protected override async ValueTask DisposeAsyncCore()
         }
         catch (Exception ex)
         {
-            return (dispatched, ex);
+            return new(dispatched, ex);
         }
 
-        return (dispatched, null);
+        return new(dispatched, null);
     }
 }
diff --git a/src/SlimMessageBus.Host.Redis/RedisMessageBus.cs b/src/SlimMessageBus.Host.Redis/RedisMessageBus.cs
@@ -147,9 +147,13 @@ void AddTopicConsumer(string topic, ISubscriber subscriber, IMessageProcessor<Me
 
     #region Overrides of MessageBusBase
 
-    protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
+    protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
+#if NETSTANDARD2_0
         if (envelopes is null) throw new ArgumentNullException(nameof(envelopes));
+#else
+        ArgumentNullException.ThrowIfNull(envelopes);
+#endif
 
         AssertActive();
 
@@ -184,10 +188,10 @@ void AddTopicConsumer(string topic, ISubscriber subscriber, IMessageProcessor<Me
         }
         catch (Exception ex)
         {
-            return (dispatched, ex);
+            return new(dispatched, ex);
         }
 
-        return (dispatched, null);
+        return new(dispatched, null);
     }
 
     #endregion
diff --git a/src/SlimMessageBus.Host.Sql/GlobalUsings.cs b/src/SlimMessageBus.Host.Sql/GlobalUsings.cs
@@ -1,6 +1,5 @@
-﻿global using System.Data;
-
-global using Microsoft.Data.SqlClient;
+﻿global using Microsoft.Data.SqlClient;
+global using Microsoft.Extensions.DependencyInjection;
 global using Microsoft.Extensions.DependencyInjection.Extensions;
 global using Microsoft.Extensions.Logging;
 
diff --git a/src/SlimMessageBus.Host.Sql/SqlMessageBus.cs b/src/SlimMessageBus.Host.Sql/SqlMessageBus.cs
@@ -1,7 +1,5 @@
 ﻿namespace SlimMessageBus.Host.Sql;
 
-using Microsoft.Extensions.DependencyInjection;
-
 public class SqlMessageBus : MessageBusBase<SqlMessageBusSettings>
 {
     public SqlMessageBus(MessageBusSettings settings, SqlMessageBusSettings providerSettings)
@@ -27,12 +25,12 @@ public override async Task ProvisionTopology()
         await provisioningService.Migrate(CancellationToken); // provisining happens asynchronously
     }
 
-    protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)
+    protected override Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
         var sqlRepository = targetBus.ServiceProvider.GetService<ISqlRepository>();
 
         // ToDo: Save to table
 
-        return ([], new NotImplementedException());
+        return Task.FromResult(new ProduceToTransportBulkResult<T>([], new NotImplementedException()));
     }
 }
diff --git a/src/SlimMessageBus.Host/IMessageBusBulkProducer.cs b/src/SlimMessageBus.Host/IMessageBusBulkProducer.cs
@@ -11,8 +11,11 @@ public interface IMessageBusBulkProducer
     /// <summary>
     /// The maximum number of messages that can take part in a <see cref="TransactionScope"/>. Null if transaction scopes are not supported by the message bus.
     /// </summary>
-    int? MaxMessagesPerTransaction { get; }
-    Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default) where T: Envelope;
+    int? MaxMessagesPerTransaction { get; }
+
+    Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default) where T : BulkMessageEnvelope;
 }
 
-public record Envelope(object Message, Type MessageType, IDictionary<string, object> Headers);
+public record BulkMessageEnvelope(object Message, Type MessageType, IDictionary<string, object> Headers);
+
+public record ProduceToTransportBulkResult<T>(IReadOnlyCollection<T> Dispatched, Exception Exception);
diff --git a/src/SlimMessageBus.Host/MessageBusBase.cs b/src/SlimMessageBus.Host/MessageBusBase.cs
diff --git a/src/SlimMessageBus.Host/Retry.cs b/src/SlimMessageBus.Host/Retry.cs
diff --git a/src/Tests/SlimMessageBus.Host.Outbox.Test/Interceptors/OutboxSendingTaskTests.cs b/src/Tests/SlimMessageBus.Host.Outbox.Test/Interceptors/OutboxSendingTaskTests.cs
diff --git a/src/Tests/SlimMessageBus.Host.Test/MessageBusTested.cs b/src/Tests/SlimMessageBus.Host.Test/MessageBusTested.cs

Original file line number	Diff line number	Diff line change
`@@ -104,43 +104,42 @@ protected override async Task OnStart()`
`104`	`104`	`}`
`105`	`105`	`}`
`106`	`106`
`107`		`- protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)`
	`107`	`+ protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)`
`108`	`108`	`{`
`109`	`109`	`AssertActive();`
`110`	`110`
`111`	`111`	`var dispatched = new List<T>(envelopes.Count);`
`112`	`112`	`try`
`113`	`113`	`{`
`114`		`- var messages = envelopes`
	`114`	`+ var messagesByPartition = envelopes`
`115`	`115`	`.Where(x => x.Message != null)`
`116`		`- .Select(`
`117`		`- envelope =>`
`118`		`- {`
`119`		`- var messageType = envelope.Message?.GetType();`
`120`		`- var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);`
	`116`	`+ .Select(envelope =>`
	`117`	`+ {`
	`118`	`+ var messageType = envelope.Message?.GetType();`
	`119`	`+ var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);`
`121`	`120`
`122`		`- _logger.LogDebug("Producing message {Message} of Type {MessageType} on Path {Path} with Size {MessageSize}", envelope.Message, messageType?.Name, path, messagePayload?.Length ?? 0);`
	`121`	`+ _logger.LogDebug("Producing message {Message} of Type {MessageType} on Path {Path} with Size {MessageSize}", envelope.Message, messageType?.Name, path, messagePayload?.Length ?? 0);`
`123`	`122`
`124`		`- var ev = envelope.Message != null ? new EventData(messagePayload) : new EventData();`
	`123`	`+ var ev = envelope.Message != null ? new EventData(messagePayload) : new EventData();`
`125`	`124`
`126`		`- if (envelope.Headers != null)`
	`125`	`+ if (envelope.Headers != null)`
	`126`	`+ {`
	`127`	`+ foreach (var header in envelope.Headers)`
`127`	`128`	`{`
`128`		`- foreach (var header in envelope.Headers)`
`129`		`- {`
`130`		`- ev.Properties.Add(header.Key, header.Value);`
`131`		`- }`
	`129`	`+ ev.Properties.Add(header.Key, header.Value);`
`132`	`130`	`}`
	`131`	`+ }`
`133`	`132`
`134`		`- var partitionKey = messageType != null`
`135`		`- ? GetPartitionKey(messageType, envelope.Message)`
`136`		`- : null;`
	`133`	`+ var partitionKey = messageType != null`
	`134`	`+ ? GetPartitionKey(messageType, envelope.Message)`
	`135`	`+ : null;`
`137`	`136`
`138`		`- return (Envelope: envelope, Message: ev, PartitionKey: partitionKey);`
`139`		`- })`
`140`		`- .GroupBy(x => x.PartitionKey);`
	`137`	`+ return (Envelope: envelope, Message: ev, PartitionKey: partitionKey);`
	`138`	`+ })`
	`139`	`+ .GroupBy(x => x.PartitionKey);`
`141`	`140`
`142`	`141`	`var producer = _producerByPath[path];`
`143`		`- foreach (var partition in messages)`
	`142`	`+ foreach (var partition in messagesByPartition)`
`144`	`143`	`{`
`145`	`144`	`EventDataBatch batch = null;`
`146`	`145`	`try`
`@@ -184,7 +183,7 @@ protected override async Task OnStart()`
`184`	`183`	`batch = null;`
`185`	`184`	`}`
`186`	`185`
`187`		`- return (dispatched, null);`
	`186`	`+ return new(dispatched, null);`
`188`	`187`	`}`
`189`	`188`	`finally`
`190`	`189`	`{`
`@@ -194,10 +193,10 @@ protected override async Task OnStart()`
`194`	`193`	`}`
`195`	`194`	`catch (Exception ex)`
`196`	`195`	`{`
`197`		`- return (dispatched, ex);`
	`196`	`+ return new(dispatched, ex);`
`198`	`197`	`}`
`199`	`198`
`200`		`- return (dispatched, null);`
	`199`	`+ return new(dispatched, null);`
`201`	`200`	`}`
`202`	`201`
`203`	`202`	`#endregion`
Original file line number	Diff line number	Diff line change
`@@ -116,15 +116,14 @@ void AddConsumerFrom(TopicSubscriptionParams topicSubscription, IMessageProcesso`
`116`	`116`	`}`
`117`	`117`	`}`
`118`	`118`
`119`		`- protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)`
	`119`	`+ protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)`
`120`	`120`	`{`
`121`		`- async Task SendBatchAsync(ServiceBusSender senderClient, ServiceBusMessageBatch batch, CancellationToken cancellationToken)`
`122`		`- {`
`123`		`- await Retry.WithDelay(`
	`121`	`+ Task SendBatchAsync(ServiceBusSender senderClient, ServiceBusMessageBatch batch, CancellationToken cancellationToken) =>`
	`122`	`+ Retry.WithDelay(`
`124`	`123`	`async cancellationToken =>`
`125`	`124`	`{`
`126`	`125`	`await senderClient.SendMessagesAsync(batch, cancellationToken).ConfigureAwait(false);`
`127`		`- _logger.LogDebug("Batch of {BatchSize} message(s) dispatched to {path} ({SizeInBytes} bytes)", batch.Count, path, batch.SizeInBytes);`
	`126`	`+ _logger.LogDebug("Batch of {BatchSize} message(s) dispatched to {Path} ({SizeInBytes} bytes)", batch.Count, path, batch.SizeInBytes);`
`128`	`127`	`},`
`129`	`128`	`(exception, attempt) =>`
`130`	`129`	`{`
`@@ -135,18 +134,16 @@ await Retry.WithDelay(`
`135`	`134`	`_logger.LogWarning("Service bus throttled. Backing off (Attempt: {Attempt}).", attempt);`
`136`	`135`	`return true;`
`137`	`136`	`}`
`138`		`-`
`139`	`137`	`return false;`
`140`	`138`	`},`
`141`		`- TimeSpan.FromSeconds(2),`
`142`		`- TimeSpan.FromSeconds(1),`
	`139`	`+ delay: TimeSpan.FromSeconds(2),`
	`140`	`+ jitter: TimeSpan.FromSeconds(1),`
`143`	`141`	`cancellationToken);`
`144`		`- }`
`145`	`142`
`146`	`143`	`AssertActive();`
`147`	`144`
`148`		`- var messages = envelopes.Select(`
`149`		`- envelope =>`
	`145`	`+ var messages = envelopes`
	`146`	`+ .Select(envelope =>`
`150`	`147`	`{`
`151`	`148`	`var messageType = envelope.Message?.GetType();`
`152`	`149`	`var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);`
`@@ -190,12 +187,12 @@ await Retry.WithDelay(`
`190`	`187`	`await senderClient.SendMessageAsync(item.ServiceBusMessage, cancellationToken: cancellationToken).ConfigureAwait(false);`
`191`	`188`	`_logger.LogDebug("Delivered item {Message} of type {MessageType} to {Path}", item.Envelope.Message, item.Envelope.MessageType?.Name, path);`
`192`	`189`
`193`		`- return ([item.Envelope], null);`
	`190`	`+ return new([item.Envelope], null);`
`194`	`191`	`}`
`195`	`192`	`catch (Exception ex)`
`196`	`193`	`{`
`197`	`194`	`_logger.LogDebug(ex, "Producing message {Message} of type {MessageType} to path {Path} resulted in error {Error}", item.Envelope.Message, item.Envelope.MessageType?.Name, path, ex.Message);`
`198`		`- return ([], ex);`
	`195`	`+ return new([], ex);`
`199`	`196`	`}`
`200`	`197`	`}`
`201`	`198`
`@@ -233,12 +230,12 @@ await Retry.WithDelay(`
`233`	`230`	`batch = null;`
`234`	`231`	`}`
`235`	`232`
`236`		`- return (dispatched, null);`
	`233`	`+ return new(dispatched, null);`
`237`	`234`	`}`
`238`	`235`	`catch (Exception ex)`
`239`	`236`	`{`
`240`	`237`	`_logger.LogError(ex, "Producing message batch to path {Path} resulted in error {Error}", path, ex.Message);`
`241`		`- return (dispatched, ex);`
	`238`	`+ return new(dispatched, ex);`
`242`	`239`	`}`
`243`	`240`	`finally`
`244`	`241`	`{`
Original file line number	Diff line number	Diff line change
`@@ -114,7 +114,7 @@ protected override async ValueTask DisposeAsyncCore()`
`114`	`114`	`}`
`115`	`115`	`}`
`116`	`116`
`117`		`- protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)`
	`117`	`+ protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)`
`118`	`118`	`{`
`119`	`119`	`AssertActive();`
`120`	`120`
`@@ -172,10 +172,10 @@ protected override async ValueTask DisposeAsyncCore()`
`172`	`172`	`}`
`173`	`173`	`catch (Exception ex)`
`174`	`174`	`{`
`175`		`- return (dispatched, ex);`
	`175`	`+ return new(dispatched, ex);`
`176`	`176`	`}`
`177`	`177`
`178`		`- return (dispatched, null);`
	`178`	`+ return new(dispatched, null);`
`179`	`179`	`}`
`180`	`180`
`181`	`181`	`protected byte[] GetMessageKey(ProducerSettings producerSettings, Type messageType, object message, string topic)`
Original file line number	Diff line number	Diff line change
`@@ -123,8 +123,8 @@ private IMessageProcessorQueue CreateMessageProcessorQueue(IMessageProcessor<obj`
`123`	`123`	`: new MessageProcessorQueue(messageProcessor, LoggerFactory.CreateLogger<MessageProcessorQueue>(), CancellationToken);`
`124`	`124`	`}`
`125`	`125`
`126`		`- protected override Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)`
`127`		`- => Task.FromResult<(IReadOnlyCollection<T> Dispatched, Exception Exception)>(([], null)); // Not used`
	`126`	`+ protected override Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)`
	`127`	`+ => Task.FromResult<ProduceToTransportBulkResult<T>>(new([], null)); // Not used`
`128`	`128`
`129`	`129`	`public override Task ProduceResponse(string requestId, object request, IReadOnlyDictionary<string, object> requestHeaders, object response, Exception responseException, IMessageTypeConsumerInvokerSettings consumerInvoker)`
`130`	`130`	`=> Task.CompletedTask; // Not used to responses`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,5 @@`
`1`	`1`	`namespace SlimMessageBus.Host.Mqtt;`
`2`	`2`
`3`		`-using System.Security.Cryptography;`
`4`		`-`
`5`	`3`	`using MQTTnet.Extensions.ManagedClient;`
`6`	`4`
`7`	`5`	`public class MqttMessageBus : MessageBusBase<MqttMessageBusSettings>`
`@@ -103,10 +101,10 @@ private Task OnMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)`
`103`	`101`	`return Task.CompletedTask;`
`104`	`102`	`}`
`105`	`103`
`106`		`- protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)`
	`104`	`+ protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)`
`107`	`105`	`{`
`108`		`- var messages = envelopes.Select(`
`109`		`- envelope =>`
	`106`	`+ var messages = envelopes`
	`107`	`+ .Select(envelope =>`
`110`	`108`	`{`
`111`	`109`	`var messagePayload = Serializer.Serialize(envelope.MessageType, envelope.Message);`
`112`	`110`
`@@ -156,10 +154,10 @@ private Task OnMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)`
`156`	`154`	`}`
`157`	`155`	`catch (Exception ex)`
`158`	`156`	`{`
`159`		`- return (dispatched, ex);`
	`157`	`+ return new(dispatched, ex);`
`160`	`158`	`}`
`161`	`159`	`}`
`162`	`160`
`163`		`- return (dispatched, null);`
	`161`	`+ return new(dispatched, null);`
`164`	`162`	`}`
`165`	`163`	`}`
Original file line number	Diff line number	Diff line change
`@@ -252,7 +252,7 @@ internal async Task<int> SendMessages(IServiceProvider serviceProvider, IOutboxR`
`252`	`252`	`outboxMessage =>`
`253`	`253`	`{`
`254`	`254`	`var message = bus.Serializer.Deserialize(outboxMessage.MessageType, outboxMessage.MessagePayload);`
`255`		`- return new EnvelopeWithId(outboxMessage.Id, message, outboxMessage.MessageType, outboxMessage.Headers ?? new Dictionary<string, object>());`
	`255`	`+ return new OutboxBulkMessage(outboxMessage.Id, message, outboxMessage.MessageType, outboxMessage.Headers ?? new Dictionary<string, object>());`
`256`	`256`	`})`
`257`	`257`	`.Batch(bulkProducer.MaxMessagesPerTransaction ?? defaultBatchSize);`
`258`	`258`
`@@ -271,12 +271,12 @@ internal async Task<int> SendMessages(IServiceProvider serviceProvider, IOutboxR`
`271`	`271`	`return count;`
`272`	`272`	`}`
`273`	`273`
`274`		`- internal async Task<(bool Success, int Published)> DispatchBatchAsync(IOutboxRepository outboxRepository, IMessageBusBulkProducer producer, IMessageBusTarget messageBusTarget, IReadOnlyCollection<EnvelopeWithId> batch, string busName, string path, CancellationToken cancellationToken)`
	`274`	`+ internal async Task<(bool Success, int Published)> DispatchBatchAsync(IOutboxRepository outboxRepository, IMessageBusBulkProducer producer, IMessageBusTarget messageBusTarget, IReadOnlyCollection<OutboxBulkMessage> batch, string busName, string path, CancellationToken cancellationToken)`
`275`	`275`	`{`
`276`	`276`	`_logger.LogDebug("Publishing batch of {MessageCount} messages to pathGroup {Path} on {BusName} bus", batch.Count, path, busName);`
`277`	`277`
`278`	`278`	`// TOOD: Enclose in a transaction`
`279`		`- var results = await producer.ProduceToTransport(batch, path, messageBusTarget, cancellationToken).ConfigureAwait(false);`
	`279`	`+ var results = await producer.ProduceToTransportBulk(batch, path, messageBusTarget, cancellationToken).ConfigureAwait(false);`
`280`	`280`	`if (cancellationToken.IsCancellationRequested && results.Dispatched.Count == 0)`
`281`	`281`	`{`
`282`	`282`	`// if cancellation has been requested, only return if no messages were published`
`@@ -311,14 +311,14 @@ private static IMasterMessageBus GetBus(ICompositeMessageBus compositeMessageBus`
`311`	`311`	`return null;`
`312`	`312`	`}`
`313`	`313`
`314`		`- public record EnvelopeWithId : Envelope`
	`314`	`+ public record OutboxBulkMessage : BulkMessageEnvelope`
`315`	`315`	`{`
`316`		`- public EnvelopeWithId(Guid id, object Message, Type MessageType, IDictionary<string, object> Headers)`
`317`		`- : base(Message, MessageType, Headers)`
	`316`	`+ public Guid Id { get; }`
	`317`	`+`
	`318`	`+ public OutboxBulkMessage(Guid id, object message, Type messageType, IDictionary<string, object> headers)`
	`319`	`+ : base(message, messageType, headers)`
`318`	`320`	`{`
`319`	`321`	`Id = id;`
`320`	`322`	`}`
`321`		`-`
`322`		`- public Guid Id { get; }`
`323`	`323`	`}`
`324`	`324`	`}`
Original file line number	Diff line number	Diff line change
`@@ -128,7 +128,7 @@ protected override async ValueTask DisposeAsyncCore()`
`128`	`128`	`}`
`129`	`129`	`}`
`130`	`130`
`131`		`- protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)`
	`131`	`+ protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)`
`132`	`132`	`{`
`133`	`133`	`await EnsureInitFinished();`
`134`	`134`
`@@ -178,9 +178,9 @@ protected override async ValueTask DisposeAsyncCore()`
`178`	`178`	`}`
`179`	`179`	`catch (Exception ex)`
`180`	`180`	`{`
`181`		`- return (dispatched, ex);`
	`181`	`+ return new(dispatched, ex);`
`182`	`182`	`}`
`183`	`183`
`184`		`- return (dispatched, null);`
	`184`	`+ return new(dispatched, null);`
`185`	`185`	`}`
`186`	`186`	`}`
Original file line number	Diff line number	Diff line change
`@@ -147,9 +147,13 @@ void AddTopicConsumer(string topic, ISubscriber subscriber, IMessageProcessor<Me`
`147`	`147`
`148`	`148`	`#region Overrides of MessageBusBase`
`149`	`149`
`150`		`- protected override async Task<(IReadOnlyCollection<T> Dispatched, Exception Exception)> ProduceToTransport<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken = default)`
	`150`	`+ protected override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)`
`151`	`151`	`{`
	`152`	`+#if NETSTANDARD2_0`
`152`	`153`	`if (envelopes is null) throw new ArgumentNullException(nameof(envelopes));`
	`154`	`+#else`
	`155`	`+ ArgumentNullException.ThrowIfNull(envelopes);`
	`156`	`+#endif`
`153`	`157`
`154`	`158`	`AssertActive();`
`155`	`159`
`@@ -184,10 +188,10 @@ void AddTopicConsumer(string topic, ISubscriber subscriber, IMessageProcessor<Me`
`184`	`188`	`}`
`185`	`189`	`catch (Exception ex)`
`186`	`190`	`{`
`187`		`- return (dispatched, ex);`
	`191`	`+ return new(dispatched, ex);`
`188`	`192`	`}`
`189`	`193`
`190`		`- return (dispatched, null);`
	`194`	`+ return new(dispatched, null);`
`191`	`195`	`}`
`192`	`196`
`193`	`197`	`#endregion`