diff --git a/.gitignore b/.gitignore index e956b9a..dae1143 100644 --- a/.gitignore +++ b/.gitignore @@ -329,3 +329,5 @@ ASALocalRun/ .mfractor/ uml4net.xmi.Extensions.EnterpriseArchitect.Tests/Resources/UML_EA.DTD + +settings.local.json diff --git a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs index 6b1063e..805de8f 100644 --- a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs +++ b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs @@ -20,6 +20,9 @@ namespace Mercurio.Tests { + using System.Text.Json; + + using Mercurio.Configuration.SerializationConfiguration; using Mercurio.Extensions; using Mercurio.Hosting; using Mercurio.Messaging; @@ -40,6 +43,8 @@ public class MessagingBackgroundServiceTestFixture { private const string ConfiguredConnectionName = "RabbitMQ"; private const string RpcServerQueueName = "RPC"; + private const string InvalidMessageExchangeName = "BackgroundTestInvalidMessage"; + private const string InvalidMessageAsyncExchangeName = "BackgroundTestInvalidMessageAsync"; private Mock configurationMock; private TestMessagingBackgroundService backgroundService; private ServiceProvider serviceProvider; @@ -90,7 +95,7 @@ public async Task VerifyBackgroundServiceBehaviour() foreach (var message in messages) { - this.backgroundService.PushMessage(message,new FanoutExchangeConfiguration("BackgroundTest"), cancellationToken: cancellationTokenSource.Token); + await this.backgroundService.PushMessageAsync(message,new FanoutExchangeConfiguration("BackgroundTest"), cancellationToken: cancellationTokenSource.Token); await Task.Delay(TimeSpan.FromMilliseconds(300), cancellationTokenSource.Token); } @@ -114,7 +119,7 @@ public async Task VerifyBackgroundServiceBehaviourPushMultiple() string[] messages = ["ABC", "DEF", "GHI"]; - this.backgroundService.PushMessages(messages,new FanoutExchangeConfiguration("BackgroundTest"), cancellationToken: cancellationTokenSource.Token); + await this.backgroundService.PushMessagesAsync(messages,new FanoutExchangeConfiguration("BackgroundTest"), cancellationToken: cancellationTokenSource.Token); await Task.Delay(TimeSpan.FromMilliseconds(1200), CancellationToken.None); await cancellationTokenSource.CancelAsync(); @@ -183,7 +188,7 @@ public async Task VerifyBackgroundServiceBehaviourTaskCanceled() string[] messages = ["ABC", "DEF", "GHI"]; - this.backgroundService.PushMessages(messages,new FanoutExchangeConfiguration("BackgroundTest"), cancellationToken: cancellationTokenSource.Token); + await this.backgroundService.PushMessagesAsync(messages,new FanoutExchangeConfiguration("BackgroundTest"), cancellationToken: cancellationTokenSource.Token); await cancellationTokenSource.CancelAsync(); await Task.Delay(TimeSpan.FromMilliseconds(1000), CancellationToken.None); @@ -191,6 +196,69 @@ public async Task VerifyBackgroundServiceBehaviourTaskCanceled() Assert.That(this.backgroundService.ReceivedMessages, Is.Empty); } + [Test] + public async Task VerifyInvalidMessageDeserialization() + { + using var cancellationTokenSource = new CancellationTokenSource(); + + var invalidMessageService = new InvalidMessageBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService>(), this.configurationMock.Object); + _ = invalidMessageService.StartAsync(cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + var exchangeConfiguration = new FanoutExchangeConfiguration(InvalidMessageExchangeName); + var messageClientService = this.serviceProvider.GetRequiredService(); + + await messageClientService.PushAsync(ConfiguredConnectionName, 42, exchangeConfiguration, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + using (Assert.EnterMultipleScope()) + { + Assert.That(invalidMessageService.ReceivedMessages, Is.EquivalentTo([42])); + Assert.That(invalidMessageService.Errors, Is.Empty); + } + + await messageClientService.PushAsync(ConfiguredConnectionName, "not a number", exchangeConfiguration, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + using (Assert.EnterMultipleScope()) + { + Assert.That(invalidMessageService.ReceivedMessages, Is.EquivalentTo([42]), "an invalid message is not delivered to the listener"); + Assert.That(invalidMessageService.Errors, Has.Count.EqualTo(1), "the deserialization failure is reported to the onError callback"); + Assert.That(invalidMessageService.Errors[0], Is.InstanceOf(), "the deserialization failure is reported as an InvalidMessageException"); + } + + var invalidMessageException = (InvalidMessageException)invalidMessageService.Errors[0]; + + using (Assert.EnterMultipleScope()) + { + Assert.That(invalidMessageException.MessageType, Is.EqualTo(typeof(int))); + Assert.That(invalidMessageException.InnerException, Is.InstanceOf()); + Assert.That(invalidMessageException.GetBodyAsString(), Is.EqualTo("\"not a number\"")); + Assert.That(invalidMessageException.Exchange, Is.EqualTo(InvalidMessageExchangeName)); + Assert.That(invalidMessageException.ContentType, Is.EqualTo(SupportedSerializationFormat.JsonFormat)); + } + + await messageClientService.PushAsync(ConfiguredConnectionName, 43, exchangeConfiguration, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + Assert.That(invalidMessageService.ReceivedMessages, Does.Contain(43), "the listener keeps on listening after an invalid message"); + + var asyncExchangeConfiguration = new FanoutExchangeConfiguration(InvalidMessageAsyncExchangeName); + + await messageClientService.PushAsync(ConfiguredConnectionName, "not a number", asyncExchangeConfiguration, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + Assert.That(invalidMessageService.AsyncErrors, Has.Exactly(1).InstanceOf(), "the deserialization failure is also reported on an async listener"); + + await messageClientService.PushAsync(ConfiguredConnectionName, 44, asyncExchangeConfiguration, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken.None); + + Assert.That(invalidMessageService.AsyncReceivedMessages, Does.Contain(44), "the async listener keeps on listening after an invalid message"); + + await cancellationTokenSource.CancelAsync(); + invalidMessageService.Dispose(); + } + [Test] public async Task VerifyInvalidInitialization() { @@ -237,16 +305,16 @@ public async Task VerifyAutoRecoverySystem() await Task.Delay(TimeSpan.FromMilliseconds(100), CancellationToken.None); var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTest"); - autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token); await Task.Delay(100, cancellationTokenSource.Token); Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(1)); - autoRecoveryBackground.PushMessage("123", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("123", configuration, cancellationToken: cancellationTokenSource.Token); await Task.Delay(100, cancellationTokenSource.Token); - autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); - autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token); await Task.Delay(100, cancellationTokenSource.Token); Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(3)); @@ -262,16 +330,16 @@ public async Task VerifyAsyncAutoRecoverySystem() await Task.Delay(TimeSpan.FromMilliseconds(200), CancellationToken.None); var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTestAsync"); - autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token); await Task.Delay(100, cancellationTokenSource.Token); Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(1)); - autoRecoveryBackground.PushMessage("123", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("123", configuration, cancellationToken: cancellationTokenSource.Token); await Task.Delay(50, cancellationTokenSource.Token); - autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); - autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await autoRecoveryBackground.PushMessageAsync("abc", configuration, cancellationToken: cancellationTokenSource.Token); await Task.Delay(50, cancellationTokenSource.Token); Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(3)); @@ -304,6 +372,53 @@ protected override Task InitializeAsync() } } + private class InvalidMessageBackgroundService : MessagingBackgroundService + { + /// + /// Stores all received messages + /// + public readonly List ReceivedMessages = []; + + /// + /// Stores all s that have been notified through the onError callback + /// + public readonly List Errors = []; + + public readonly List AsyncReceivedMessages = []; + + public readonly List AsyncErrors = []; + + /// + /// Initializes a new instance of the + /// + /// + /// The injected that allow to resolve + /// instance, even if not registered as scope + /// + /// The injected to allow logging + /// The injected to provides configuration information for service initialization + public InvalidMessageBackgroundService(IServiceProvider serviceProvider, ILogger logger, IConfiguration configuration) : base(serviceProvider, logger, configuration) + { + } + + /// + /// Initializes this service (e.g. to set the and register subscriptions + /// collection + /// + /// An awaitable + protected override async Task InitializeAsync() + { + this.ConnectionName = ConfiguredConnectionName; + await this.RegisterListener(() => this.MessageClientService.ListenAsync(this.ConnectionName, new FanoutExchangeConfiguration(InvalidMessageExchangeName)), this.ReceivedMessages.Add, onError: this.Errors.Add); + + await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync(this.ConnectionName, new FanoutExchangeConfiguration(InvalidMessageAsyncExchangeName)), x => + { + this.AsyncReceivedMessages.Add(x); + return Task.CompletedTask; + }, onError: this.AsyncErrors.Add); + } + } + public class AutoRecoveryTestMessagingBackgroundService: MessagingBackgroundService { /// diff --git a/Mercurio.Tests/Messaging/InvalidMessageExceptionTestFixture.cs b/Mercurio.Tests/Messaging/InvalidMessageExceptionTestFixture.cs new file mode 100644 index 0000000..e07d6c7 --- /dev/null +++ b/Mercurio.Tests/Messaging/InvalidMessageExceptionTestFixture.cs @@ -0,0 +1,44 @@ +// ------------------------------------------------------------------------------------------------- +// +// +// Copyright 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.Tests.Messaging +{ + using Mercurio.Messaging; + + using RabbitMQ.Client; + using RabbitMQ.Client.Events; + + [TestFixture] + public class InvalidMessageExceptionTestFixture + { + [Test] + public void VerifyInvalidMessageExceptionConstructor() + { + var basicDeliveryEvent = new BasicDeliverEventArgs("tag", 1, true, "abc","route", new BasicProperties(), new ReadOnlyMemory()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(() => new InvalidMessageException(null, null, null), Throws.ArgumentNullException); + Assert.That(() => new InvalidMessageException(typeof(int), null, null), Throws.ArgumentNullException); + Assert.That(() => new InvalidMessageException(typeof(int), basicDeliveryEvent, null), Throws.ArgumentNullException); + } + } + } +} diff --git a/Mercurio/Messaging/InvalidMessageException.cs b/Mercurio/Messaging/InvalidMessageException.cs new file mode 100644 index 0000000..8ad99cf --- /dev/null +++ b/Mercurio/Messaging/InvalidMessageException.cs @@ -0,0 +1,111 @@ +// ------------------------------------------------------------------------------------------------- +// +// +// 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 System.Text; + + using RabbitMQ.Client.Events; + + /// + /// The is the that is reported when a received message could not + /// be deserialized into the type that is being listened for + /// + /// + /// The content of the received message is copied, so that the can safely be kept alive + /// after the reception of the message has been handled, which is not the case of the + /// + public class InvalidMessageException : Exception + { + /// + /// Initializes a new instance of the + /// + /// The into which the received message could not be deserialized + /// The of the received message + /// The that has been thrown during the deserialization + /// If any of the provided arguments is null + public InvalidMessageException(Type messageType, BasicDeliverEventArgs deliverEventArgs, Exception innerException) + : base($"The received message could not be deserialized into a {messageType?.Name}", innerException) + { + if (messageType == null) + { + throw new ArgumentNullException(nameof(messageType)); + } + + if (deliverEventArgs == null) + { + throw new ArgumentNullException(nameof(deliverEventArgs)); + } + + if (innerException == null) + { + throw new ArgumentNullException(nameof(innerException)); + } + + var properties = deliverEventArgs.BasicProperties; + + this.MessageType = messageType; + this.Body = deliverEventArgs.Body.ToArray(); + this.Exchange = deliverEventArgs.Exchange; + this.RoutingKey = deliverEventArgs.RoutingKey; + this.ContentType = properties?.ContentType; + this.CorrelationId = properties?.CorrelationId; + } + + /// + /// Gets the into which the received message could not be deserialized + /// + public Type MessageType { get; } + + /// + /// Gets a copy of the raw content of the received message + /// + public byte[] Body { get; } + + /// + /// Gets the name of the exchange the received message has been published to + /// + public string Exchange { get; } + + /// + /// Gets the routing key that has been used to publish the received message + /// + public string RoutingKey { get; } + + /// + /// Gets the content type of the received message + /// + public string ContentType { get; } + + /// + /// Gets the correlation id of the received message, if any + /// + public string CorrelationId { get; } + + /// + /// Gets the raw content of the received message as an UTF8 + /// + /// The content of the received message + public string GetBodyAsString() + { + return Encoding.UTF8.GetString(this.Body); + } + } +} diff --git a/Mercurio/Messaging/MessageClientService.cs b/Mercurio/Messaging/MessageClientService.cs index 8211abf..8f84fc4 100644 --- a/Mercurio/Messaging/MessageClientService.cs +++ b/Mercurio/Messaging/MessageClientService.cs @@ -417,12 +417,25 @@ async Task ConsumerOnReceivedAsync(object o, BasicDeliverEventArgs message) try { - using var stream = message.Body.AsStream(); - var content = await this.SerializationProviderService.ResolveDeserializer(message.BasicProperties.ContentType).DeserializeAsync(stream, cancellationToken); + TMessage content; + + try + { + await using var stream = message.Body.AsStream(); + content = await this.SerializationProviderService.ResolveDeserializer(message.BasicProperties.ContentType).DeserializeAsync(stream, cancellationToken); + } + catch (Exception deserializationException) + { + 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); + activity?.SetStatus(ActivityStatusCode.Error, invalidMessageException.Message); + observer.OnError(invalidMessageException); + return; + } + observer.OnNext(content); activity?.SetStatus(ActivityStatusCode.Ok); - - await Task.CompletedTask; } catch (Exception ex) {