diff --git a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs index fe16b65..6b1063e 100644 --- a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs +++ b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs @@ -126,6 +126,55 @@ public async Task VerifyBackgroundServiceBehaviourPushMultiple() } } + [Test] + public async Task VerifyPushMessageAsync() + { + using var cancellationTokenSource = new CancellationTokenSource(); + _ = this.backgroundService.StartAsync(cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync"); + + await this.backgroundService.PushMessageAsync("ABC", exchange, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), this.configurationMock.Object); + + await Assert.ThatAsync(() => faultyService.PushMessageAsync("XYZ", exchange, cancellationToken: cancellationTokenSource.Token), Throws.Exception); + faultyService.Dispose(); + + await cancellationTokenSource.CancelAsync(); + + Assert.That(this.backgroundService.ReceivedMessages, Does.Contain("ABC")); + } + + [Test] + public async Task VerifyPushMessagesAsync() + { + using var cancellationTokenSource = new CancellationTokenSource(); + _ = this.backgroundService.StartAsync(cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + var exchange = new FanoutExchangeConfiguration("BackgroundTestPushAsync"); + string[] messages = ["ABC", "DEF", "GHI"]; + + await this.backgroundService.PushMessagesAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(800), CancellationToken.None); + + var faultyService = new TestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), this.configurationMock.Object); + + await Assert.ThatAsync(() => faultyService.PushMessagesAsync(messages, exchange, cancellationToken: cancellationTokenSource.Token), Throws.Exception); + faultyService.Dispose(); + + await cancellationTokenSource.CancelAsync(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(this.backgroundService.ReceivedMessages, Has.Count.EqualTo(3)); + Assert.That(this.backgroundService.ReceivedMessages, Is.EquivalentTo(messages)); + } + } + [Test] public async Task VerifyBackgroundServiceBehaviourTaskCanceled() { @@ -335,8 +384,9 @@ protected override async Task InitializeAsync() { this.ConnectionName = ConfiguredConnectionName; await this.RegisterListener(() => this.MessageClientService.ListenAsync(this.ConnectionName, new FanoutExchangeConfiguration("BackgroundTest")), this.ReceivedMessages.Add, onError: _ => this.ReceivedMessages.Clear()); - - await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync(this.ConnectionName, new DirectExchangeConfiguration("BackgroundTestInt")), (x) => + await this.RegisterListener(() => this.MessageClientService.ListenAsync(this.ConnectionName, new FanoutExchangeConfiguration("BackgroundTestPushAsync")), this.ReceivedMessages.Add, onError: _ => this.ReceivedMessages.Clear()); + + await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync(this.ConnectionName, new DirectExchangeConfiguration("BackgroundTestInt")), (x) => { this.ReceivedMessages.Add(x.ToString()); return Task.CompletedTask; diff --git a/Mercurio.Hosting/IMessagingBackgroundService.cs b/Mercurio.Hosting/IMessagingBackgroundService.cs index d02cbe2..968d436 100644 --- a/Mercurio.Hosting/IMessagingBackgroundService.cs +++ b/Mercurio.Hosting/IMessagingBackgroundService.cs @@ -68,5 +68,39 @@ public interface IMessagingBackgroundService /// as 'application/json" /// void PushMessages(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + + /// + /// Asynchronously pushes the specified to the specified queue via the + /// , bypassing the internal background queue so the caller can + /// await completion and observe any publishing errors. + /// + /// 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 completes when the message has been published + /// + /// By default, the is configured to use the mode and sets the + /// as 'application/json" + /// + Task PushMessageAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default); + + /// + /// Asynchronously pushes the specified to the specified queue via the + /// , bypassing the internal background queue so the caller can + /// await completion and observe any publishing errors. + /// + /// 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 completes when the messages have been published + /// + /// By default, the is configured to use the mode and sets the + /// as 'application/json" + /// + Task PushMessagesAsync(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 dc82f45..ab79e7e 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.4 + 2.0.5 12.0 Starion Group S.A. Copyright © Starion Group S.A. @@ -28,9 +28,7 @@ true true - [Update] Microsoft.Extensions.Diagnostics.HealthChecks.Abstractions to version 10.0.7 - [Update] Microsoft.Extensions.Hosting to version 10.0.7 - [Update] Mercurio to version 2.0.3 + [Add] PushMessageAsync and PushMessagesAsync overloads on IMessagingBackgroundService / MessagingBackgroundService to allow awaiting the publish and observing errors diff --git a/Mercurio.Hosting/MessagingBackgroundService.cs b/Mercurio.Hosting/MessagingBackgroundService.cs index b78be14..2c13722 100644 --- a/Mercurio.Hosting/MessagingBackgroundService.cs +++ b/Mercurio.Hosting/MessagingBackgroundService.cs @@ -162,6 +162,52 @@ public void PushMessages(IEnumerable messages, IExchangeConf this.messagesToPush.Enqueue((() => this.MessageClientService.PushAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken), activityContext, activityName)); } + /// + /// Asynchronously pushes the specified to the specified queue via the + /// , bypassing the internal background queue so the caller can + /// await completion and observe any publishing errors. + /// + /// 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 completes when the message has been published + /// + /// By default, the is configured to use the mode and sets the + /// as 'application/json" + /// + public async Task PushMessageAsync(TMessage message, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + var activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName); + var activityName = $"Background Push {typeof(TMessage).Name} [{exchangeConfiguration}]"; + using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); + await this.MessageClientService.PushAsync(this.ConnectionName, message, exchangeConfiguration, configureProperties, cancellationToken); + } + + /// + /// Asynchronously pushes the specified to the specified queue via the + /// , bypassing the internal background queue so the caller can + /// await completion and observe any publishing errors. + /// + /// 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 completes when the messages have been published + /// + /// By default, the is configured to use the mode and sets the + /// as 'application/json" + /// + public async Task PushMessagesAsync(IEnumerable messages, IExchangeConfiguration exchangeConfiguration, Action configureProperties = null, CancellationToken cancellationToken = default) + { + var activitySource = this.connectionProvider.GetRegisteredActivitySource(this.ConnectionName); + var activityName = $"Background Push Multiple {typeof(TMessage).Name} [{exchangeConfiguration}]"; + using var activity = activitySource?.StartActivity(activityName, ActivityKind.Producer, Activity.Current?.Context ?? default); + await this.MessageClientService.PushAsync(this.ConnectionName, messages, exchangeConfiguration, configureProperties, cancellationToken); + } + /// /// Triggered when the application host is ready to start the service. /// diff --git a/Mercurio/Mercurio.csproj b/Mercurio/Mercurio.csproj index 1ba3f80..bc9b3e6 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.3 + 2.0.5 12.0 Starion Group S.A. Copyright © Starion Group S.A. @@ -28,12 +28,7 @@ true true - [Fix] Don't allow transient_nonexcl_queues. Deprecated since RabbitMQ version 4.x - [Update] CommunityToolkit.HighPerformance to version 8.4.2 - [Update] Microsoft.Extensions.DependencyInjection.Abstractions to version 10.0.7 - [Update] Polly to version 8.6.6 - [Update] RabbitMQ.Client to version 7.2.1 - [Update] System.Text.Json to version 10.0.7 + [Bump] For version align