Fix #43; async overload on backgroundservice

Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs

@@ -126,6 +126,55 @@ public async Task VerifyBackgroundServiceBehaviourPushMultiple()
             }
         }
 
+        [Test]
+        public async Task VerifyPushMessageAsync()
+        {
+            using var cancellationTokenSource = new CancellationTokenSource();
+            _ = this.backgroundService.StartAsync(cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
+            var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync");
+
+            await this.backgroundService.PushMessageAsync("ABC", exchange, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
+            var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<TestMessagingBackgroundService>>(), this.configurationMock.Object);
+
+            await Assert.ThatAsync(() => faultyService.PushMessageAsync("XYZ", exchange, cancellationToken: cancellationTokenSource.Token), Throws.Exception);
+            faultyService.Dispose();
+
+            await cancellationTokenSource.CancelAsync();
+
+            Assert.That(this.backgroundService.ReceivedMessages, Does.Contain("ABC"));
+        }
+
+        [Test]
+        public async Task VerifyPushMessagesAsync()
+        {
+            using var cancellationTokenSource = new CancellationTokenSource();
+            _ = this.backgroundService.StartAsync(cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
+            var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync");
+            string[] messages = ["ABC", "DEF", "GHI"];
+
+            await this.backgroundService.PushMessagesAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(800), CancellationToken.None);
+
+            var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<TestMessagingBackgroundService>>(), this.configurationMock.Object);
+
+            await Assert.ThatAsync(() => faultyService.PushMessagesAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token), Throws.Exception);
+            faultyService.Dispose();
+
+            await cancellationTokenSource.CancelAsync();
+
+            using (Assert.EnterMultipleScope())
+            {
+                Assert.That(this.backgroundService.ReceivedMessages, Has.Count.EqualTo(3));
+                Assert.That(this.backgroundService.ReceivedMessages, Is.EquivalentTo(messages));
+            }
+        }
+
         [Test]
         public async Task VerifyBackgroundServiceBehaviourTaskCanceled()
         {
@@ -335,8 +384,9 @@ protected override async Task InitializeAsync()
             {
                 this.ConnectionName = ConfiguredConnectionName;
                 await this.RegisterListener(() => this.MessageClientService.ListenAsync<string>(this.ConnectionName, new FanoutExchangeConfiguration("BackgroundTest")), this.ReceivedMessages.Add, onError: _ => this.ReceivedMessages.Clear());
-
-                await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync<int>(this.ConnectionName, new DirectExchangeConfiguration("BackgroundTestInt")), (x) => 
+                await this.RegisterListener(() => this.MessageClientService.ListenAsync<string>(this.ConnectionName, new FanoutExchangeConfiguration("BackgroundTestPushAsync")), this.ReceivedMessages.Add, onError: _ => this.ReceivedMessages.Clear());
+
+                await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync<int>(this.ConnectionName, new DirectExchangeConfiguration("BackgroundTestInt")), (x) =>
                 {
                     this.ReceivedMessages.Add(x.ToString());
                     return Task.CompletedTask;

Mercurio.Hosting/IMessagingBackgroundService.cs

@@ -68,5 +68,39 @@ public interface IMessagingBackgroundService
         /// <see cref="BasicProperties.ContentType" /> as 'application/json"
         /// </remarks>
         void PushMessages<TMessage>(IEnumerable<TMessage> messages, IExchangeConfiguration exchangeConfiguration, Action<BasicProperties> configureProperties = null, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Asynchronously pushes the specified <paramref name="message" /> to the specified queue via the
+        /// <paramref name="exchangeConfiguration" />, bypassing the internal background queue so the caller can
+        /// await completion and observe any publishing errors.
+        /// </summary>
+        /// <typeparam name="TMessage">The type of message</typeparam>
+        /// <param name="message">The <typeparamref name="TMessage" /> to push</param>
+        /// <param name="exchangeConfiguration">The <see cref="IExchangeConfiguration" /> that should be used to configure the queue and exchange to use</param>
+        /// <param name="configureProperties">Possible action to configure additional properties</param>
+        /// <param name="cancellationToken">An optional <see cref="CancellationToken" /></param>
+        /// <returns>An awaitable <see cref="Task" /> that completes when the message has been published</returns>
+        /// <remarks>
+        /// By default, the <see cref="BasicProperties" /> is configured to use the <see cref="DeliveryModes.Persistent" /> mode and sets the
+        /// <see cref="BasicProperties.ContentType" /> as 'application/json"
+        /// </remarks>
+        Task PushMessageAsync<TMessage>(TMessage message, IExchangeConfiguration exchangeConfiguration, Action<BasicProperties> configureProperties = null, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Asynchronously pushes the specified <paramref name="messages" /> to the specified queue via the
+        /// <paramref name="exchangeConfiguration" />, bypassing the internal background queue so the caller can
+        /// await completion and observe any publishing errors.
+        /// </summary>
+        /// <typeparam name="TMessage">The type of message</typeparam>
+        /// <param name="messages">The collection of <typeparamref name="TMessage" /> to push</param>
+        /// <param name="exchangeConfiguration">The <see cref="IExchangeConfiguration" /> that should be used to configure the queue and exchange to use</param>
+        /// <param name="configureProperties">Possible action to configure additional properties</param>
+        /// <param name="cancellationToken">An optional <see cref="CancellationToken" /></param>
+        /// <returns>An awaitable <see cref="Task" /> that completes when the messages have been published</returns>
+        /// <remarks>
+        /// By default, the <see cref="BasicProperties" /> is configured to use the <see cref="DeliveryModes.Persistent" /> mode and sets the
+        /// <see cref="BasicProperties.ContentType" /> as 'application/json"
+        /// </remarks>
+        Task PushMessagesAsync<TMessage>(IEnumerable<TMessage> messages, IExchangeConfiguration exchangeConfiguration, Action<BasicProperties> configureProperties = null, CancellationToken cancellationToken = default);
     }
 }

Mercurio.Hosting/Mercurio.Hosting.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>disable</Nullable>
         <TargetFrameworks>net10.0;net9.0;net8.0;netstandard2.1</TargetFrameworks>
-        <Version>2.0.4</Version>
+        <Version>2.0.5</Version>
         <LangVersion>12.0</LangVersion>
         <Company>Starion Group S.A.</Company>
         <Copyright>Copyright © Starion Group S.A.</Copyright>
@@ -28,9 +28,7 @@
         <GenerateSBOM>true</GenerateSBOM>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
         <PackageReleaseNotes>
-            [Update] Microsoft.Extensions.Diagnostics.HealthChecks.Abstractions to version 10.0.7
-            [Update] Microsoft.Extensions.Hosting to version 10.0.7
-            [Update] Mercurio to version 2.0.3
+            [Add] PushMessageAsync and PushMessagesAsync overloads on IMessagingBackgroundService / MessagingBackgroundService to allow awaiting the publish and observing errors
         </PackageReleaseNotes>
     </PropertyGroup>
 

Mercurio.Hosting/MessagingBackgroundService.cs

@@ -162,6 +162,52 @@ public void PushMessages<TMessage>(IEnumerable<TMessage> messages, IExchangeConf
             this.messagesToPush.Enqueue((() => this.MessageClientService.PushAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken), activityContext, activityName));
         }
 
+        /// <summary>
+        /// Asynchronously pushes the specified <paramref name="message" /> to the specified queue via the
+        /// <paramref name="exchangeConfiguration" />, bypassing the internal background queue so the caller can
+        /// await completion and observe any publishing errors.
+        /// </summary>
+        /// <typeparam name="TMessage">The type of message</typeparam>
+        /// <param name="message">The <typeparamref name="TMessage" /> to push</param>
+        /// <param name="exchangeConfiguration">The <see cref="IExchangeConfiguration" /> that should be used to configure the queue and exchange to use</param>
+        /// <param name="configureProperties">Possible action to configure additional properties</param>
+        /// <param name="cancellationToken">An optional <see cref="CancellationToken" /></param>
+        /// <returns>An awaitable <see cref="Task" /> that completes when the message has been published</returns>
+        /// <remarks>
+        /// By default, the <see cref="BasicProperties" /> is configured to use the <see cref="DeliveryModes.Persistent" /> mode and sets the
+        /// <see cref="BasicProperties.ContentType" /> as 'application/json"
+        /// </remarks>
+        public async Task PushMessageAsync<TMessage>(TMessage message, IExchangeConfiguration exchangeConfiguration, Action<BasicProperties> configureProperties = null, CancellationToken cancellationToken = default)
+        {
+            var activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName);
+            var activityName = $"Background Push {typeof(TMessage).Name} [{exchangeConfiguration}]";
+            using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default);
+            await this.MessageClientService.PushAsync(this.ConnectionName, message, exchangeConfiguration, configureProperties, cancellationToken);
+        }
+
+        /// <summary>
+        /// Asynchronously pushes the specified <paramref name="messages" /> to the specified queue via the
+        /// <paramref name="exchangeConfiguration" />, bypassing the internal background queue so the caller can
+        /// await completion and observe any publishing errors.
+        /// </summary>
+        /// <typeparam name="TMessage">The type of message</typeparam>
+        /// <param name="messages">The collection of <typeparamref name="TMessage" /> to push</param>
+        /// <param name="exchangeConfiguration">The <see cref="IExchangeConfiguration" /> that should be used to configure the queue and exchange to use</param>
+        /// <param name="configureProperties">Possible action to configure additional properties</param>
+        /// <param name="cancellationToken">An optional <see cref="CancellationToken" /></param>
+        /// <returns>An awaitable <see cref="Task" /> that completes when the messages have been published</returns>
+        /// <remarks>
+        /// By default, the <see cref="BasicProperties" /> is configured to use the <see cref="DeliveryModes.Persistent" /> mode and sets the
+        /// <see cref="BasicProperties.ContentType" /> as 'application/json"
+        /// </remarks>
+        public async Task PushMessagesAsync<TMessage>(IEnumerable<TMessage> messages, IExchangeConfiguration exchangeConfiguration, Action<BasicProperties> configureProperties = null, CancellationToken cancellationToken = default)
+        {
+            var activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName);
+            var activityName = $"Background Push Multiple {typeof(TMessage).Name} [{exchangeConfiguration}]";
+            using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default);
+            await this.MessageClientService.PushAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken);
+        }
+
         /// <summary>
         /// Triggered when the application host is ready to start the service.
         /// </summary>

Mercurio/Mercurio.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>disable</Nullable>
         <TargetFrameworks>net10.0;net9.0;net8.0;netstandard2.1</TargetFrameworks>
-        <Version>2.0.3</Version>
+        <Version>2.0.5</Version>
         <LangVersion>12.0</LangVersion>
         <Company>Starion Group S.A.</Company>
         <Copyright>Copyright © Starion Group S.A.</Copyright>
@@ -28,12 +28,7 @@
         <GenerateSBOM>true</GenerateSBOM>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
         <PackageReleaseNotes>
-            [Fix] Don't allow transient_nonexcl_queues. Deprecated since RabbitMQ version 4.x
-            [Update] CommunityToolkit.HighPerformance to version 8.4.2
-            [Update] Microsoft.Extensions.DependencyInjection.Abstractions to version 10.0.7
-            [Update] Polly to version 8.6.6
-            [Update] RabbitMQ.Client to version 7.2.1
-            [Update] System.Text.Json to version 10.0.7
+            [Bump] For version align
         </PackageReleaseNotes>
     </PropertyGroup>