diff --git a/.github/workflows/CodeQuality.yml b/.github/workflows/CodeQuality.yml index 4242334..a97166b 100644 --- a/.github/workflows/CodeQuality.yml +++ b/.github/workflows/CodeQuality.yml @@ -37,7 +37,7 @@ jobs: SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} run: | dotnet tool install --global dotnet-sonarscanner - dotnet sonarscanner begin /k:"STARIONGROUP_Mercurio" /o:"stariongroup" /d:sonar.login="$SONAR_TOKEN" /d:sonar.host.url="https://sonarcloud.io" /d:sonar.cs.opencover.reportsPaths="./CoverageResults/coverage.opencover.xml" + dotnet sonarscanner begin /k:"STARIONGROUP_Mercurio" /o:"stariongroup" /d:sonar.token="$SONAR_TOKEN" /d:sonar.host.url="https://sonarcloud.io" /d:sonar.cs.opencover.reportsPaths="./CoverageResults/coverage.opencover.xml" - name: Build run: dotnet build --no-restore /p:ContinuousIntegrationBuild=true @@ -51,5 +51,5 @@ jobs: - name: Sonarqube end env: SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} - run: dotnet sonarscanner end /d:sonar.login="$SONAR_TOKEN" + run: dotnet sonarscanner end /d:sonar.token="$SONAR_TOKEN" \ No newline at end of file diff --git a/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj b/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj index 89c576b..becf5fc 100644 --- a/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj +++ b/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj @@ -11,7 +11,7 @@ - + diff --git a/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs b/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs index c533e97..10206b3 100644 --- a/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs +++ b/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs @@ -32,8 +32,7 @@ public class RabbitMqContainerSetupFixture [OneTimeSetUp] public async Task Setup() { - RabbitMqContainer = new RabbitMqBuilder() - .WithImage("rabbitmq:4.2.0-alpine") + RabbitMqContainer = new RabbitMqBuilder("rabbitmq:4.2.0-alpine") .WithUsername("guest") .WithPassword("guest") .Build(); diff --git a/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj b/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj index 81791be..162a775 100644 --- a/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj +++ b/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj @@ -21,13 +21,13 @@ all runtime; build; native; contentfiles; analyzers; buildtransitive - - + + all runtime; build; native; contentfiles; analyzers; buildtransitive - + diff --git a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs index b63f57a..19c9320 100644 --- a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs +++ b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs @@ -162,7 +162,7 @@ public async Task VerifyRpcServer() var rpcObservable = await client.SendRequestAsync(ConfiguredConnectionName, RpcServerQueueName, 45, cancellationToken: cancellationTokenSource.Token); var taskCompletion = new TaskCompletionSource(); rpcObservable.Subscribe(result => taskCompletion.SetResult(result)); - await Task.Delay(50, cancellationTokenSource.Token); + await Task.Delay(100, cancellationTokenSource.Token); await taskCompletion.Task; Assert.That(taskCompletion.Task.Result, Is.False); @@ -170,7 +170,7 @@ public async Task VerifyRpcServer() rpcObservable = await client.SendRequestAsync(ConfiguredConnectionName, RpcServerQueueName, 44, cancellationToken: cancellationTokenSource.Token); var newTaskCompletion = new TaskCompletionSource(); rpcObservable.Subscribe(result => newTaskCompletion.SetResult(result)); - await Task.Delay(50, cancellationTokenSource.Token); + await Task.Delay(100, cancellationTokenSource.Token); await taskCompletion.Task; Assert.That(newTaskCompletion.Task.Result, Is.True); @@ -179,6 +179,56 @@ public async Task VerifyRpcServer() client.Dispose(); } + [Test] + 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); + + var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTest"); + autoRecoveryBackground.PushMessage("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 Task.Delay(100, cancellationTokenSource.Token); + + autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(100, cancellationTokenSource.Token); + + Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(3)); + autoRecoveryBackground.Dispose(); + } + + [Test] + 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); + + var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTestAsync"); + autoRecoveryBackground.PushMessage("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 Task.Delay(50, cancellationTokenSource.Token); + + autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token); + await Task.Delay(50, cancellationTokenSource.Token); + + Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(3)); + autoRecoveryBackground.Dispose(); + } + private class InvalidInitializationBackgroundService : MessagingBackgroundService { /// @@ -204,6 +254,57 @@ protected override Task InitializeAsync() return Task.CompletedTask; } } + + public class AutoRecoveryTestMessagingBackgroundService: MessagingBackgroundService + { + /// + /// Stores all received message + /// + public readonly List ReceivedMessages = []; + + /// + /// Initializes a new instance of the + /// + /// + /// The injected that allow to resolve + /// instance, even if not registered as scope + /// + /// + /// + public AutoRecoveryTestMessagingBackgroundService(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("AutoRecoveryBackgroundTest")), this.HandleMessage); + await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync(this.ConnectionName, new FanoutExchangeConfiguration("AutoRecoveryBackgroundTestAsync")), this.HandleMessageAsync); + } + + private Task HandleMessageAsync(string arg) + { + this.HandleMessage(arg); + return Task.CompletedTask; + } + + private void HandleMessage(string obj) + { + if (int.TryParse(obj, out _)) + { + throw new InvalidOperationException("I want to throw something"); + } + else + { + this.ReceivedMessages.Add(obj); + } + } + } public class TestMessagingBackgroundService: MessagingBackgroundService { diff --git a/Mercurio.Hosting/Mercurio.Hosting.csproj b/Mercurio.Hosting/Mercurio.Hosting.csproj index dbd0267..3e6571c 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.0 + 2.0.1 12.0 Starion Group S.A. Copyright © Starion Group S.A. @@ -28,7 +28,7 @@ true true - [BUMP] Bump to Net10.0 + [FIX] Fix Bug with thrown exception during onNext Action/Function @@ -43,9 +43,9 @@ - - - + + + diff --git a/Mercurio.Hosting/MessagingBackgroundService.cs b/Mercurio.Hosting/MessagingBackgroundService.cs index d9b39a3..00effa7 100644 --- a/Mercurio.Hosting/MessagingBackgroundService.cs +++ b/Mercurio.Hosting/MessagingBackgroundService.cs @@ -232,7 +232,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) this.Logger.LogError(exception, "Error occured while processing a message"); } - await Task.Delay(TimeSpan.FromMilliseconds(10), stoppingToken); + await Task.Delay(TimeSpan.FromMilliseconds(5), stoppingToken); } this.IsHealthy = false; @@ -262,7 +262,7 @@ protected async Task RegisterListener(Func> var key = Guid.NewGuid(); var observable = await observableFunction(); - var subscription = observable.Subscribe(onReceive, x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted); + var subscription = observable.SubscribeSafe(onReceive,onNextError:onError, onObservableError: x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted: onCompleted); this.subscriptions[key] = subscription; } @@ -285,7 +285,7 @@ protected async Task RegisterAsyncListener(Func _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted); + var subscription = observable.SubscribeSafeAsync(onReceive, onNextError: onError,onObservableError: x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted: onCompleted); this.subscriptions[key] = subscription; } diff --git a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs index 670125c..43e9538 100644 --- a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs +++ b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs @@ -36,10 +36,12 @@ public void VerifyNullArgumentsThrow() using (Assert.EnterMultipleScope()) { IObservable source = null; - Assert.That(() => source.SubscribeAsync(_ => Task.CompletedTask), Throws.ArgumentNullException); + Assert.That(() => source.SubscribeSafeAsync(_ => Task.CompletedTask), Throws.ArgumentNullException); + Assert.That(() => source.SubscribeSafe(_ => { }), Throws.ArgumentNullException); var observable = new Subject(); - Assert.That(() => observable.SubscribeAsync(null), Throws.ArgumentNullException); + Assert.That(() => observable.SubscribeSafeAsync(null), Throws.ArgumentNullException); + Assert.That(() => observable.SubscribeSafe(onNext: null), Throws.ArgumentNullException); } } @@ -50,7 +52,7 @@ public async Task VerifySubscription() var completed = false; var subject = new Subject(); - using var subscription = subject.SubscribeAsync(x => + using var subscription = subject.SubscribeSafeAsync(x => { results.Add(x); return Task.CompletedTask; @@ -75,7 +77,7 @@ public async Task VerifyErrorIsHandled() var subject = new Subject(); Exception captured = null; - using var subscription = subject.SubscribeAsync(_ => Task.FromException(new InvalidOperationException("boom")), ex => captured = ex); + using var subscription = subject.SubscribeSafeAsync(_ => Task.FromException(new InvalidOperationException("boom")), onObservableError: ex => captured = ex); subject.OnNext(1); diff --git a/Mercurio.Tests/Mercurio.Tests.csproj b/Mercurio.Tests/Mercurio.Tests.csproj index 104b745..8803245 100644 --- a/Mercurio.Tests/Mercurio.Tests.csproj +++ b/Mercurio.Tests/Mercurio.Tests.csproj @@ -13,9 +13,9 @@ all runtime; build; native; contentfiles; analyzers; buildtransitive - - - + + + @@ -23,21 +23,21 @@ all runtime; build; native; contentfiles; analyzers; buildtransitive - - + + all runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - - - - - + + + + + + + + + diff --git a/Mercurio/Extensions/ObservableExtensions.cs b/Mercurio/Extensions/ObservableExtensions.cs index a9fa0bc..4425767 100644 --- a/Mercurio/Extensions/ObservableExtensions.cs +++ b/Mercurio/Extensions/ObservableExtensions.cs @@ -1,7 +1,7 @@ // ------------------------------------------------------------------------------------------------- // // -// Copyright 2025 Starion Group S.A. +// 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. @@ -28,15 +28,18 @@ namespace Mercurio.Extensions public static class ObservableExtensions { /// - /// Subscribe to an with async capabilities + /// Safely subscribe to an with async capabilities /// /// An object /// The source /// The to call - /// An optional to handle exception - /// An optional to handle completed action + /// An optional to handle exception thrown by the + /// An optional to handle exception thrown by the + /// An optional to handle completed action + /// If one of the provided or + /// is null /// The created - public static IDisposable SubscribeAsync(this IObservable source, Func onNextAsync, Action onError = null, + public static IDisposable SubscribeSafeAsync(this IObservable source, Func onNextAsync, Action onNextError = null, Action onObservableError = null, Action onCompleted = null) { if (source == null) @@ -49,9 +52,56 @@ public static IDisposable SubscribeAsync(this IObservable source, Func Observable.FromAsync(() => onNextAsync(x))).Concat() - .Subscribe(_ => { }, onError ?? (_ => { }), onCompleted ?? (() => { })); + return source.Select(x => Observable.FromAsync(() => + { + try + { + return onNextAsync(x); + } + catch (Exception e) + { + onNextError?.Invoke(e); + return Task.CompletedTask; + } + })).Concat() + .Subscribe(_ => { }, onObservableError ?? (_ => { }), onCompleted ?? (() => { })); + } + + /// + /// Safely subscribe to an + /// + /// An object + /// The source + /// The to call + /// An optional to handle exception thrown by the + /// An optional to handle exception thrown by the + /// An optional to handle completed action + /// If one of the provided or + /// is null + /// The created + public static IDisposable SubscribeSafe(this IObservable source, Action onNext, Action onNextError = null, Action onObservableError = null, Action onCompleted = null) + { + if (source == null) + { + throw new ArgumentNullException(nameof(source)); + } + + if (onNext == null) + { + throw new ArgumentNullException(nameof(onNext)); + } + + return source.Subscribe(x => + { + try + { + onNext(x); + } + catch (Exception e) + { + onNextError?.Invoke(e); + } + }, onObservableError ?? (_ => { }), onCompleted ?? (() => { })); } } } - \ No newline at end of file diff --git a/Mercurio/Mercurio.csproj b/Mercurio/Mercurio.csproj index 1e7337e..33bbf5a 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.0 + 2.0.1 12.0 Starion Group S.A. Copyright © Starion Group S.A. @@ -28,19 +28,19 @@ true true - [BUMP] Bump to Net10.0 + [FIX] Fix Bug with thrown exception during onNext Action/Function - - - + + + - - + +