diff --git a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs index 805de8f..9d3d77d 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,67 @@ public async Task VerifyPushMessageAsync() Assert.That(this.backgroundService.ReceivedMessages, Does.Contain("ABC")); } + [Test] + public async Task VerifyPushMessageWithConfirmationAsync() + { + using var cancellationTokenSource = new CancellationTokenSource(); + _ = this.backgroundService.StartAsync(cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync"); + + var acknowledged = await this.backgroundService.PushMessageWithConfirmationAsync("ABC", exchange, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + using (Assert.EnterMultipleScope()) + { + Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged the message"); + Assert.That(this.backgroundService.ReceivedMessages, Does.Contain("ABC")); + } + + var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), this.configurationMock.Object); + var notRegistered = await faultyService.PushMessageWithConfirmationAsync("XYZ", exchange, cancellationToken: cancellationTokenSource.Token); + faultyService.Dispose(); + + Assert.That(notRegistered, Is.False, "a service that has not been initialized has no registered connection, which is reported instead of being thrown"); + + await Assert.ThatAsync(() => this.backgroundService.PushMessageWithConfirmationAsync((string)null!, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentNullException, "an invalid argument remains an exception"); + + await cancellationTokenSource.CancelAsync(); + } + + [Test] + public async Task VerifyPushMessagesWithConfirmationAsync() + { + using var cancellationTokenSource = new CancellationTokenSource(); + _ = this.backgroundService.StartAsync(cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync"); + + // the cast is required: an array also binds to the single message overload, with TMessage being the array itself + IEnumerable messages = ["ABC", "DEF", "GHI"]; + + var acknowledged = await this.backgroundService.PushMessagesWithConfirmationAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(800), CancellationToken.None); + + using (Assert.EnterMultipleScope()) + { + Assert.That(acknowledged, Is.True, "the RabbitMQ server should have acknowledged every message"); + Assert.That(this.backgroundService.ReceivedMessages, Is.EquivalentTo(messages)); + } + + var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), this.configurationMock.Object); + var notRegistered = await faultyService.PushMessagesWithConfirmationAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token); + faultyService.Dispose(); + + Assert.That(notRegistered, Is.False, "a service that has not been initialized has no registered connection, which is reported instead of being thrown"); + + await Assert.ThatAsync(() => this.backgroundService.PushMessagesWithConfirmationAsync((IEnumerable)null!, exchange, cancellationToken: cancellationTokenSource.Token), Throws.ArgumentException, "an invalid argument remains an exception"); + + await cancellationTokenSource.CancelAsync(); + } + [Test] public async Task VerifyPushMessagesAsync() { @@ -302,8 +366,11 @@ public async Task VerifyAutoRecoverySystem() using var cancellationTokenSource = new CancellationTokenSource(); var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), this.configurationMock.Object); _ = autoRecoveryBackground.StartAsync(cancellationTokenSource.Token); - await Task.Delay(TimeSpan.FromMilliseconds(100), CancellationToken.None); - + + // a fanout message published before the listener has bound its queue is silently discarded, so the subscriptions have to be + // established before pushing anything + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTest"); await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token); await Task.Delay(100, cancellationTokenSource.Token); @@ -327,8 +394,11 @@ public async Task VerifyAsyncAutoRecoverySystem() using var cancellationTokenSource = new CancellationTokenSource(); var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), 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..f8b0043 100644 --- a/Mercurio.Hosting/IMessagingBackgroundService.cs +++ b/Mercurio.Hosting/IMessagingBackgroundService.cs @@ -102,5 +102,50 @@ 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 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, 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); + + /// + /// 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 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. + /// + 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..c03f156 100644 --- a/Mercurio.Hosting/MessagingBackgroundService.cs +++ b/Mercurio.Hosting/MessagingBackgroundService.cs @@ -208,6 +208,102 @@ 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 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, 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) + { + var activityName = $"Background Push with confirmation {typeof(TMessage).Name} [{exchangeConfiguration}]"; + + if (!this.TryGetRegisteredActivitySource(out var activitySource)) + { + return false; + } + + using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); + return await this.MessageClientService.PushWithConfirmationAsync(this.ConnectionName, message, exchangeConfiguration, configureProperties, cancellationToken); + } + + /// + /// 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 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. + /// + 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)) + { + return false; + } + + using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); + return await this.MessageClientService.PushWithConfirmationAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken); + } + + /// + /// Tries to resolve the that is registered for the current + /// + /// The resolved , if any + /// 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 logs operational + /// failures and returns false, that exception is converted here. + /// + private bool TryGetRegisteredActivitySource(out ActivitySource activitySource) + { + try + { + activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName); + return true; + } + catch (ArgumentException exception) + { + if (this.Logger.IsEnabled(LogLevel.Warning)) + { + this.Logger.LogWarning(exception, "No connection has been registered under the name {ConnectionName}", this.ConnectionName); + } + + activitySource = null; + return false; + } + } + /// /// 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..1e890cf 100644 --- a/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs +++ b/Mercurio.Tests/Messaging/MessageClientBaseServiceTestFixture.cs @@ -167,5 +167,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 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(); + } + + /// + /// 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 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 1513e42..58160fb 100644 --- a/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs +++ b/Mercurio.Tests/Messaging/MessageClientServiceTestFixture.cs @@ -28,6 +28,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 +130,104 @@ 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, Is.True, "the RabbitMQ server should have acknowledged the message"); + } + + 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, Is.False, "an unregistered connection should be reported instead of being swallowed"); + } + + using (Assert.EnterMultipleScope()) + { + 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"); + + await this.firstService.PushWithConfirmationAsync(FirstConnectionName, FirstSentMessage, rejectingConfiguration); + var nacked = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, SecondSentMessage, rejectingConfiguration); + + using (Assert.EnterMultipleScope()) + { + Assert.That(nacked, Is.False, "a message the server refuses should be reported"); + } + } + + [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, Is.True, "the RabbitMQ server should have acknowledged every message"); + + using (Assert.EnterMultipleScope()) + { + 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, Is.True, "the rejecting queue accepts a single message before reaching its capacity"); + + var refused = await this.firstService.PushWithConfirmationAsync(FirstConnectionName, messages, rejectingConfiguration); + + Assert.That(refused, Is.False, "the queue is already full, so the batch stops at its very first message"); + } + + /// + /// 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) @@ -360,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); } } } @@ -406,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.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..037ecab 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,7 +28,7 @@ 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..c4192b5 100644 --- a/Mercurio/Messaging/IMessageClientService.cs +++ b/Mercurio/Messaging/IMessageClientService.cs @@ -85,6 +85,53 @@ 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 that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not + /// + /// When the provided or is null + /// + /// Contrary to , + /// 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); + + /// + /// 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 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, 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); + /// /// Asynchronously leases a channel from the pool or creates one if necessary. /// @@ -92,5 +139,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..e6c3eda 100644 --- a/Mercurio/Messaging/MessageClientBaseService.cs +++ b/Mercurio/Messaging/MessageClientBaseService.cs @@ -126,6 +126,53 @@ 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 that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not + /// + /// When the provided or is null + /// + /// Contrary to , + /// 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); + + /// + /// 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 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, 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); + /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// @@ -146,6 +193,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. /// @@ -253,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 8f84fc4..b6123d6 100644 --- a/Mercurio/Messaging/MessageClientService.cs +++ b/Mercurio/Messaging/MessageClientService.cs @@ -36,6 +36,7 @@ namespace Mercurio.Messaging using RabbitMQ.Client; using RabbitMQ.Client.Events; + using RabbitMQ.Client.Exceptions; /// /// The is the concrete implementation of the @@ -78,7 +79,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 +97,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 +162,82 @@ 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 that provides true once the RabbitMQ server has acknowledged the message, false + /// if it has not + /// + /// When the provided or is null + /// + /// Contrary to , + /// 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) + { + 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 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, 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) + { + 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. /// @@ -218,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(() => @@ -309,12 +389,143 @@ 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) + { + 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); + } + } + } + + /// + /// 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 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) + { + 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 isAcknowledged = await this.TryPublishAsync(connectionName, messagesList[messageIndex], exchangeConfiguration, configureProperties, subActivityName, context, cancellationToken); + + if (!isAcknowledged) + { + return false; + } + } + + return true; + } + + /// + /// Publishes the specified to the specified queue via the + /// , 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. + /// 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 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) + { + try + { + await this.PublishInternalAsync(connectionName, message, exchangeConfiguration, configureProperties, activityName, activityContext, true, cancellationToken); + return true; + } + catch (PublishException 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) + { + 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; + } + } + + /// + /// 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 +561,21 @@ await channelLease.Channel.BasicPublishAsync(exchangeConfiguration.PushExchangeN routingKey, false, properties, body, cancellationToken); - 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.Debug)) + { + this.Logger.LogDebug("Message Body {Body}", Encoding.UTF8.GetString(body.ToArray())); + } + + 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) { - 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 { @@ -428,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/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/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..7a97192 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,19 @@ 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)); + (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) { @@ -341,9 +409,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 +425,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..db769af 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,29 @@ 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 returns whether the server acknowledged the message. + +```csharp +var messageClientService = serviceProvider.GetRequiredService(); +var exchangeConfiguration = new DirectExchangeConfiguration("DirectQueue", "AnExchange", "SomeRouting"); + +if (!await messageClientService.PushWithConfirmationAsync("RegisteredConnection", "A message to be sent", exchangeConfiguration)) +{ + // the reason has been logged, including the exception the RabbitMQ client reported +} +``` + +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 logged and reported as `false`. + #### Listen After Message ```csharp var messageClientService = serviceProvider.GetRequiredService();