Fix #47: publish with ACK

Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs

@@ -115,7 +115,10 @@ public async Task VerifyBackgroundServiceBehaviourPushMultiple()
         {
             using var cancellationTokenSource = new CancellationTokenSource();
             _ = this.backgroundService.StartAsync(cancellationTokenSource.Token);
-            await Task.Delay(TimeSpan.FromMilliseconds(100), CancellationToken.None);
+
+            // a fanout message published before the listener has bound its queue is silently discarded, so the subscriptions have to be
+            // established before pushing anything
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
 
             string[] messages = ["ABC", "DEF", "GHI"];
 
@@ -153,6 +156,67 @@ public async Task VerifyPushMessageAsync()
             Assert.That(this.backgroundService.ReceivedMessages, Does.Contain("ABC"));
         }
 
+        [Test]
+        public async Task VerifyPushMessageWithConfirmationAsync()
+        {
+            using var cancellationTokenSource = new CancellationTokenSource();
+            _ = this.backgroundService.StartAsync(cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
+            var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync");
+
+            var acknowledged = await this.backgroundService.PushMessageWithConfirmationAsync("ABC", exchange, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
+            using (Assert.EnterMultipleScope())
+            {
+                Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged the message");
+                Assert.That(this.backgroundService.ReceivedMessages, Does.Contain("ABC"));
+            }
+
+            var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<TestMessagingBackgroundService>>(), this.configurationMock.Object);
+            var notRegistered = await faultyService.PushMessageWithConfirmationAsync("XYZ", exchange, cancellationToken: cancellationTokenSource.Token);
+            faultyService.Dispose();
+
+            Assert.That(notRegistered, Is.False, "a service that has not been initialized has no registered connection, which is reported instead of being thrown");
+
+            await Assert.ThatAsync(() => this.backgroundService.PushMessageWithConfirmationAsync((string)null!, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentNullException, "an invalid argument remains an exception");
+
+            await cancellationTokenSource.CancelAsync();
+        }
+
+        [Test]
+        public async Task VerifyPushMessagesWithConfirmationAsync()
+        {
+            using var cancellationTokenSource = new CancellationTokenSource();
+            _ = this.backgroundService.StartAsync(cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
+            var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync");
+
+            // the cast is required: an array also binds to the single message overload, with TMessage being the array itself
+            IEnumerable<string> messages = ["ABC", "DEF", "GHI"];
+
+            var acknowledged = await this.backgroundService.PushMessagesWithConfirmationAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(800), CancellationToken.None);
+
+            using (Assert.EnterMultipleScope())
+            {
+                Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged every message");
+                Assert.That(this.backgroundService.ReceivedMessages, Is.EquivalentTo(messages));
+            }
+
+            var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<TestMessagingBackgroundService>>(), this.configurationMock.Object);
+            var notRegistered = await faultyService.PushMessagesWithConfirmationAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token);
+            faultyService.Dispose();
+
+            Assert.That(notRegistered, Is.False, "a service that has not been initialized has no registered connection, which is reported instead of being thrown");
+
+            await Assert.ThatAsync(() => this.backgroundService.PushMessagesWithConfirmationAsync((IEnumerable<string>)null!, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentException, "an invalid argument remains an exception");
+
+            await cancellationTokenSource.CancelAsync();
+        }
+
         [Test]
         public async Task VerifyPushMessagesAsync()
         {
@@ -302,8 +366,11 @@ public async Task VerifyAutoRecoverySystem()
             using var cancellationTokenSource = new CancellationTokenSource();
             var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<AutoRecoveryTestMessagingBackgroundService>>(), this.configurationMock.Object);
             _ = autoRecoveryBackground.StartAsync(cancellationTokenSource.Token);
-            await Task.Delay(TimeSpan.FromMilliseconds(100), CancellationToken.None);
-
+
+            // a fanout message published before the listener has bound its queue is silently discarded, so the subscriptions have to be
+            // established before pushing anything
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
             var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTest");
             await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token);
             await Task.Delay(100, cancellationTokenSource.Token);
