Refactor message producers

* Naming change

* Refactor logic of sending messages, to simplify and apply formatting for each sending message

* Update readme.md

Author: dvlaskin

readme.md

@@ -14,8 +14,8 @@
 * Override GUID, DateTime values in a message before sending
 * Ability to compress and encode a message before sending
 * Send a **single message**
-* Send a **batch of identical messages**
-* Send a **batch of identical messages** with a **time delay** between each message
+* Send a **batch of messages**
+* Send a **batch of messages** with a **time delay** between each message
 
 ### Receiving Messages from Event Hubs
 

src/Application/Services/EventHubConsumerService.cs

@@ -39,7 +39,7 @@ public async IAsyncEnumerable<EventHubMessage> StartReceiveMessageAsync(
 
         await foreach (var message in channel.Reader.ReadAllAsync(cancellationToken))
         {
-            logger.LogInformation("Received message {MsgData}", message);
+            logger.LogInformation("Received message {MsgData}", message.Message);
             message.Message = textProcessingPipeline.Process(message.Message);
 
             yield return message;

src/Application/Services/MessageProducers/BaseMessageProducer.cs

@@ -0,0 +1,84 @@
+using Domain.Interfaces.Providers;
+using Domain.Interfaces.Services;
+using Domain.Models;
+
+namespace Application.Services.MessageProducers;
+
+public abstract class BaseMessageProducer<T> : IMessageProducerService
+{
+    private readonly IMessageProducerProvider messageProducerProvider;
+    private readonly MessageOptions? messageOptions;
+    private bool disposed;
+
+
+    protected BaseMessageProducer(
+        IMessageProducerProvider messageProducerProvider, 
+        MessageOptions? messageOptions = null
+    )
+    {
+        this.messageProducerProvider = messageProducerProvider;
+        this.messageOptions = messageOptions;
+    }
+    
+
+    public async Task SendMessagesAsync(
+        string? messageText, uint numberOfMessages = 1, TimeSpan? delayToSend = null, CancellationToken cancellationToken = default
+    )
+    {
+        if (string.IsNullOrWhiteSpace(messageText))
+            return;
+
+        var messageModifier = CreateMessageModifier();
+        
+        if (numberOfMessages <= 1)
+        {
+            await messageProducerProvider
+                .SendMessageAsync(messageText, messageModifier, cancellationToken)
+                .ConfigureAwait(false);
+        }
+        else if (delayToSend is null || delayToSend.Value <= TimeSpan.Zero)
+        {
+            await messageProducerProvider
+                .SendMessagesAsync(messageText, messageModifier, numberOfMessages, cancellationToken)
+                .ConfigureAwait(false);
+        }
+        else
+        {
+            await messageProducerProvider.SendMessagesWithDelayAsync(
+                messageText, messageModifier, numberOfMessages, delayToSend.Value, cancellationToken
+            ).ConfigureAwait(false);
+        }
+    }
+
+    
+    private Func<string, BinaryData> CreateMessageModifier()
+    {
+        return messageInput =>
+        {
+            var formatterMessage = ApplyFormattingOptions(messageInput);
+            var messageToSend = ApplyEncodingOptions(formatterMessage);
+            return EncodeToBinaryData(messageToSend);
+        };
+    }
+
+    protected virtual string ApplyFormattingOptions(string messageText)
+    {
+        return messageOptions?.TextProcessingPipeline?.Process(messageText) ?? messageText;
+    }
+    
+    protected abstract T ApplyEncodingOptions(string message);
+    protected abstract BinaryData EncodeToBinaryData(T message);
+    
+    
+    public async ValueTask DisposeAsync()
+    {
+        if (disposed)
+            return;
+        
+        await messageProducerProvider.DisposeAsync().ConfigureAwait(false);
+        GC.SuppressFinalize(this);
+        disposed = true;
+    }
+    
+    ~BaseMessageProducer() => DisposeAsync().AsTask().GetAwaiter().GetResult();
+}

src/Application/Services/MessageProducers/BytesMessageProducer.cs

@@ -1,22 +1,18 @@
 using Application.Utils;
 using Domain.Interfaces.Providers;
 using Domain.Models;
-using Microsoft.Extensions.Logging;
 
 namespace Application.Services.MessageProducers;
 
-public class BytesMessageProducer : MessageProducerBase<byte[]>
+public class BytesMessageProducer : BaseMessageProducer<byte[]>
 {
-    private readonly ILogger<BytesMessageProducer> logger;
     private readonly MessageOptions? messageOptions;
 
     public BytesMessageProducer(
-        ILogger<BytesMessageProducer> logger,
         IMessageProducerProvider messageProducerProvider,
         MessageOptions? messageOptions = null
     ) : base(messageProducerProvider, messageOptions)
     {
-        this.logger = logger;
         this.messageOptions = messageOptions;
     }
 
@@ -28,31 +24,8 @@ protected override byte[] ApplyEncodingOptions(string message)
         return message.Compress();
     }
 
-    protected override async Task SendSingleMessageAsync(byte[] message, CancellationToken cancellationToken)
+    protected override BinaryData EncodeToBinaryData(byte[] message)
     {
-        await MessageProducerProvider.SendMessageAsync(message, cancellationToken);
-    }
-
-    protected override async Task SendBatchMessagesAsync(
-        byte[] message, int numberOfMessages, CancellationToken cancellationToken
-    )
-    {
-        var messages = MultipleMessages(message, numberOfMessages);
-        await MessageProducerProvider.SendMessagesAsync(messages, cancellationToken);
-    }
-
-    protected override async Task SendMessagesWithDelayAsync(
-        byte[] message, int numberOfMessages, TimeSpan timeDelay, CancellationToken cancellationToken
-    )
-    {
-        try
-        {
-            var messages = MultipleMessages(message, numberOfMessages);
-            await MessageProducerProvider.SendMessagesAsync(messages, timeDelay, cancellationToken);
-        }
-        catch (OperationCanceledException)
-        {
-            logger.LogWarning("Task SendMessagesWithDelayAsync was canceled");
-        }
+        return BinaryData.FromBytes(message);
     }
 }

src/Application/Services/MessageProducers/MessageProducerBase.cs

@@ -1,22 +1,18 @@
 using Application.Utils;
 using Domain.Interfaces.Providers;
 using Domain.Models;
-using Microsoft.Extensions.Logging;
 
 namespace Application.Services.MessageProducers;
 
-public class StringMessageProducer : MessageProducerBase<string>
+public class StringMessageProducer : BaseMessageProducer<string>
 {
-    private readonly ILogger<StringMessageProducer> logger;
     private readonly MessageOptions? messageOptions;
 
     public StringMessageProducer(
-        ILogger<StringMessageProducer> logger,
         IMessageProducerProvider messageProducerProvider,
         MessageOptions? messageOptions = null
     ) : base(messageProducerProvider, messageOptions)
     {
-        this.logger = logger;
         this.messageOptions = messageOptions;
     }
 
@@ -28,31 +24,8 @@ protected override string ApplyEncodingOptions(string message)
         return message.Compress().EncodeBase64();
     }
 
-    protected override async Task SendSingleMessageAsync(string message, CancellationToken cancellationToken)
+    protected override BinaryData EncodeToBinaryData(string message)
     {
-        await MessageProducerProvider.SendMessageAsync(message, cancellationToken);
-    }
-
-    protected override async Task SendBatchMessagesAsync(
-        string message, int numberOfMessages, CancellationToken cancellationToken
-    )
-    {
-        var messages = MultipleMessages(message, numberOfMessages);
-        await MessageProducerProvider.SendMessagesAsync(messages, cancellationToken);
-    }
-
-    protected override async Task SendMessagesWithDelayAsync(
-        string message, int numberOfMessages, TimeSpan timeDelay, CancellationToken cancellationToken
-    )
-    {
-        try
-        {
-            var messages = MultipleMessages(message, numberOfMessages);
-            await MessageProducerProvider.SendMessagesAsync(messages, timeDelay, cancellationToken);
-        }
-        catch (OperationCanceledException)
-        {
-            logger.LogWarning("Task SendMessagesWithDelayAsync was canceled");
-        }
+        return BinaryData.FromString(message);
     }
 }

