From de56f5c6a698d7b84c0fc5b73f81ce6c7748e661 Mon Sep 17 00:00:00 2001 From: atheate Date: Wed, 2 Sep 2026 15:55:18 +0200 Subject: [PATCH 1/3] Fix #47: publish with ACK --- .../MessagingBackgroundServiceTestFixture.cs | 88 ++++++- .../IMessagingBackgroundService.cs | 50 ++++ Mercurio.Hosting/Mercurio.Hosting.csproj | 4 +- .../MessagingBackgroundService.cs | 100 ++++++++ .../MessageClientBaseServiceTestFixture.cs | 34 +++ .../MessageClientServiceTestFixture.cs | 112 +++++++++ .../RabbitMqConnectionProviderTestFixture.cs | 28 +++ Mercurio/Mercurio.csproj | 5 +- Mercurio/Messaging/IMessageClientService.cs | 63 +++++ .../Messaging/MessageClientBaseService.cs | 66 ++++++ Mercurio/Messaging/MessageClientService.cs | 215 +++++++++++++++++- Mercurio/Messaging/MessagingErrors.cs | 134 +++++++++++ .../Provider/IRabbitMqConnectionProvider.cs | 16 ++ .../Provider/RabbitMqConnectionProvider.cs | 94 +++++++- README.md | 26 +++ 15 files changed, 1011 insertions(+), 24 deletions(-) create mode 100644 Mercurio/Messaging/MessagingErrors.cs diff --git a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs index 805de8f..b915862 100644 --- a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs +++ b/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,75 @@ 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.IsError, Is.False, "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>(), this.configurationMock.Object); + var notRegistered = await faultyService.PushMessageWithConfirmationAsync("XYZ", exchange, cancellationToken: cancellationTokenSource.Token); + faultyService.Dispose(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(notRegistered.IsError, Is.True, "a service that has not been initialized has no registered connection"); + Assert.That(notRegistered.FirstError.Code, Is.EqualTo(MessagingErrors.ConnectionNotRegisteredCode), "the missing registration is reported instead of being thrown"); + } + + Assert.That(() => 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 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.IsError, Is.False, "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>(), this.configurationMock.Object); + var notRegistered = await faultyService.PushMessagesWithConfirmationAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token); + faultyService.Dispose(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(notRegistered.IsError, Is.True, "a service that has not been initialized has no registered connection"); + Assert.That(notRegistered.FirstError.Code, Is.EqualTo(MessagingErrors.ConnectionNotRegisteredCode), "the missing registration is reported instead of being thrown"); + } + + Assert.That(() => this.backgroundService.PushMessagesWithConfirmationAsync((IEnumerable)null, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentException, "an invalid argument remains an exception"); + + await cancellationTokenSource.CancelAsync(); + } + [Test] public async Task VerifyPushMessagesAsync() { @@ -302,8 +374,11 @@ public async Task VerifyAutoRecoverySystem() using var cancellationTokenSource = new CancellationTokenSource(); var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), 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 +402,11 @@ public async Task VerifyAsyncAutoRecoverySystem() using var cancellationTokenSource = new CancellationTokenSource(); var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), 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); diff --git a/Mercurio.Hosting/IMessagingBackgroundService.cs b/Mercurio.Hosting/IMessagingBackgroundService.cs index 968d436..dc9a6e0 100644 --- a/Mercurio.Hosting/IMessagingBackgroundService.cs +++ b/Mercurio.Hosting/IMessagingBackgroundService.cs @@ -20,6 +20,8 @@ namespace Mercurio.Hosting { + using ErrorOr; + using Mercurio.Messaging; using Mercurio.Model; @@ -102,5 +104,53 @@ public interface IMessagingBackgroundService /// as 'application/json" /// Task PushMessagesAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + + /// + /// Asynchronously pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken + /// responsibility for the message + /// + /// The type of message + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// + /// When the provided or is null + /// + /// This bypasses the internal background queue, so that the caller can await the acknowledgment. Invalid arguments still + /// throw, only the operational failures are reported as an , see for the + /// reported ones. The acknowledgment only asserts that the server took responsibility for the message, not that any consumer + /// received it. + /// + Task> PushMessageWithConfirmationAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + + /// + /// Asynchronously pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken + /// responsibility for each message + /// + /// The type of message + /// The collection of to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// + /// When the provided collection is null + /// When the provided is null + /// + /// 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, the reported + /// provides how many of them under the + /// key. + /// + Task> PushMessagesWithConfirmationAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); } } diff --git a/Mercurio.Hosting/Mercurio.Hosting.csproj b/Mercurio.Hosting/Mercurio.Hosting.csproj index d766945..c2aebbf 100644 --- a/Mercurio.Hosting/Mercurio.Hosting.csproj +++ b/Mercurio.Hosting/Mercurio.Hosting.csproj @@ -4,7 +4,7 @@ enable disable net10.0;net9.0;net8.0;netstandard2.1 - 2.0.6 + 2.1.0 12.0 Starion Group S.A. Copyright © Starion Group S.A. @@ -28,7 +28,7 @@ true true - [Fix] Silent deserialization failure fix + [ADD] Provides Push ACK capability diff --git a/Mercurio.Hosting/MessagingBackgroundService.cs b/Mercurio.Hosting/MessagingBackgroundService.cs index 2c13722..4d9e56c 100644 --- a/Mercurio.Hosting/MessagingBackgroundService.cs +++ b/Mercurio.Hosting/MessagingBackgroundService.cs @@ -23,6 +23,8 @@ namespace Mercurio.Hosting using System.Collections.Concurrent; using System.Diagnostics; + using ErrorOr; + using Mercurio.Extensions; using Mercurio.Messaging; using Mercurio.Model; @@ -208,6 +210,104 @@ public async Task PushMessagesAsync(IEnumerable messages, IE await this.MessageClientService.PushAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken); } + /// + /// Asynchronously pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken + /// responsibility for the message + /// + /// The type of message + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// + /// When the provided or is null + /// + /// This bypasses the internal background queue, so that the caller can await the acknowledgment. Invalid arguments still + /// throw, only the operational failures are reported as an , see for the + /// reported ones. The acknowledgment only asserts that the server took responsibility for the message, not that any consumer + /// received it. + /// + public async Task> PushMessageWithConfirmationAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + var activityName = $"Background Push with confirmation {typeof(TMessage).Name} [{exchangeConfiguration}]"; + + if (!this.TryGetRegisteredActivitySource(out var activitySource, out var error)) + { + return error; + } + + using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); + return await this.MessageClientService.PushWithConfirmationAsync(this.ConnectionName, message, exchangeConfiguration, configureProperties, cancellationToken); + } + + /// + /// Asynchronously pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken + /// responsibility for each message + /// + /// The type of message + /// The collection of to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// + /// When the provided collection is null + /// When the provided is null + /// + /// 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, the reported + /// provides how many of them under the + /// key. + /// + public async Task> PushMessagesWithConfirmationAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + var activityName = $"Background Push with confirmation Multiple {typeof(TMessage).Name} [{exchangeConfiguration}]"; + + if (!this.TryGetRegisteredActivitySource(out var activitySource, out var error)) + { + return error; + } + + using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); + return await this.MessageClientService.PushWithConfirmationAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken); + } + + /// + /// Tries to resolve the that is registered for the current + /// + /// The resolved , if any + /// The that reports that the is not registered + /// The assert that the could be resolved + /// + /// The throws when the is + /// not registered, which happens when the initialization did not set it. Since the confirmation based push reports operational + /// failures as an , that exception is converted here. + /// + private bool TryGetRegisteredActivitySource(out ActivitySource activitySource, out Error error) + { + try + { + activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName); + error = default; + return true; + } + catch (ArgumentException exception) + { + this.Logger.LogWarning(exception, "No connection has been registered under the name {ConnectionName}", this.ConnectionName); + activitySource = null; + error = MessagingErrors.ConnectionNotRegistered(this.ConnectionName, exception); + return false; + } + } + /// /// Triggered when the application host is ready to start the service. /// diff --git a/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs b/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs index 6579eb8..0b15a23 100644 --- a/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs +++ b/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs @@ -22,6 +22,8 @@ namespace Mercurio.Tests.Messaging { using System.Diagnostics.CodeAnalysis; + using ErrorOr; + using Mercurio.Messaging; using Mercurio.Model; using Mercurio.Provider; @@ -167,5 +169,37 @@ public override Task PushAsync(string connectionName, TMessage message { throw new NotSupportedException(); } + + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for the message + /// + /// The type of message + /// The name of the registered connection to use. + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// A possible + /// An awaitable of + public override Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + throw new NotSupportedException(); + } + + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for each message + /// + /// The type of message + /// The name of the registered connection to use. + /// The collection of to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// An awaitable of + public override Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + throw new NotSupportedException(); + } } } diff --git a/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs b/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs index 1513e42..50ae524 100644 --- a/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs +++ b/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs @@ -20,6 +20,8 @@ namespace Mercurio.Tests.Messaging { + using ErrorOr; + using Mercurio.Extensions; using Mercurio.Messaging; using Mercurio.Model; @@ -28,6 +30,7 @@ namespace Mercurio.Tests.Messaging using Microsoft.Extensions.Logging; using RabbitMQ.Client; + using RabbitMQ.Client.Exceptions; using System.Collections.Concurrent; using System.Diagnostics; @@ -129,6 +132,115 @@ public async Task VerifyMessageExchangeWithDefaultExchangeAsync() } } + [Test] + public async Task VerifyPushWithConfirmationAsync() + { + const string exchangeName = "ConfirmationExchange"; + var exchangeConfiguration = new FanoutExchangeConfiguration(exchangeName); + + var listenObservable = await this.secondService.ListenAsync(SecondConnectionName, exchangeConfiguration); + var taskCompletion = new TaskCompletionSource(); + using var subscription = listenObservable.Subscribe(message => taskCompletion.TrySetResult(message)); + await Task.Delay(TimeOut); + + var acknowledged = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, FirstSentMessage, exchangeConfiguration); + + using (Assert.EnterMultipleScope()) + { + Assert.That(acknowledged.IsError, Is.False, "the RabbitMQ server should have acknowledged the message"); + Assert.That(acknowledged.Value, Is.EqualTo(Result.Success)); + } + + var received = await Task.WhenAny(taskCompletion.Task, Task.Delay(2000)); + Assert.That(received, Is.EqualTo(taskCompletion.Task), "an acknowledged message should still reach the listener"); + + var unregistered = await this.firstService.PushWithConfirmationAsync("UnregisteredConnection", FirstSentMessage, exchangeConfiguration); + + using (Assert.EnterMultipleScope()) + { + Assert.That(unregistered.IsError, Is.True, "an unregistered connection should be reported instead of being swallowed"); + Assert.That(unregistered.FirstError.Code, Is.EqualTo(MessagingErrors.PublishFailedCode)); + Assert.That(unregistered.FirstError.Metadata[MessagingErrors.ExceptionMetadataKey], Is.InstanceOf()); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, (string)null, exchangeConfiguration), Throws.ArgumentNullException, "an invalid argument remains an exception"); + Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, FirstSentMessage, null), Throws.ArgumentNullException); + } + + var rejectingConfiguration = await this.DeclareRejectPublishExchangeAsync("ConfirmationRejectExchange", "ConfirmationRejectQueue"); + + await this.firstService.PushWithConfirmationAsync(FirstConnectionName, FirstSentMessage, rejectingConfiguration); + var nacked = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, SecondSentMessage, rejectingConfiguration); + + using (Assert.EnterMultipleScope()) + { + Assert.That(nacked.IsError, Is.True, "a message the server refuses should be reported"); + Assert.That(nacked.FirstError.Code, Is.EqualTo(MessagingErrors.NotAcknowledgedCode)); + Assert.That(nacked.FirstError.Metadata[MessagingErrors.ExceptionMetadataKey], Is.InstanceOf()); + } + } + + [Test] + public async Task VerifyPushWithConfirmationMessagesAsync() + { + const string exchangeName = "ConfirmationCollectionExchange"; + var exchangeConfiguration = new FanoutExchangeConfiguration(exchangeName); + + // the cast is required: an array also binds to the single message overload, with TMessage being the array itself + IEnumerable messages = ["ABC", "DEF", "GHI"]; + + var acknowledged = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, exchangeConfiguration); + Assert.That(acknowledged.IsError, Is.False, "the RabbitMQ server should have acknowledged every message"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, (IEnumerable)null, exchangeConfiguration), Throws.ArgumentException, "an invalid argument remains an exception"); + Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, null), Throws.ArgumentNullException); + } + + var rejectingConfiguration = await this.DeclareRejectPublishExchangeAsync("ConfirmationCollectionRejectExchange", "ConfirmationCollectionRejectQueue"); + + var fillingTheQueue = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, "fills the queue", rejectingConfiguration); + Assert.That(fillingTheQueue.IsError, Is.False, "the rejecting queue accepts a single message before reaching its capacity"); + + var refused = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, rejectingConfiguration); + + using (Assert.EnterMultipleScope()) + { + Assert.That(refused.IsError, Is.True, "the batch should stop at the first refused message"); + Assert.That(refused.FirstError.Code, Is.EqualTo(MessagingErrors.NotAcknowledgedCode)); + Assert.That(refused.FirstError.Metadata[MessagingErrors.PublishedCountMetadataKey], Is.EqualTo(0), "the queue is already full, so nothing of the batch got through"); + Assert.That(refused.FirstError.Metadata[MessagingErrors.FailedIndexMetadataKey], Is.EqualTo(0)); + } + } + + /// + /// Declares a fanout exchange bound to a queue that only accepts a single message, and that asks the RabbitMQ server to reject + /// any further publication instead of dropping the oldest message + /// + /// The name of the exchange to declare + /// The name of the rejecting queue to declare + /// The that publishes to the declared exchange + private async Task DeclareRejectPublishExchangeAsync(string exchangeName, string queueName) + { + await using var lease = await this.firstService.LeaseChannelAsync(FirstConnectionName); + + await lease.Channel.ExchangeDeclareAsync(exchangeName, ExchangeType.Fanout, true); + await lease.Channel.QueueDeleteAsync(queueName, false, false); + + await lease.Channel.QueueDeclareAsync(queueName, false, false, false, new Dictionary + { + { "x-max-length", 1 }, + { "x-overflow", "reject-publish" } + }); + + await lease.Channel.QueueBindAsync(queueName, exchangeName, string.Empty); + + return new FanoutExchangeConfiguration(exchangeName); + } + [Test] [TestCase(20, 5, 10)] public async Task Should_ReUse_Channel_Under_Stress(int listenerCount = 20, int producerCount = 5, int pushRepetitions = 10) diff --git a/Mercurio.Tests/Messaging/RabbitMqConnectionProviderTestFixture.cs b/Mercurio.Tests/Messaging/RabbitMqConnectionProviderTestFixture.cs index 7197dbb..767c217 100644 --- a/Mercurio.Tests/Messaging/RabbitMqConnectionProviderTestFixture.cs +++ b/Mercurio.Tests/Messaging/RabbitMqConnectionProviderTestFixture.cs @@ -101,6 +101,34 @@ public async Task Should_Create_And_Cache_Connection_And_Channel() Assert.That(newLease.Channel.IsOpen, Is.True); } + [Test] + public async Task Should_Pool_Publisher_Confirmations_Channels_Separately() + { + IChannel plainChannel; + IChannel acknowledgingChannel; + + await using (var lease = await this.service.LeaseChannelAsync(ConnectionName)) + { + plainChannel = lease.Channel; + } + + await using (var lease = await this.service.LeaseChannelAsync(ConnectionName, true)) + { + acknowledgingChannel = lease.Channel; + } + + Assert.That(acknowledgingChannel, Is.Not.SameAs(plainChannel), "a channel supporting publisher confirmations cannot be reused as a regular one"); + + await using var pooledPlainLease = await this.service.LeaseChannelAsync(ConnectionName); + await using var pooledConfirmationLease = await this.service.LeaseChannelAsync(ConnectionName, true); + + using (Assert.EnterMultipleScope()) + { + Assert.That(pooledPlainLease.Channel, Is.SameAs(plainChannel), "each pool hands back its own released channel"); + Assert.That(pooledConfirmationLease.Channel, Is.SameAs(acknowledgingChannel)); + } + } + [Test] public void Should_Throw_When_Connection_Not_Registered() { diff --git a/Mercurio/Mercurio.csproj b/Mercurio/Mercurio.csproj index 65bc0f4..22f6a3f 100644 --- a/Mercurio/Mercurio.csproj +++ b/Mercurio/Mercurio.csproj @@ -4,7 +4,7 @@ enable disable net10.0;net9.0;net8.0;netstandard2.1 - 2.0.6 + 2.1.0 12.0 Starion Group S.A. Copyright © Starion Group S.A. @@ -28,12 +28,13 @@ true true - [Fix] Silent deserialization failure fix + [ADD] Provides Push ACK capability + diff --git a/Mercurio/Messaging/IMessageClientService.cs b/Mercurio/Messaging/IMessageClientService.cs index ebe246f..94f9254 100644 --- a/Mercurio/Messaging/IMessageClientService.cs +++ b/Mercurio/Messaging/IMessageClientService.cs @@ -20,6 +20,8 @@ namespace Mercurio.Messaging { + using ErrorOr; + using Mercurio.Model; using RabbitMQ.Client; @@ -85,6 +87,55 @@ public interface IMessageClientService : IDisposable /// Task PushAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for the message + /// + /// The type of message + /// The name of the registered connection to use. + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// A possible + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// + /// When the provided or is null + /// + /// Contrary to , + /// any publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, only the + /// operational failures are reported as an , see for the reported ones. + /// Publisher confirmations throttle the amount of outstanding publications, so this is slower than a regular push. The + /// acknowledgment only asserts that the server took responsibility for the message, not that any consumer received it. + /// + Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for each message + /// + /// The type of message + /// The name of the registered connection to use. + /// The collection of to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// + /// When the provided collection is null + /// When the provided is null + /// + /// Invalid arguments still throw, only the operational failures are reported as an , see + /// for the reported ones. 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, the reported provides how many of them + /// under the key. + /// + Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + /// /// Asynchronously leases a channel from the pool or creates one if necessary. /// @@ -92,5 +143,17 @@ public interface IMessageClientService : IDisposable /// An optional /// A of ValueTask LeaseChannelAsync(string connectionName, CancellationToken cancellationToken = default); + + /// + /// Asynchronously leases a channel from the pool or creates one if necessary. + /// + /// The name of the registered connection that should be used to establish the connection + /// + /// Asserts that the leased channel has to support publisher confirmations, so that the RabbitMQ server acknowledges any + /// published message + /// + /// An optional + /// A of + ValueTask LeaseChannelAsync(string connectionName, bool publisherConfirmationsEnabled, CancellationToken cancellationToken = default); } } diff --git a/Mercurio/Messaging/MessageClientBaseService.cs b/Mercurio/Messaging/MessageClientBaseService.cs index 54fac96..c4a5d25 100644 --- a/Mercurio/Messaging/MessageClientBaseService.cs +++ b/Mercurio/Messaging/MessageClientBaseService.cs @@ -22,6 +22,8 @@ namespace Mercurio.Messaging { using System.Diagnostics; + using ErrorOr; + using Mercurio.Extensions; using Mercurio.Model; using Mercurio.Provider; @@ -126,6 +128,55 @@ protected MessageClientBaseService(IRabbitMqConnectionProvider connectionProvide /// public abstract Task PushAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for the message + /// + /// The type of message + /// The name of the registered connection to use. + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// A possible + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// + /// When the provided or is null + /// + /// Contrary to , + /// any publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, only the + /// operational failures are reported as an , see for the reported ones. + /// Publisher confirmations throttle the amount of outstanding publications, so this is slower than a regular push. The + /// acknowledgment only asserts that the server took responsibility for the message, not that any consumer received it. + /// + public abstract Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for each message + /// + /// The type of message + /// The name of the registered connection to use. + /// The collection of to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// + /// When the provided collection is null + /// When the provided is null + /// + /// Invalid arguments still throw, only the operational failures are reported as an , see + /// for the reported ones. 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, the reported provides how many of them + /// under the key. + /// + public abstract Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// @@ -146,6 +197,21 @@ public async ValueTask LeaseChannelAsync(string connectionName, Ca return await this.ConnectionProvider.LeaseChannelAsync(connectionName, cancellationToken); } + /// + /// Asynchronously leases a channel from the pool or creates one if necessary. + /// + /// The name of the registered connection that should be used to establish the connection + /// + /// Asserts that the leased channel has to support publisher confirmations, so that the RabbitMQ server acknowledges any + /// published message + /// + /// An optional + /// A of + public async ValueTask LeaseChannelAsync(string connectionName, bool publisherConfirmationsEnabled, CancellationToken cancellationToken = default) + { + return await this.ConnectionProvider.LeaseChannelAsync(connectionName, publisherConfirmationsEnabled, cancellationToken); + } + /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// diff --git a/Mercurio/Messaging/MessageClientService.cs b/Mercurio/Messaging/MessageClientService.cs index 8f84fc4..066bde6 100644 --- a/Mercurio/Messaging/MessageClientService.cs +++ b/Mercurio/Messaging/MessageClientService.cs @@ -27,6 +27,8 @@ namespace Mercurio.Messaging using CommunityToolkit.HighPerformance; + using ErrorOr; + using Mercurio.Extensions; using Mercurio.Model; using Mercurio.Provider; @@ -36,6 +38,7 @@ namespace Mercurio.Messaging using RabbitMQ.Client; using RabbitMQ.Client.Events; + using RabbitMQ.Client.Exceptions; /// /// The is the concrete implementation of the @@ -78,7 +81,7 @@ public override Task> ListenAsync(string connect { if (exchangeConfiguration == null) { - throw new ArgumentNullException(nameof(exchangeConfiguration), "The exchange configuration cannot be null"); + throw new ArgumentNullException(nameof(exchangeConfiguration)); } return this.ListenInternalAsync(connectionName, exchangeConfiguration, cancellationToken); @@ -96,7 +99,7 @@ public override Task AddListenerAsync(string connectionName, IExcha { if (exchangeConfiguration == null) { - throw new ArgumentNullException(nameof(exchangeConfiguration), "The exchange configuration cannot be null"); + throw new ArgumentNullException(nameof(exchangeConfiguration)); } return this.AddListenerInternalAsync(connectionName, exchangeConfiguration, onReceiveAsync, cancellationToken); @@ -161,6 +164,84 @@ public override Task PushAsync(string connectionName, TMessage message return this.PushInternalAsync(connectionName, message, exchangeConfiguration, configureProperties, activityName, context, cancellationToken); } + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for the message + /// + /// The type of message + /// The name of the registered connection to use. + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// A possible + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// + /// When the provided or is null + /// + /// Contrary to , + /// any publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, only the + /// operational failures are reported as an , see for the reported ones. + /// Publisher confirmations throttle the amount of outstanding publications, so this is slower than a regular push. The + /// acknowledgment only asserts that the server took responsibility for the message, not that any consumer received it. + /// + public override Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + if (Equals(message, default(TMessage))) + { + throw new ArgumentNullException(nameof(message), "The message to be sent can not be null"); + } + + if (exchangeConfiguration == null) + { + throw new ArgumentNullException(nameof(exchangeConfiguration), "The exchange configuration cannot be null"); + } + + var context = Activity.Current == null ? default : Activity.Current.Context; + var activityName = $"Push with confirmation {typeof(TMessage).Name} [{exchangeConfiguration}]"; + + return this.TryPublishAsync(connectionName, message, exchangeConfiguration, configureProperties, activityName, context, cancellationToken); + } + + /// + /// Pushes the specified to the specified queue via the + /// and waits for the RabbitMQ server to acknowledge that it has taken responsibility for each message + /// + /// The type of message + /// The name of the registered connection to use. + /// The collection of to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// + /// When the provided collection is null + /// When the provided is null + /// + /// Invalid arguments still throw, only the operational failures are reported as an , see + /// for the reported ones. 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, the reported provides how many of them + /// under the key. + /// + public override Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + if (messages == null) + { + throw new ArgumentException("The messages collection cannot be null", nameof(messages)); + } + + if (exchangeConfiguration == null) + { + throw new ArgumentNullException(nameof(exchangeConfiguration), "The exchange configuration cannot be null"); + } + + return this.PushMultipleWithConfirmationInternalAsync(connectionName, messages, exchangeConfiguration, configureProperties, cancellationToken); + } + /// /// Listens for messages of type on the specified queue. /// @@ -309,12 +390,132 @@ private async Task PushMultipleInternalAsync(string connectionName, IE /// as 'application/json" /// private async Task PushInternalAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties, string activityName, ActivityContext activityContext, CancellationToken cancellationToken) + { + try + { + await this.PublishInternalAsync(connectionName, message, exchangeConfiguration, configureProperties, activityName, activityContext, false, cancellationToken); + } + catch (Exception exception) + { + this.Logger.LogError(exception, "The message {MessageName} could not be queued to {MessageQueue} reason : {Exception}", typeof(TMessage).Name, exchangeConfiguration.QueueName, exception.Message); + } + } + + /// + /// Publishes the specified to the specified queue via the + /// , waiting for the RabbitMQ server to acknowledge each of them + /// + /// The type of message + /// The name of the registered connection to use. + /// The collection of to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// An optional + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// + /// + /// The process stops at the first message that is not acknowledged, so that a broken connection does not have every remaining + /// message pay the connection retry backoff + /// + private async Task> PushMultipleWithConfirmationInternalAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties, CancellationToken cancellationToken) + { + var activitySource = this.ConnectionProvider.GetRegisteredActivitySource(connectionName); + + var context = Activity.Current == null ? default : Activity.Current.Context; + var activityName = $"Push with confirmation {typeof(TMessage).Name} collection [{exchangeConfiguration}]"; + + using var activity = this.StartActivity(activitySource, context, activityName, ActivityKind.Producer); + + var messagesList = messages.ToList(); + + for (var messageIndex = 0; messageIndex < messagesList.Count; messageIndex++) + { + var subActivityName = activity == null ? null : $"{activityName} [{messageIndex + 1}/{messagesList.Count}]"; + var result = await this.TryPublishAsync(connectionName, messagesList[messageIndex], exchangeConfiguration, configureProperties, subActivityName, context, cancellationToken); + + if (result.IsError) + { + return MessagingErrors.BatchFailed(result.FirstError, messageIndex, messageIndex); + } + } + + return Result.Success; + } + + /// + /// Publishes the specified to the specified queue via the + /// , waiting for the RabbitMQ server to acknowledge it and reporting any failure as + /// an instead of throwing + /// + /// The type of message + /// The name of the registered connection to use. + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// + /// Defines the name of an that should be initialized before sending the message, for traceability. + /// In case of null or empty, no is started + /// + /// An optional . If not set, current context will be based on + /// A possible + /// + /// An awaitable of that provides once the + /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// + private async Task> TryPublishAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties, string activityName, ActivityContext activityContext, CancellationToken cancellationToken) + { + try + { + await this.PublishInternalAsync(connectionName, message, exchangeConfiguration, configureProperties, activityName, activityContext, true, cancellationToken); + return Result.Success; + } + catch (PublishException publishException) + { + this.Logger.LogWarning(publishException, "The message {MessageName} has not been acknowledged by the RabbitMQ server for {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); + return MessagingErrors.NotAcknowledged(publishException); + } + catch (Exception exception) + { + this.Logger.LogWarning(exception, "The message {MessageName} could not be published to {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); + return MessagingErrors.PublishFailed(exception); + } + } + + /// + /// Publishes the specified to the specified queue via the + /// , reporting any failure to the caller + /// + /// The type of message + /// The name of the registered connection to use. + /// The to push + /// The that should be used to configure the queue and exchange to use + /// Possible action to configure additional properties + /// + /// Defines the name of an that should be initialized before sending the message, for traceability. + /// information will be sent in the message header. + /// In case of null or empty, no is started + /// + /// An optional . If not set, current context will be based on + /// + /// Asserts that the message has to be published on a channel that supports publisher confirmations, making the + /// + /// await the acknowledgment of the RabbitMQ server + /// + /// A possible + /// An awaitable + /// + /// By default, the is configured to use the mode and sets the + /// as 'application/json" + /// + private async Task PublishInternalAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties, string activityName, ActivityContext activityContext, bool publisherConfirmationsEnabled, CancellationToken cancellationToken) { Activity activity = null; try { - await using var channelLease = await this.LeaseChannelAsync(connectionName, cancellationToken); + await using var channelLease = await this.LeaseChannelAsync(connectionName, publisherConfirmationsEnabled, cancellationToken); await exchangeConfiguration.EnsureQueueAndExchangeAreDeclaredAsync(channelLease.Channel, true); var properties = new BasicProperties @@ -350,14 +551,18 @@ await channelLease.Channel.BasicPublishAsync(exchangeConfiguration.PushExchangeN routingKey, false, properties, body, cancellationToken); - this.Logger.LogDebug("Message Body {Body}", Encoding.UTF8.GetString(body.ToArray())); + if (this.Logger.IsEnabled(LogLevel.Debug)) + { + this.Logger.LogDebug("Message Body {Body}", Encoding.UTF8.GetString(body.ToArray())); + } + this.Logger.LogInformation("Message {MessageName} sent to {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); activity?.SetStatus(ActivityStatusCode.Ok); } catch (Exception exception) { - this.Logger.LogError(exception, "The message {MessageName} could not be queued to {MessageQueue} reason : {Exception}", typeof(TMessage).Name, exchangeConfiguration.QueueName, exception.Message); activity?.SetStatus(ActivityStatusCode.Error, exception.Message); + throw; } finally { diff --git a/Mercurio/Messaging/MessagingErrors.cs b/Mercurio/Messaging/MessagingErrors.cs new file mode 100644 index 0000000..f522da3 --- /dev/null +++ b/Mercurio/Messaging/MessagingErrors.cs @@ -0,0 +1,134 @@ +// ------------------------------------------------------------------------------------------------- +// +// +// Copyright 2025 - 2026 Starion Group S.A. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// +// ------------------------------------------------------------------------------------------------ + +namespace Mercurio.Messaging +{ + using ErrorOr; + + using RabbitMQ.Client.Exceptions; + + /// + /// Provides all that can be reported by the when publishing a message + /// with publisher confirmations enabled + /// + /// + /// The does not carry an , so the original is provided + /// through the , under the key + /// + public static class MessagingErrors + { + /// + /// The key that provides the at the origin of the + /// + public const string ExceptionMetadataKey = "exception"; + + /// + /// The key that provides the amount of messages that have been acknowledged before a batch failed + /// + public const string PublishedCountMetadataKey = "publishedCount"; + + /// + /// The key that provides the index of the message that failed within a batch + /// + public const string FailedIndexMetadataKey = "failedIndex"; + + /// + /// The reported when the RabbitMQ server did not acknowledge the published message + /// + public const string NotAcknowledgedCode = "Mercurio.Publish.NotAcknowledged"; + + /// + /// The reported when the message could not be published at all + /// + public const string PublishFailedCode = "Mercurio.Publish.Failed"; + + /// + /// The reported when the connection that should be used is not registered + /// + public const string ConnectionNotRegisteredCode = "Mercurio.Publish.ConnectionNotRegistered"; + + /// + /// Provides the that reports that the RabbitMQ server did not acknowledge the published message + /// + /// The that reports the negative acknowledgment + /// The + public static Error NotAcknowledged(PublishException exception) + { + return Error.Failure(NotAcknowledgedCode, $"The RabbitMQ server did not acknowledge the message: {exception?.Message}", + new Dictionary + { + { ExceptionMetadataKey, exception }, + { "isReturn", exception?.IsReturn }, + { "publishSequenceNumber", exception?.PublishSequenceNumber } + }); + } + + /// + /// Provides the that reports that the message could not be published + /// + /// The that occured while publishing + /// The + public static Error PublishFailed(Exception exception) + { + return Error.Unexpected(PublishFailedCode, $"The message could not be published: {exception?.Message}", + new Dictionary + { + { ExceptionMetadataKey, exception } + }); + } + + /// + /// Provides the that reports that the connection that should be used is not registered + /// + /// The name of the connection that is not registered + /// The that reports the missing registration + /// The + public static Error ConnectionNotRegistered(string connectionName, Exception exception) + { + return Error.Failure(ConnectionNotRegisteredCode, $"No connection has been registered under the name {connectionName}", + new Dictionary + { + { ExceptionMetadataKey, exception } + }); + } + + /// + /// Provides the that reports that a message of a batch could not be published, keeping the + /// , and of the reported + /// + /// + /// The that has been reported for the failing message + /// The amount of messages that have been acknowledged before the failure + /// The index, within the batch, of the message that failed + /// The + /// + /// Since publishing a batch is not transactional, the messages that have been acknowledged before the failure are already held + /// by the RabbitMQ server + /// + public static Error BatchFailed(Error error, int publishedCount, int failedIndex) + { + var metadata = error.Metadata == null ? new Dictionary() : new Dictionary(error.Metadata); + metadata[PublishedCountMetadataKey] = publishedCount; + metadata[FailedIndexMetadataKey] = failedIndex; + + return Error.Custom(error.NumericType, error.Code, error.Description, metadata); + } + } +} diff --git a/Mercurio/Provider/IRabbitMqConnectionProvider.cs b/Mercurio/Provider/IRabbitMqConnectionProvider.cs index 1cb8247..d39522b 100644 --- a/Mercurio/Provider/IRabbitMqConnectionProvider.cs +++ b/Mercurio/Provider/IRabbitMqConnectionProvider.cs @@ -48,6 +48,22 @@ public interface IRabbitMqConnectionProvider : ICanReleaseChannel /// A of ValueTask LeaseChannelAsync(string connectionName, CancellationToken cancellationToken = default); + /// + /// Asynchronously leases a channel from the pool or creates one if necessary. + /// + /// The name of the registered connection that should be used to establish the connection + /// + /// Asserts that the leased channel has to support publisher confirmations, so that the RabbitMQ server acknowledges any + /// published message + /// + /// An optional + /// A of + /// + /// Channels that support publisher confirmations are pooled separately, since the publisher confirmations can only be enabled + /// at the creation of the channel + /// + ValueTask LeaseChannelAsync(string connectionName, bool publisherConfirmationsEnabled, CancellationToken cancellationToken = default); + /// /// Register a new that could be resolved to provide an /// diff --git a/Mercurio/Provider/RabbitMqConnectionProvider.cs b/Mercurio/Provider/RabbitMqConnectionProvider.cs index 1501475..be7f07d 100644 --- a/Mercurio/Provider/RabbitMqConnectionProvider.cs +++ b/Mercurio/Provider/RabbitMqConnectionProvider.cs @@ -39,6 +39,12 @@ namespace Mercurio.Provider /// internal sealed class RabbitMqConnectionProvider : IRabbitMqConnectionProvider, IDisposable { + /// + /// The suffix that is appended to a connection name to identify the pool that holds the channels for which the publisher + /// confirmations are enabled + /// + private const string ConfirmationPoolKeySuffix = ":confirmations"; + /// /// Gets the that caches that will provides traceabilities per connection /// @@ -49,6 +55,16 @@ internal sealed class RabbitMqConnectionProvider : IRabbitMqConnectionProvider, /// private readonly ConcurrentDictionary> channelPool = new(); + /// + /// The that keeps track of the created for which the + /// publisher confirmations are enabled + /// + /// + /// The does not expose whether it has been created with publisher confirmations enabled, so the + /// provider has to keep track of it to be able to release a channel into the pool it has been leased from + /// + private readonly ConcurrentDictionary confirmationChannels = new(); + /// /// Gets the that caches living created based on registered /// @@ -134,8 +150,13 @@ public void Dispose() existingConnection.Value.Dispose(); } - if (this.channelPool.TryGetValue(existingConnection.Key, out var disposablePool)) + foreach (var poolKey in new[] { GetPoolKey(existingConnection.Key, false), GetPoolKey(existingConnection.Key, true) }) { + if (!this.channelPool.TryGetValue(poolKey, out var disposablePool)) + { + continue; + } + foreach (var channel in disposablePool) { channel.Dispose(); @@ -153,6 +174,7 @@ public void Dispose() activitySource?.Dispose(); } + this.confirmationChannels.Clear(); this.connections.Clear(); } @@ -199,7 +221,26 @@ public void RegisterFactory(string name, FuncA of public async ValueTask LeaseChannelAsync(string connectionName, CancellationToken cancellationToken = default) { - return new ChannelLease(connectionName, await this.GetChannelAsync(connectionName, cancellationToken), this); + return await this.LeaseChannelAsync(connectionName, false, cancellationToken); + } + + /// + /// Asynchronously leases a channel from the pool or creates one if necessary. + /// + /// The name of the registered connection that should be used to establish the connection + /// + /// Asserts that the leased channel has to support publisher confirmations, so that the RabbitMQ server acknowledges any + /// published message + /// + /// An optional + /// A of + /// + /// Channels that support publisher confirmations are pooled separately, since the publisher confirmations can only be enabled + /// at the creation of the channel + /// + public async ValueTask LeaseChannelAsync(string connectionName, bool publisherConfirmationsEnabled, CancellationToken cancellationToken = default) + { + return new ChannelLease(connectionName, await this.GetChannelAsync(connectionName, publisherConfirmationsEnabled, cancellationToken), this); } /// @@ -267,6 +308,10 @@ private void RegisterActivitySource(string connectionName, ActivitySource activi /// Asynchronously acquires a channel from the pool or creates one if necessary. /// /// The name of the registered connection that should be used to establish the connection + /// + /// Asserts that the acquired channel has to support publisher confirmations, so that the RabbitMQ server acknowledges any + /// published message + /// /// /// The used to observe cancellation requests. /// @@ -276,22 +321,39 @@ private void RegisterActivitySource(string connectionName, ActivitySource activi /// /// Thrown if the operation is cancelled before a channel is acquired. /// - private async Task GetChannelAsync(string connectionName, CancellationToken cancellationToken = default) + private async Task GetChannelAsync(string connectionName, bool publisherConfirmationsEnabled, CancellationToken cancellationToken = default) { await this.poolGates.GetOrAdd(connectionName, new SemaphoreSlim(this.registeredFactoriesPoolSize[connectionName])).WaitAsync(cancellationToken).ConfigureAwait(false); - if (!this.channelPool.TryGetValue(connectionName, out var queue) || !queue.TryDequeue(out var channel) || channel is { IsOpen: false }) + var poolKey = GetPoolKey(connectionName, publisherConfirmationsEnabled); + + if (!this.channelPool.TryGetValue(poolKey, out var queue) || !queue.TryDequeue(out var channel) || channel is { IsOpen: false }) { - channel = await this.CreateChannelWithRetryAsync(connectionName, cancellationToken).ConfigureAwait(false); + channel = await this.CreateChannelWithRetryAsync(connectionName, publisherConfirmationsEnabled, cancellationToken).ConfigureAwait(false); } return channel; } + /// + /// Computes the key of the channel pool that serves the provided + /// + /// The name of the registered connection + /// Asserts that the pool holds channels that support publisher confirmations + /// The key of the channel pool + private static string GetPoolKey(string connectionName, bool publisherConfirmationsEnabled) + { + return publisherConfirmationsEnabled ? connectionName + ConfirmationPoolKeySuffix : connectionName; + } + /// /// Attempts to create a new RabbitMQ channel with retry logic using exponential backoff strategy. /// /// The name of the registered connection that should be used to establish the connection + /// + /// Asserts that the created channel has to support publisher confirmations, so that the RabbitMQ server acknowledges any + /// published message + /// /// A to observe while waiting for the task to complete. /// /// A task representing the asynchronous operation. The result is an open for the current thread. @@ -307,13 +369,13 @@ private async Task GetChannelAsync(string connectionName, Cancellation /// Thrown if the operation is cancelled via /// . /// - private async Task CreateChannelWithRetryAsync(string connectionName, CancellationToken cancellationToken) + private async Task CreateChannelWithRetryAsync(string connectionName, bool publisherConfirmationsEnabled, CancellationToken cancellationToken) { var result = await Policy .Handle() .WaitAndRetryAsync(this.retryPolicyConfiguration.MaxConnectionRetryAttempts, attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)), (exception, delay, attemptNumber, _) => this.logger.LogWarning(exception, "ChannelLease creation attempt {Attempt} failed. Retrying in {Delay}", attemptNumber, delay)) - .ExecuteAndCaptureAsync(() => this.CreateChannelAsync(connectionName, cancellationToken)); + .ExecuteAndCaptureAsync(() => this.CreateChannelAsync(connectionName, publisherConfirmationsEnabled, cancellationToken)); if (result.Outcome != OutcomeType.Successful) { @@ -341,9 +403,10 @@ private void ReleaseChannel(string connectionName, IChannel channel) case null: return; case { IsOpen: true }: - this.channelPool.GetOrAdd(connectionName, new ConcurrentQueue()).Enqueue(channel); + this.channelPool.GetOrAdd(GetPoolKey(connectionName, this.confirmationChannels.ContainsKey(channel)), new ConcurrentQueue()).Enqueue(channel); break; default: + this.confirmationChannels.TryRemove(channel, out _); channel.Dispose(); break; } @@ -356,16 +419,27 @@ private void ReleaseChannel(string connectionName, IChannel channel) /// a new connection is established using the provided connection factory before creating the channel. /// /// The name of the registered connection that should be used to establish the connection + /// + /// Asserts that the created channel has to support publisher confirmations, so that the RabbitMQ server acknowledges any + /// published message + /// /// A token to monitor for cancellation requests. /// A task that represents the asynchronous operation, containing the created . /// Thrown if the operation is canceled via the cancellation token. /// Thrown if the connection factory is not properly initialized. /// Thrown if connection or channel creation fails due to network or server issues. - private async Task CreateChannelAsync(string connectionName, CancellationToken cancellationToken) + private async Task CreateChannelAsync(string connectionName, bool publisherConfirmationsEnabled, CancellationToken cancellationToken) { var connection = await this.GetConnectionAsync(connectionName, cancellationToken).ConfigureAwait(false); - var channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + if (!publisherConfirmationsEnabled) + { + return await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + } + + var options = new CreateChannelOptions(true, true); + var channel = await connection.CreateChannelAsync(options, cancellationToken).ConfigureAwait(false); + this.confirmationChannels[channel] = true; return channel; } diff --git a/README.md b/README.md index 159a553..e1b0d72 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,32 @@ var exchangeConfiguration = new DirectExchangeConfiguration("DirectQueue", "AnEx await messageClientService.PushAsync("RegisteredConnection","A message to be sent",exchangeConfiguration); ``` +#### Push Message And Wait For The Server Acknowledgment +_PushAsync_ is fire and forget : it returns as soon as the message has been handed to the transport, and any failure is only logged. +When you need to know that the RabbitMQ server took responsibility for the message, use _PushWithConfirmationAsync_. It publishes on a +channel that has publisher confirmations enabled and reports the outcome as an `ErrorOr` instead of throwing. + +```csharp +var messageClientService = serviceProvider.GetRequiredService(); +var exchangeConfiguration = new DirectExchangeConfiguration("DirectQueue", "AnExchange", "SomeRouting"); +var result = await messageClientService.PushWithConfirmationAsync("RegisteredConnection", "A message to be sent", exchangeConfiguration); + +if (result.IsError) +{ + // MessagingErrors exposes every reported code, and the originating exception is available in the metadata + Console.WriteLine($"{result.FirstError.Code}: {result.FirstError.Description}"); + var exception = (Exception)result.FirstError.Metadata[MessagingErrors.ExceptionMetadataKey]; +} +``` + +Two things to keep in mind : +- Publisher confirmations throttle the amount of outstanding publications, so a confirmed push is slower than a regular one. Channels + supporting confirmations are pooled separately, since confirmations can only be enabled when the channel is created. +- An acknowledgment only asserts that the **server** took responsibility for the message, never that a consumer received it. A message + published to an exchange that routes to no queue is legitimately acknowledged. + +Invalid arguments still throw : only the operational failures are reported as an `Error`. + #### Listen After Message ```csharp var messageClientService = serviceProvider.GetRequiredService(); From f2d984116018ab01dcd81e2ebe98a07526e851da Mon Sep 17 00:00:00 2001 From: atheate Date: Wed, 2 Sep 2026 16:05:12 +0200 Subject: [PATCH 2/3] SQ fix --- Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs b/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs index 50ae524..f3a8383 100644 --- a/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs +++ b/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs @@ -211,8 +211,8 @@ public async Task VerifyPushWithConfirmationMessagesAsync() { Assert.That(refused.IsError, Is.True, "the batch should stop at the first refused message"); Assert.That(refused.FirstError.Code, Is.EqualTo(MessagingErrors.NotAcknowledgedCode)); - Assert.That(refused.FirstError.Metadata[MessagingErrors.PublishedCountMetadataKey], Is.EqualTo(0), "the queue is already full, so nothing of the batch got through"); - Assert.That(refused.FirstError.Metadata[MessagingErrors.FailedIndexMetadataKey], Is.EqualTo(0)); + Assert.That(refused.FirstError.Metadata[MessagingErrors.PublishedCountMetadataKey], Is.Zero, "the queue is already full, so nothing of the batch got through"); + Assert.That(refused.FirstError.Metadata[MessagingErrors.FailedIndexMetadataKey], Is.Zero); } } From 4176de316cd927cfbd041c0cddf20951175ee55f Mon Sep 17 00:00:00 2001 From: atheate Date: Thu, 3 Sep 2026 11:12:06 +0200 Subject: [PATCH 3/3] Removes ErrorOr dependency --- .../MessagingBackgroundServiceTestFixture.cs | 20 +-- .../IMessagingBackgroundService.cs | 23 ++- .../MessagingBackgroundService.cs | 46 +++--- .../MessageClientBaseServiceTestFixture.cs | 10 +- .../MessageClientServiceTestFixture.cs | 43 +++--- Mercurio/Mercurio.csproj | 1 - Mercurio/Messaging/IMessageClientService.cs | 30 ++-- .../Messaging/MessageClientBaseService.cs | 35 +++-- Mercurio/Messaging/MessageClientService.cs | 92 +++++++----- Mercurio/Messaging/MessagingErrors.cs | 134 ------------------ Mercurio/Messaging/RpcServerService.cs | 5 +- .../Provider/RabbitMqConnectionProvider.cs | 8 +- README.md | 11 +- 13 files changed, 155 insertions(+), 303 deletions(-) delete mode 100644 Mercurio/Messaging/MessagingErrors.cs diff --git a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs index b915862..9d3d77d 100644 --- a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs +++ b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs @@ -170,7 +170,7 @@ public async Task VerifyPushMessageWithConfirmationAsync() using (Assert.EnterMultipleScope()) { - Assert.That(acknowledged.IsError, Is.False, "the RabbitMQ server should have acknowledged the message"); + Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged the message"); Assert.That(this.backgroundService.ReceivedMessages, Does.Contain("ABC")); } @@ -178,13 +178,9 @@ public async Task VerifyPushMessageWithConfirmationAsync() var notRegistered = await faultyService.PushMessageWithConfirmationAsync("XYZ", exchange, cancellationToken: cancellationTokenSource.Token); faultyService.Dispose(); - using (Assert.EnterMultipleScope()) - { - Assert.That(notRegistered.IsError, Is.True, "a service that has not been initialized has no registered connection"); - Assert.That(notRegistered.FirstError.Code, Is.EqualTo(MessagingErrors.ConnectionNotRegisteredCode), "the missing registration is reported instead of being thrown"); - } + Assert.That(notRegistered, Is.False, "a service that has not been initialized has no registered connection, which is reported instead of being thrown"); - Assert.That(() => this.backgroundService.PushMessageWithConfirmationAsync((string)null, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentNullException, "an invalid argument remains an exception"); + await Assert.ThatAsync(() => this.backgroundService.PushMessageWithConfirmationAsync((string)null!, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentNullException, "an invalid argument remains an exception"); await cancellationTokenSource.CancelAsync(); } @@ -206,7 +202,7 @@ public async Task VerifyPushMessagesWithConfirmationAsync() using (Assert.EnterMultipleScope()) { - Assert.That(acknowledged.IsError, Is.False, "the RabbitMQ server should have acknowledged every message"); + Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged every message"); Assert.That(this.backgroundService.ReceivedMessages, Is.EquivalentTo(messages)); } @@ -214,13 +210,9 @@ public async Task VerifyPushMessagesWithConfirmationAsync() var notRegistered = await faultyService.PushMessagesWithConfirmationAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token); faultyService.Dispose(); - using (Assert.EnterMultipleScope()) - { - Assert.That(notRegistered.IsError, Is.True, "a service that has not been initialized has no registered connection"); - Assert.That(notRegistered.FirstError.Code, Is.EqualTo(MessagingErrors.ConnectionNotRegisteredCode), "the missing registration is reported instead of being thrown"); - } + Assert.That(notRegistered, Is.False, "a service that has not been initialized has no registered connection, which is reported instead of being thrown"); - Assert.That(() => this.backgroundService.PushMessagesWithConfirmationAsync((IEnumerable)null, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentException, "an invalid argument remains an exception"); + await Assert.ThatAsync(() => this.backgroundService.PushMessagesWithConfirmationAsync((IEnumerable)null!, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentException, "an invalid argument remains an exception"); await cancellationTokenSource.CancelAsync(); } diff --git a/Mercurio.Hosting/IMessagingBackgroundService.cs b/Mercurio.Hosting/IMessagingBackgroundService.cs index dc9a6e0..f8b0043 100644 --- a/Mercurio.Hosting/IMessagingBackgroundService.cs +++ b/Mercurio.Hosting/IMessagingBackgroundService.cs @@ -20,8 +20,6 @@ namespace Mercurio.Hosting { - using ErrorOr; - using Mercurio.Messaging; using Mercurio.Model; @@ -116,17 +114,16 @@ public interface IMessagingBackgroundService /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not /// /// When the provided or is null /// /// This bypasses the internal background queue, so that the caller can await the acknowledgment. Invalid arguments still - /// throw, only the operational failures are reported as an , see for the - /// reported ones. The acknowledgment only asserts that the server took responsibility for the message, not that any consumer - /// received it. + /// 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. /// - Task> PushMessageWithConfirmationAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + Task PushMessageWithConfirmationAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); /// /// Asynchronously pushes the specified to the specified queue via the @@ -139,18 +136,16 @@ public interface IMessagingBackgroundService /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged all the messages, + /// false as soon as one of them is not acknowledged /// /// When the provided collection is null /// When the provided is null /// /// 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, the reported - /// provides how many of them under the - /// key. + /// the messages that have been acknowledged before the failure are already held by the server. /// - Task> PushMessagesWithConfirmationAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + Task PushMessagesWithConfirmationAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); } } diff --git a/Mercurio.Hosting/MessagingBackgroundService.cs b/Mercurio.Hosting/MessagingBackgroundService.cs index 4d9e56c..c03f156 100644 --- a/Mercurio.Hosting/MessagingBackgroundService.cs +++ b/Mercurio.Hosting/MessagingBackgroundService.cs @@ -23,8 +23,6 @@ namespace Mercurio.Hosting using System.Collections.Concurrent; using System.Diagnostics; - using ErrorOr; - using Mercurio.Extensions; using Mercurio.Messaging; using Mercurio.Model; @@ -221,23 +219,22 @@ public async Task PushMessagesAsync(IEnumerable messages, IE /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not /// /// When the provided or is null /// /// This bypasses the internal background queue, so that the caller can await the acknowledgment. Invalid arguments still - /// throw, only the operational failures are reported as an , see for the - /// reported ones. The acknowledgment only asserts that the server took responsibility for the message, not that any consumer - /// received it. + /// 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. /// - public async Task> PushMessageWithConfirmationAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + public async Task PushMessageWithConfirmationAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) { var activityName = $"Background Push with confirmation {typeof(TMessage).Name} [{exchangeConfiguration}]"; - if (!this.TryGetRegisteredActivitySource(out var activitySource, out var error)) + if (!this.TryGetRegisteredActivitySource(out var activitySource)) { - return error; + return false; } using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); @@ -255,25 +252,23 @@ public async Task> PushMessageWithConfirmationAsync(T /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged all the messages, + /// false as soon as one of them is not acknowledged /// /// When the provided collection is null /// When the provided is null /// /// 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, the reported - /// provides how many of them under the - /// key. + /// the messages that have been acknowledged before the failure are already held by the server. /// - public async Task> PushMessagesWithConfirmationAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + public async Task PushMessagesWithConfirmationAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) { var activityName = $"Background Push with confirmation Multiple {typeof(TMessage).Name} [{exchangeConfiguration}]"; - if (!this.TryGetRegisteredActivitySource(out var activitySource, out var error)) + if (!this.TryGetRegisteredActivitySource(out var activitySource)) { - return error; + return false; } using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); @@ -284,26 +279,27 @@ public async Task> PushMessagesWithConfirmationAsync( /// Tries to resolve the that is registered for the current /// /// The resolved , if any - /// The that reports that the is not registered /// The assert that the could be resolved /// /// The throws when the is - /// not registered, which happens when the initialization did not set it. Since the confirmation based push reports operational - /// failures as an , that exception is converted here. + /// 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. /// - private bool TryGetRegisteredActivitySource(out ActivitySource activitySource, out Error error) + private bool TryGetRegisteredActivitySource(out ActivitySource activitySource) { try { activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName); - error = default; return true; } catch (ArgumentException exception) { - this.Logger.LogWarning(exception, "No connection has been registered under the name {ConnectionName}", this.ConnectionName); + if (this.Logger.IsEnabled(LogLevel.Warning)) + { + this.Logger.LogWarning(exception, "No connection has been registered under the name {ConnectionName}", this.ConnectionName); + } + activitySource = null; - error = MessagingErrors.ConnectionNotRegistered(this.ConnectionName, exception); return false; } } diff --git a/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs b/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs index 0b15a23..1e890cf 100644 --- a/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs +++ b/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs @@ -22,8 +22,6 @@ namespace Mercurio.Tests.Messaging { using System.Diagnostics.CodeAnalysis; - using ErrorOr; - using Mercurio.Messaging; using Mercurio.Model; using Mercurio.Provider; @@ -180,8 +178,8 @@ public override Task PushAsync(string connectionName, TMessage message /// The that should be used to configure the queue and exchange to use /// Possible action to configure additional properties /// A possible - /// An awaitable of - public override Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + /// An awaitable that asserts the acknowledgment of the RabbitMQ server + public override Task PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) { throw new NotSupportedException(); } @@ -196,8 +194,8 @@ public override Task> PushWithConfirmationAsync(strin /// The that should be used to configure the queue and exchange to use /// Possible action to configure additional properties /// An optional - /// An awaitable of - public override Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + /// An awaitable that asserts the acknowledgment of the RabbitMQ server + public override Task PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) { throw new NotSupportedException(); } diff --git a/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs b/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs index f3a8383..58160fb 100644 --- a/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs +++ b/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs @@ -20,8 +20,6 @@ namespace Mercurio.Tests.Messaging { - using ErrorOr; - using Mercurio.Extensions; using Mercurio.Messaging; using Mercurio.Model; @@ -147,8 +145,7 @@ public async Task VerifyPushWithConfirmationAsync() using (Assert.EnterMultipleScope()) { - Assert.That(acknowledged.IsError, Is.False, "the RabbitMQ server should have acknowledged the message"); - Assert.That(acknowledged.Value, Is.EqualTo(Result.Success)); + Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged the message"); } var received = await Task.WhenAny(taskCompletion.Task, Task.Delay(2000)); @@ -158,15 +155,13 @@ public async Task VerifyPushWithConfirmationAsync() using (Assert.EnterMultipleScope()) { - Assert.That(unregistered.IsError, Is.True, "an unregistered connection should be reported instead of being swallowed"); - Assert.That(unregistered.FirstError.Code, Is.EqualTo(MessagingErrors.PublishFailedCode)); - Assert.That(unregistered.FirstError.Metadata[MessagingErrors.ExceptionMetadataKey], Is.InstanceOf()); + Assert.That(unregistered, Is.False, "an unregistered connection should be reported instead of being swallowed"); } using (Assert.EnterMultipleScope()) { - Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, (string)null, exchangeConfiguration), Throws.ArgumentNullException, "an invalid argument remains an exception"); - Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, FirstSentMessage, null), Throws.ArgumentNullException); + await Assert.ThatAsync(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, (string)null, exchangeConfiguration), Throws.ArgumentNullException, "an invalid argument remains an exception"); + await Assert.ThatAsync(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, FirstSentMessage, null), Throws.ArgumentNullException); } var rejectingConfiguration = await this.DeclareRejectPublishExchangeAsync("ConfirmationRejectExchange", "ConfirmationRejectQueue"); @@ -176,9 +171,7 @@ public async Task VerifyPushWithConfirmationAsync() using (Assert.EnterMultipleScope()) { - Assert.That(nacked.IsError, Is.True, "a message the server refuses should be reported"); - Assert.That(nacked.FirstError.Code, Is.EqualTo(MessagingErrors.NotAcknowledgedCode)); - Assert.That(nacked.FirstError.Metadata[MessagingErrors.ExceptionMetadataKey], Is.InstanceOf()); + Assert.That(nacked, Is.False, "a message the server refuses should be reported"); } } @@ -192,28 +185,22 @@ public async Task VerifyPushWithConfirmationMessagesAsync() IEnumerable messages = ["ABC", "DEF", "GHI"]; var acknowledged = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, exchangeConfiguration); - Assert.That(acknowledged.IsError, Is.False, "the RabbitMQ server should have acknowledged every message"); + Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged every message"); using (Assert.EnterMultipleScope()) { - Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, (IEnumerable)null, exchangeConfiguration), Throws.ArgumentException, "an invalid argument remains an exception"); - Assert.That(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, null), Throws.ArgumentNullException); + await Assert.ThatAsync(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, (IEnumerable)null, exchangeConfiguration), Throws.ArgumentException, "an invalid argument remains an exception"); + await Assert.ThatAsync(() => this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, null), Throws.ArgumentNullException); } var rejectingConfiguration = await this.DeclareRejectPublishExchangeAsync("ConfirmationCollectionRejectExchange", "ConfirmationCollectionRejectQueue"); var fillingTheQueue = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, "fills the queue", rejectingConfiguration); - Assert.That(fillingTheQueue.IsError, Is.False, "the rejecting queue accepts a single message before reaching its capacity"); + Assert.That(fillingTheQueue, Is.True, "the rejecting queue accepts a single message before reaching its capacity"); var refused = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, rejectingConfiguration); - using (Assert.EnterMultipleScope()) - { - Assert.That(refused.IsError, Is.True, "the batch should stop at the first refused message"); - Assert.That(refused.FirstError.Code, Is.EqualTo(MessagingErrors.NotAcknowledgedCode)); - Assert.That(refused.FirstError.Metadata[MessagingErrors.PublishedCountMetadataKey], Is.Zero, "the queue is already full, so nothing of the batch got through"); - Assert.That(refused.FirstError.Metadata[MessagingErrors.FailedIndexMetadataKey], Is.Zero); - } + Assert.That(refused, Is.False, "the queue is already full, so the batch stops at its very first message"); } /// @@ -472,12 +459,13 @@ public async Task VerifyActivities() } Activity.CurrentChanged -= ActivityOnCurrentChanged; - + return; + void ActivityOnCurrentChanged(object sender, ActivityChangedEventArgs e) { if (e.Current is { Source.Name: FirstConnectionName or SecondConnectionName }) { - activities.Add(e.Current!); + activities.Add(e.Current); } } } @@ -518,12 +506,13 @@ public async Task VerifyActivitiesMultipleMessages() } Activity.CurrentChanged -= ActivityOnCurrentChanged; - + return; + void ActivityOnCurrentChanged(object sender, ActivityChangedEventArgs e) { if (e.Current is { Source.Name: FirstConnectionName or SecondConnectionName }) { - activities.Add(e.Current!); + activities.Add(e.Current); } } } diff --git a/Mercurio/Mercurio.csproj b/Mercurio/Mercurio.csproj index 22f6a3f..037ecab 100644 --- a/Mercurio/Mercurio.csproj +++ b/Mercurio/Mercurio.csproj @@ -34,7 +34,6 @@ - diff --git a/Mercurio/Messaging/IMessageClientService.cs b/Mercurio/Messaging/IMessageClientService.cs index 94f9254..c4192b5 100644 --- a/Mercurio/Messaging/IMessageClientService.cs +++ b/Mercurio/Messaging/IMessageClientService.cs @@ -20,8 +20,6 @@ namespace Mercurio.Messaging { - using ErrorOr; - using Mercurio.Model; using RabbitMQ.Client; @@ -98,18 +96,18 @@ public interface IMessageClientService : IDisposable /// Possible action to configure additional properties /// A possible /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not /// /// When the provided or is null /// /// Contrary to , - /// any publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, only the - /// operational failures are reported as an , see for the reported ones. - /// Publisher confirmations throttle the amount of outstanding publications, so this is slower than a regular push. The - /// acknowledgment only asserts that the server took responsibility for the message, not that any consumer received it. + /// a publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, any + /// operational failure is logged and returns false. Publisher confirmations throttle the amount of outstanding publications, + /// so this is slower than a regular push. The acknowledgment only asserts that the server took responsibility for the + /// message, not that any consumer received it. /// - Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + Task PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); /// /// Pushes the specified to the specified queue via the @@ -122,19 +120,17 @@ public interface IMessageClientService : IDisposable /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged all the messages, + /// false as soon as one of them is not acknowledged /// /// When the provided collection is null /// When the provided is null /// - /// Invalid arguments still throw, only the operational failures are reported as an , see - /// for the reported ones. 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, the reported provides how many of them - /// under the key. + /// Invalid arguments still throw, any operational failure is logged and returns false. 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, the log reports how many of them. /// - Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + Task PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); /// /// Asynchronously leases a channel from the pool or creates one if necessary. diff --git a/Mercurio/Messaging/MessageClientBaseService.cs b/Mercurio/Messaging/MessageClientBaseService.cs index c4a5d25..e6c3eda 100644 --- a/Mercurio/Messaging/MessageClientBaseService.cs +++ b/Mercurio/Messaging/MessageClientBaseService.cs @@ -22,8 +22,6 @@ namespace Mercurio.Messaging { using System.Diagnostics; - using ErrorOr; - using Mercurio.Extensions; using Mercurio.Model; using Mercurio.Provider; @@ -139,18 +137,18 @@ protected MessageClientBaseService(IRabbitMqConnectionProvider connectionProvide /// Possible action to configure additional properties /// A possible /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not /// /// When the provided or is null /// /// Contrary to , - /// any publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, only the - /// operational failures are reported as an , see for the reported ones. - /// Publisher confirmations throttle the amount of outstanding publications, so this is slower than a regular push. The - /// acknowledgment only asserts that the server took responsibility for the message, not that any consumer received it. + /// a publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, any + /// operational failure is logged and returns false. Publisher confirmations throttle the amount of outstanding publications, + /// so this is slower than a regular push. The acknowledgment only asserts that the server took responsibility for the + /// message, not that any consumer received it. /// - public abstract Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + public abstract Task PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); /// /// Pushes the specified to the specified queue via the @@ -163,19 +161,17 @@ protected MessageClientBaseService(IRabbitMqConnectionProvider connectionProvide /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged all the messages, + /// false as soon as one of them is not acknowledged /// /// When the provided collection is null /// When the provided is null /// - /// Invalid arguments still throw, only the operational failures are reported as an , see - /// for the reported ones. 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, the reported provides how many of them - /// under the key. + /// Invalid arguments still throw, any operational failure is logged and returns false. 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, the log reports how many of them. /// - public abstract Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + public abstract Task PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. @@ -319,7 +315,10 @@ private Activity StartActivityInternal(ActivitySource activitySource, ActivityCo return null; } - this.Logger.LogTrace("{ActivityName} started!", activityName); + if (this.Logger.IsEnabled(LogLevel.Trace)) + { + this.Logger.LogTrace("{ActivityName} started!", activityName); + } return activity; } diff --git a/Mercurio/Messaging/MessageClientService.cs b/Mercurio/Messaging/MessageClientService.cs index 066bde6..b6123d6 100644 --- a/Mercurio/Messaging/MessageClientService.cs +++ b/Mercurio/Messaging/MessageClientService.cs @@ -27,8 +27,6 @@ namespace Mercurio.Messaging using CommunityToolkit.HighPerformance; - using ErrorOr; - using Mercurio.Extensions; using Mercurio.Model; using Mercurio.Provider; @@ -175,18 +173,18 @@ public override Task PushAsync(string connectionName, TMessage message /// Possible action to configure additional properties /// A possible /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not /// /// When the provided or is null /// /// Contrary to , - /// any publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, only the - /// operational failures are reported as an , see for the reported ones. - /// Publisher confirmations throttle the amount of outstanding publications, so this is slower than a regular push. The - /// acknowledgment only asserts that the server took responsibility for the message, not that any consumer received it. + /// a publication failure is reported to the caller instead of being logged only. Invalid arguments still throw, any + /// operational failure is logged and returns false. Publisher confirmations throttle the amount of outstanding publications, + /// so this is slower than a regular push. The acknowledgment only asserts that the server took responsibility for the + /// message, not that any consumer received it. /// - public override Task> PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + public override Task PushWithConfirmationAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) { if (Equals(message, default(TMessage))) { @@ -215,19 +213,17 @@ public override Task> PushWithConfirmationAsync(strin /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged all the messages, + /// false as soon as one of them is not acknowledged /// /// When the provided collection is null /// When the provided is null /// - /// Invalid arguments still throw, only the operational failures are reported as an , see - /// for the reported ones. 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, the reported provides how many of them - /// under the key. + /// Invalid arguments still throw, any operational failure is logged and returns false. 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, the log reports how many of them. /// - public override Task> PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + public override Task PushWithConfirmationAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) { if (messages == null) { @@ -299,7 +295,10 @@ private async Task AddListenerInternalAsync(string connectionName, } catch (Exception exception) { - this.Logger.LogError(exception, "Error while adding a listener to the {QueueName}", exchangeConfiguration.QueueName); + if (this.Logger.IsEnabled(LogLevel.Error)) + { + this.Logger.LogError(exception, "Error while adding a listener to the {QueueName}", exchangeConfiguration.QueueName); + } } return Disposable.Create(() => @@ -397,7 +396,10 @@ private async Task PushInternalAsync(string connectionName, TMessage m } catch (Exception exception) { - this.Logger.LogError(exception, "The message {MessageName} could not be queued to {MessageQueue} reason : {Exception}", typeof(TMessage).Name, exchangeConfiguration.QueueName, exception.Message); + if (this.Logger.IsEnabled(LogLevel.Error)) + { + this.Logger.LogError(exception, "The message {MessageName} could not be queued to {MessageQueue} reason : {Exception}", typeof(TMessage).Name, exchangeConfiguration.QueueName, exception.Message); + } } } @@ -412,14 +414,14 @@ private async Task PushInternalAsync(string connectionName, TMessage m /// Possible action to configure additional properties /// An optional /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged all the messages, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged all the messages, + /// false as soon as one of them is not acknowledged /// /// /// The process stops at the first message that is not acknowledged, so that a broken connection does not have every remaining /// message pay the connection retry backoff /// - private async Task> PushMultipleWithConfirmationInternalAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties, CancellationToken cancellationToken) + private async Task PushMultipleWithConfirmationInternalAsync(string connectionName, IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties, CancellationToken cancellationToken) { var activitySource = this.ConnectionProvider.GetRegisteredActivitySource(connectionName); @@ -433,21 +435,21 @@ private async Task> PushMultipleWithConfirmationInternalAsync /// Publishes the specified to the specified queue via the - /// , waiting for the RabbitMQ server to acknowledge it and reporting any failure as - /// an instead of throwing + /// , waiting for the RabbitMQ server to acknowledge it and logging any failure + /// instead of throwing /// /// The type of message /// The name of the registered connection to use. @@ -461,25 +463,33 @@ private async Task> PushMultipleWithConfirmationInternalAsyncAn optional . If not set, current context will be based on /// A possible /// - /// An awaitable of that provides once the - /// RabbitMQ server has acknowledged the message, or the that describes the failure + /// An awaitable that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not /// - private async Task> TryPublishAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties, string activityName, ActivityContext activityContext, CancellationToken cancellationToken) + private async Task TryPublishAsync(string connectionName, TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties, string activityName, ActivityContext activityContext, CancellationToken cancellationToken) { try { await this.PublishInternalAsync(connectionName, message, exchangeConfiguration, configureProperties, activityName, activityContext, true, cancellationToken); - return Result.Success; + return true; } catch (PublishException publishException) { - this.Logger.LogWarning(publishException, "The message {MessageName} has not been acknowledged by the RabbitMQ server for {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); - return MessagingErrors.NotAcknowledged(publishException); + if (this.Logger.IsEnabled(LogLevel.Warning)) + { + this.Logger.LogWarning(publishException, "The message {MessageName} has not been acknowledged by the RabbitMQ server for {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); + } + + return false; } catch (Exception exception) { - this.Logger.LogWarning(exception, "The message {MessageName} could not be published to {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); - return MessagingErrors.PublishFailed(exception); + if (this.Logger.IsEnabled(LogLevel.Warning)) + { + this.Logger.LogWarning(exception, "The message {MessageName} could not be published to {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); + } + + return false; } } @@ -556,7 +566,10 @@ await channelLease.Channel.BasicPublishAsync(exchangeConfiguration.PushExchangeN this.Logger.LogDebug("Message Body {Body}", Encoding.UTF8.GetString(body.ToArray())); } - this.Logger.LogInformation("Message {MessageName} sent to {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); + if (this.Logger.IsEnabled(LogLevel.Information)) + { + this.Logger.LogInformation("Message {MessageName} sent to {MessageQueue}", typeof(TMessage).Name, exchangeConfiguration.QueueName); + } activity?.SetStatus(ActivityStatusCode.Ok); } catch (Exception exception) @@ -633,7 +646,10 @@ async Task ConsumerOnReceivedAsync(object o, BasicDeliverEventArgs message) { var invalidMessageException = new InvalidMessageException(typeof(TMessage), message, deserializationException); - this.Logger.LogError(deserializationException, "The message received on {QueueName} could not be deserialized into a {MessageName}", exchangeConfiguration.QueueName, typeof(TMessage).Name); + if (this.Logger.IsEnabled(LogLevel.Error)) + { + this.Logger.LogError(deserializationException, "The message received on {QueueName} could not be deserialized into a {MessageName}", exchangeConfiguration.QueueName, typeof(TMessage).Name); + } activity?.SetStatus(ActivityStatusCode.Error, invalidMessageException.Message); observer.OnError(invalidMessageException); return; diff --git a/Mercurio/Messaging/MessagingErrors.cs b/Mercurio/Messaging/MessagingErrors.cs deleted file mode 100644 index f522da3..0000000 --- a/Mercurio/Messaging/MessagingErrors.cs +++ /dev/null @@ -1,134 +0,0 @@ -// ------------------------------------------------------------------------------------------------- -// -// -// Copyright 2025 - 2026 Starion Group S.A. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. -// -// -// ------------------------------------------------------------------------------------------------ - -namespace Mercurio.Messaging -{ - using ErrorOr; - - using RabbitMQ.Client.Exceptions; - - /// - /// Provides all that can be reported by the when publishing a message - /// with publisher confirmations enabled - /// - /// - /// The does not carry an , so the original is provided - /// through the , under the key - /// - public static class MessagingErrors - { - /// - /// The key that provides the at the origin of the - /// - public const string ExceptionMetadataKey = "exception"; - - /// - /// The key that provides the amount of messages that have been acknowledged before a batch failed - /// - public const string PublishedCountMetadataKey = "publishedCount"; - - /// - /// The key that provides the index of the message that failed within a batch - /// - public const string FailedIndexMetadataKey = "failedIndex"; - - /// - /// The reported when the RabbitMQ server did not acknowledge the published message - /// - public const string NotAcknowledgedCode = "Mercurio.Publish.NotAcknowledged"; - - /// - /// The reported when the message could not be published at all - /// - public const string PublishFailedCode = "Mercurio.Publish.Failed"; - - /// - /// The reported when the connection that should be used is not registered - /// - public const string ConnectionNotRegisteredCode = "Mercurio.Publish.ConnectionNotRegistered"; - - /// - /// Provides the that reports that the RabbitMQ server did not acknowledge the published message - /// - /// The that reports the negative acknowledgment - /// The - public static Error NotAcknowledged(PublishException exception) - { - return Error.Failure(NotAcknowledgedCode, $"The RabbitMQ server did not acknowledge the message: {exception?.Message}", - new Dictionary - { - { ExceptionMetadataKey, exception }, - { "isReturn", exception?.IsReturn }, - { "publishSequenceNumber", exception?.PublishSequenceNumber } - }); - } - - /// - /// Provides the that reports that the message could not be published - /// - /// The that occured while publishing - /// The - public static Error PublishFailed(Exception exception) - { - return Error.Unexpected(PublishFailedCode, $"The message could not be published: {exception?.Message}", - new Dictionary - { - { ExceptionMetadataKey, exception } - }); - } - - /// - /// Provides the that reports that the connection that should be used is not registered - /// - /// The name of the connection that is not registered - /// The that reports the missing registration - /// The - public static Error ConnectionNotRegistered(string connectionName, Exception exception) - { - return Error.Failure(ConnectionNotRegisteredCode, $"No connection has been registered under the name {connectionName}", - new Dictionary - { - { ExceptionMetadataKey, exception } - }); - } - - /// - /// Provides the that reports that a message of a batch could not be published, keeping the - /// , and of the reported - /// - /// - /// The that has been reported for the failing message - /// The amount of messages that have been acknowledged before the failure - /// The index, within the batch, of the message that failed - /// The - /// - /// Since publishing a batch is not transactional, the messages that have been acknowledged before the failure are already held - /// by the RabbitMQ server - /// - public static Error BatchFailed(Error error, int publishedCount, int failedIndex) - { - var metadata = error.Metadata == null ? new Dictionary() : new Dictionary(error.Metadata); - metadata[PublishedCountMetadataKey] = publishedCount; - metadata[FailedIndexMetadataKey] = failedIndex; - - return Error.Custom(error.NumericType, error.Code, error.Description, metadata); - } - } -} diff --git a/Mercurio/Messaging/RpcServerService.cs b/Mercurio/Messaging/RpcServerService.cs index 08d79e4..eab357f 100644 --- a/Mercurio/Messaging/RpcServerService.cs +++ b/Mercurio/Messaging/RpcServerService.cs @@ -125,7 +125,10 @@ private async Task ListenForRequestInternalAsync diff --git a/Mercurio/Provider/RabbitMqConnectionProvider.cs b/Mercurio/Provider/RabbitMqConnectionProvider.cs index be7f07d..7a97192 100644 --- a/Mercurio/Provider/RabbitMqConnectionProvider.cs +++ b/Mercurio/Provider/RabbitMqConnectionProvider.cs @@ -374,7 +374,13 @@ private async Task CreateChannelWithRetryAsync(string connectionName, var result = await Policy .Handle() .WaitAndRetryAsync(this.retryPolicyConfiguration.MaxConnectionRetryAttempts, attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)), - (exception, delay, attemptNumber, _) => this.logger.LogWarning(exception, "ChannelLease creation attempt {Attempt} failed. Retrying in {Delay}", attemptNumber, delay)) + (exception, delay, attemptNumber, _) => + { + if (this.logger.IsEnabled(LogLevel.Warning)) + { + this.logger.LogWarning(exception, "ChannelLease creation attempt {Attempt} failed. Retrying in {Delay}", attemptNumber, delay); + } + }) .ExecuteAndCaptureAsync(() => this.CreateChannelAsync(connectionName, publisherConfirmationsEnabled, cancellationToken)); if (result.Outcome != OutcomeType.Successful) diff --git a/README.md b/README.md index e1b0d72..db769af 100644 --- a/README.md +++ b/README.md @@ -107,18 +107,15 @@ await messageClientService.PushAsync("RegisteredConnection","A message to be sen #### Push Message And Wait For The Server Acknowledgment _PushAsync_ is fire and forget : it returns as soon as the message has been handed to the transport, and any failure is only logged. When you need to know that the RabbitMQ server took responsibility for the message, use _PushWithConfirmationAsync_. It publishes on a -channel that has publisher confirmations enabled and reports the outcome as an `ErrorOr` instead of throwing. +channel that has publisher confirmations enabled and returns whether the server acknowledged the message. ```csharp var messageClientService = serviceProvider.GetRequiredService(); var exchangeConfiguration = new DirectExchangeConfiguration("DirectQueue", "AnExchange", "SomeRouting"); -var result = await messageClientService.PushWithConfirmationAsync("RegisteredConnection", "A message to be sent", exchangeConfiguration); -if (result.IsError) +if (!await messageClientService.PushWithConfirmationAsync("RegisteredConnection", "A message to be sent", exchangeConfiguration)) { - // MessagingErrors exposes every reported code, and the originating exception is available in the metadata - Console.WriteLine($"{result.FirstError.Code}: {result.FirstError.Description}"); - var exception = (Exception)result.FirstError.Metadata[MessagingErrors.ExceptionMetadataKey]; + // the reason has been logged, including the exception the RabbitMQ client reported } ``` @@ -128,7 +125,7 @@ Two things to keep in mind : - An acknowledgment only asserts that the **server** took responsibility for the message, never that a consumer received it. A message published to an exchange that routes to no queue is legitimately acknowledged. -Invalid arguments still throw : only the operational failures are reported as an `Error`. +Invalid arguments still throw : only the operational failures are logged and reported as `false`. #### Listen After Message ```csharp