@@ -327,8 +394,11 @@ public async Task VerifyAsyncAutoRecoverySystem()
             using var cancellationTokenSource = new CancellationTokenSource();
             var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<AutoRecoveryTestMessagingBackgroundService>>(), this.configurationMock.Object);
             _ = autoRecoveryBackground.StartAsync(cancellationTokenSource.Token);
-            await Task.Delay(TimeSpan.FromMilliseconds(200), CancellationToken.None);
-
+
+            // a fanout message published before the listener has bound its queue is silently discarded, so the subscriptions have to be
+            // established before pushing anything
+            await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None);
+
             var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTestAsync");
             await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token);
             await Task.Delay(100, cancellationTokenSource.Token);

Mercurio.Hosting/IMessagingBackgroundService.cs

@@ -102,5 +102,50 @@ public interface IMessagingBackgroundService
         /// <see cref="BasicProperties.ContentType" /> as 'application/json"
         /// </remarks>
         Task PushMessagesAsync<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" /> and waits for the RabbitMQ server to acknowledge that it has taken
+        /// responsibility for the message
+        /// </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{TResult}" /> that provides true once the RabbitMQ server has acknowledged the message, false
+        /// if it has not
+        /// </returns>
+        /// <exception cref="ArgumentNullException">When the provided <typeparamref name="TMessage" /> or <paramref name="exchangeConfiguration" /> is null</exception>
+        /// <remarks>
+        /// This bypasses the internal background queue, so that the caller can await the acknowledgment. Invalid arguments still
+        /// throw, any operational failure is logged and returns false. The acknowledgment only asserts that the server took
+        /// responsibility for the message, not that any consumer received it.
+        /// </remarks>
+        Task<bool> PushMessageWithConfirmationAsync<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" /> and waits for the RabbitMQ server to acknowledge that it has taken
+        /// responsibility for each message
+        /// </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{TResult}" /> that provides true once the RabbitMQ server has acknowledged all the messages,
+        /// false as soon as one of them is not acknowledged
+        /// </returns>
+        /// <exception cref="ArgumentException">When the provided <paramref name="messages" /> collection is null</exception>
+        /// <exception cref="ArgumentNullException">When the provided <paramref name="exchangeConfiguration" /> is null</exception>
+        /// <remarks>
+        /// This bypasses the internal background queue, so that the caller can await the acknowledgment. The messages are published
+        /// one by one and the process stops at the first message that is not acknowledged. Since there is no transaction involved,
+        /// the messages that have been acknowledged before the failure are already held by the server.
+        /// </remarks>
+        Task<bool> PushMessagesWithConfirmationAsync<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.6</Version>
+        <Version>2.1.0</Version>
         <LangVersion>12.0</LangVersion>
         <Company>Starion Group S.A.</Company>
         <Copyright>Copyright © Starion Group S.A.</Copyright>
@@ -28,7 +28,7 @@
         <GenerateSBOM>true</GenerateSBOM>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
         <PackageReleaseNotes>
-            [Fix] Silent deserialization failure fix
+            [ADD] Provides Push ACK capability
         </PackageReleaseNotes>
     </PropertyGroup>
 

Mercurio.Hosting/MessagingBackgroundService.cs

@@ -208,6 +208,102 @@ public async Task PushMessagesAsync<TMessage>(IEnumerable<TMessage> messages, IE
             await this.MessageClientService.PushAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken);
         }
 