src/Application/Services/MessageProducers/StringMessageProducer.cs

@@ -6,4 +6,8 @@
         <Nullable>enable</Nullable>
     </PropertyGroup>
 
+    <ItemGroup>
+      <PackageReference Include="System.Memory.Data" Version="6.0.1" />
+    </ItemGroup>
+
 </Project>

src/Domain/Domain.csproj

@@ -2,12 +2,22 @@ namespace Domain.Interfaces.Providers;
 
 public interface IMessageProducerProvider : IAsyncDisposable
 {
-    Task SendMessageAsync(string message, CancellationToken cancellationToken);
-    Task SendMessageAsync(byte[] message, CancellationToken cancellationToken);
-    
-    Task SendMessagesAsync(IReadOnlyList<string> messages, CancellationToken cancellationToken);
-    Task SendMessagesAsync(IReadOnlyList<byte[]> messages, CancellationToken cancellationToken);
-    
-    Task SendMessagesAsync(IReadOnlyList<string> messages, TimeSpan timeDelay, CancellationToken cancellationToken);
-    Task SendMessagesAsync(IReadOnlyList<byte[]> messages, TimeSpan timeDelay, CancellationToken cancellationToken);
+    Task SendMessageAsync(
+        string message, Func<string, BinaryData>? messageModifier = null, CancellationToken cancellationToken = default
+    );
+
+    Task SendMessagesAsync(
+        string message,
+        Func<string, BinaryData>? messageModifier = null,
+        uint numberOfMessages = 1,
+        CancellationToken cancellationToken = default
+    );
+
+    Task SendMessagesWithDelayAsync(
+        string message,
+        Func<string, BinaryData>? messageModifier = null,
+        uint numberOfMessages = 1,
+        TimeSpan sendDelay = default,
+        CancellationToken cancellationToken = default
+    );
 }

src/Domain/Interfaces/Providers/IMessageProducerProvider.cs

@@ -3,6 +3,6 @@ namespace Domain.Interfaces.Services;
 public interface IMessageProducerService : IAsyncDisposable
 {
     Task SendMessagesAsync(
-        string? messageText, int numberOfMessages = 1, TimeSpan? delayToSend = null, CancellationToken cancellationToken = default
+        string? messageText, uint numberOfMessages = 1, TimeSpan? delayToSend = null, CancellationToken cancellationToken = default
     );
 }