+        /// <summary>
+        /// Asynchronously pushes the specified <paramref name="message" /> to the specified queue via the
+        /// <paramref name="exchangeConfiguration" /> and waits for the RabbitMQ server to acknowledge that it has taken
+        /// responsibility for the message
+        /// </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{TResult}" /> that provides true once the RabbitMQ server has acknowledged the message, false
+        /// if it has not
+        /// </returns>
+        /// <exception cref="ArgumentNullException">When the provided <typeparamref name="TMessage" /> or <paramref name="exchangeConfiguration" /> is null</exception>
+        /// <remarks>
+        /// This bypasses the internal background queue, so that the caller can await the acknowledgment. Invalid arguments still
+        /// throw, any operational failure is logged and returns false. The acknowledgment only asserts that the server took
+        /// responsibility for the message, not that any consumer received it.
+        /// </remarks>
+        public async Task<bool> PushMessageWithConfirmationAsync<TMessage>(TMessage message, IExchangeConfiguration exchangeConfiguration, Action<BasicProperties> configureProperties = null, CancellationToken cancellationToken = default)
+        {
+            var activityName = $"Background Push with confirmation {typeof(TMessage).Name} [{exchangeConfiguration}]";
+
+            if (!this.TryGetRegisteredActivitySource(out var activitySource))
+            {
+                return false;
+            }
+
+            using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default);
+            return await this.MessageClientService.PushWithConfirmationAsync(this.ConnectionName, message, exchangeConfiguration, configureProperties, cancellationToken);
+        }
+
+        /// <summary>
+        /// Asynchronously pushes the specified <paramref name="messages" /> to the specified queue via the
+        /// <paramref name="exchangeConfiguration" /> and waits for the RabbitMQ server to acknowledge that it has taken
+        /// responsibility for each message
+        /// </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{TResult}" /> that provides true once the RabbitMQ server has acknowledged all the messages,
+        /// false as soon as one of them is not acknowledged
+        /// </returns>
+        /// <exception cref="ArgumentException">When the provided <paramref name="messages" /> collection is null</exception>
+        /// <exception cref="ArgumentNullException">When the provided <paramref name="exchangeConfiguration" /> is null</exception>
+        /// <remarks>
+        /// This bypasses the internal background queue, so that the caller can await the acknowledgment. The messages are published
+        /// one by one and the process stops at the first message that is not acknowledged. Since there is no transaction involved,
+        /// the messages that have been acknowledged before the failure are already held by the server.
+        /// </remarks>
+        public async Task<bool> PushMessagesWithConfirmationAsync<TMessage>(IEnumerable<TMessage> messages, IExchangeConfiguration exchangeConfiguration, Action<BasicProperties> configureProperties = null, CancellationToken cancellationToken = default)
+        {
+            var activityName = $"Background Push with confirmation Multiple {typeof(TMessage).Name} [{exchangeConfiguration}]";
+
+            if (!this.TryGetRegisteredActivitySource(out var activitySource))
+            {
+                return false;
+            }
+
+            using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default);
+            return await this.MessageClientService.PushWithConfirmationAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken);
+        }
+
+        /// <summary>
+        /// Tries to resolve the <see cref="ActivitySource" /> that is registered for the current <see cref="ConnectionName" />
+        /// </summary>
+        /// <param name="activitySource">The resolved <see cref="ActivitySource" />, if any</param>
+        /// <returns>The assert that the <see cref="ActivitySource" /> could be resolved</returns>
+        /// <remarks>
+        /// The <see cref="IRabbitMqConnectionProvider.GetRegisteredActivitySource" /> throws when the <see cref="ConnectionName" /> is
+        /// not registered, which happens when the initialization did not set it. Since the confirmation based push logs operational
+        /// failures and returns false, that exception is converted here.
+        /// </remarks>
+        private bool TryGetRegisteredActivitySource(out ActivitySource activitySource)
+        {
+            try
+            {
+                activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName);
+                return true;
+            }
+            catch (ArgumentException exception)
+            {
+                if (this.Logger.IsEnabled(LogLevel.Warning))
+                {
+                    this.Logger.LogWarning(exception, "No connection has been registered under the name {ConnectionName}", this.ConnectionName);
+                }
+
+                activitySource = null;
+                return false;
+            }
+        }
+
         /// <summary>
         /// Triggered when the application host is ready to start the service.
         /// </summary>