Skip to content

Fix #39: Safe subscription #40

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
antoineatstariongroup merged 5 commits into from
Jan 23, 2026
Merged

Fix #39: Safe subscription #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1e0941a
Implemented safe subscription to also handle exception on calling act…
antoineatstariongroup Jan 22, 2026
106f0cd
Moved from sonar.login to sonar.token
antoineatstariongroup Jan 22, 2026
15aed75
Upgrade deps and does not recover observable if it's the action that …
antoineatstariongroup Jan 23, 2026
d06c3db
Code coverage
antoineatstariongroup Jan 23, 2026
1c6557b
Version bump
antoineatstariongroup Jan 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/CodeQuality.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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"

2 changes: 1 addition & 1 deletion Mercurio.Core.Tests/Mercurio.Core.Tests.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@

<ItemGroup>
<PackageReference Include="NUnit" Version="4.4.0" />
<PackageReference Include="Testcontainers.RabbitMq" Version="4.8.1" />
<PackageReference Include="Testcontainers.RabbitMq" Version="4.10.0" />
</ItemGroup>

</Project>
3 changes: 1 addition & 2 deletions Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
6 changes: 3 additions & 3 deletions Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,13 @@
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="NUnit3TestAdapter" Version="5.2.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
<PackageReference Include="NUnit3TestAdapter" Version="6.1.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.2" />
<PackageReference Include="coverlet.msbuild" Version="6.0.4">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Testcontainers.RabbitMq" Version="4.8.1" />
<PackageReference Include="Testcontainers.RabbitMq" Version="4.10.0" />
</ItemGroup>

<ItemGroup>
Expand Down
105 changes: 103 additions & 2 deletions Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,15 +162,15 @@ public async Task VerifyRpcServer()
var rpcObservable = await client.SendRequestAsync(ConfiguredConnectionName, RpcServerQueueName, 45, cancellationToken: cancellationTokenSource.Token);
var taskCompletion = new TaskCompletionSource<bool>();
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);

rpcObservable = await client.SendRequestAsync(ConfiguredConnectionName, RpcServerQueueName, 44, cancellationToken: cancellationTokenSource.Token);
var newTaskCompletion = new TaskCompletionSource<bool>();
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);
Expand All @@ -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<ILogger<AutoRecoveryTestMessagingBackgroundService>>(), 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<ILogger<AutoRecoveryTestMessagingBackgroundService>>(), 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
{
/// <summary>
Expand All @@ -204,6 +254,57 @@ protected override Task InitializeAsync()
return Task.CompletedTask;
}
}

public class AutoRecoveryTestMessagingBackgroundService: MessagingBackgroundService
{
/// <summary>
/// Stores all received message
/// </summary>
public readonly List<string> ReceivedMessages = [];

/// <summary>
/// Initializes a new instance of the <see cref="MessagingBackgroundService" />
/// </summary>
/// <param name="serviceProvider">
/// The injected <see cref="IServiceProvider" /> that allow to resolve
/// <see cref="IMessageClientService" /> instance, even if not registered as scope
/// </param>
/// <param name="logger"></param>
/// <param name="configuration"></param>
public AutoRecoveryTestMessagingBackgroundService(IServiceProvider serviceProvider, ILogger<AutoRecoveryTestMessagingBackgroundService> logger, IConfiguration configuration) : base(serviceProvider, logger, configuration)
{
}

/// <summary>
/// Initializes this service (e.g. to set the <see cref="MessagingBackgroundService.ConnectionName" /> and register subscriptions
/// collection
/// </summary>
/// <returns>An awaitable <see cref="Task" /></returns>
protected override async Task InitializeAsync()
{
this.ConnectionName = ConfiguredConnectionName;
await this.RegisterListener(() => this.MessageClientService.ListenAsync<string>(this.ConnectionName, new FanoutExchangeConfiguration("AutoRecoveryBackgroundTest")), this.HandleMessage);
await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync<string>(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
{
Expand Down
10 changes: 5 additions & 5 deletions Mercurio.Hosting/Mercurio.Hosting.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>disable</Nullable>
<TargetFrameworks>net10.0;net9.0;net8.0;netstandard2.1</TargetFrameworks>
<Version>2.0.0</Version>
<Version>2.0.1</Version>
<LangVersion>12.0</LangVersion>
<Company>Starion Group S.A.</Company>
<Copyright>Copyright © Starion Group S.A.</Copyright>
Expand All @@ -28,7 +28,7 @@
<GenerateSBOM>true</GenerateSBOM>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<PackageReleaseNotes>
[BUMP] Bump to Net10.0
[FIX] Fix Bug with thrown exception during onNext Action/Function
</PackageReleaseNotes>
</PropertyGroup>

Expand All @@ -43,9 +43,9 @@
</ItemGroup>

<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
<PackageReference Include="Microsoft.Sbom.Targets" Version="4.1.3" PrivateAssets="all" />
<PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks.Abstractions" Version="10.0.2" />
<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.2" />
<PackageReference Include="Microsoft.Sbom.Targets" Version="4.1.5" PrivateAssets="all" />
</ItemGroup>

</Project>
6 changes: 3 additions & 3 deletions Mercurio.Hosting/MessagingBackgroundService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -262,7 +262,7 @@ protected async Task RegisterListener<TMessage>(Func<Task<IObservable<TMessage>>
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;
}

Expand All @@ -285,7 +285,7 @@ protected async Task RegisterAsyncListener<TMessage>(Func<Task<IObservable<TMess
var key = Guid.NewGuid();

var observable = await observableFunction();
var subscription = observable.SubscribeAsync(onReceive, x => _ = 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;
}

Expand Down
10 changes: 6 additions & 4 deletions Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,10 +36,12 @@ public void VerifyNullArgumentsThrow()
using (Assert.EnterMultipleScope())
{
IObservable<int> 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<int>();
Assert.That(() => observable.SubscribeAsync(null), Throws.ArgumentNullException);
Assert.That(() => observable.SubscribeSafeAsync(null), Throws.ArgumentNullException);
Assert.That(() => observable.SubscribeSafe(onNext: null), Throws.ArgumentNullException);
}
}

Expand All @@ -50,7 +52,7 @@ public async Task VerifySubscription()
var completed = false;
var subject = new Subject<int>();

using var subscription = subject.SubscribeAsync(x =>
using var subscription = subject.SubscribeSafeAsync(x =>
{
results.Add(x);
return Task.CompletedTask;
Expand All @@ -75,7 +77,7 @@ public async Task VerifyErrorIsHandled()
var subject = new Subject<int>();
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);

Expand Down
28 changes: 14 additions & 14 deletions Mercurio.Tests/Mercurio.Tests.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,31 +13,31 @@
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Microsoft.Extensions.Configuration" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Configuration" Version="10.0.2" />
<PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.2" />
<PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="10.0.2" />
<PackageReference Include="Moq" Version="4.20.72" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.0.1" />
<PackageReference Include="NUnit" Version="4.4.0" />
<PackageReference Include="NUnit.Analyzers" Version="4.11.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="NUnit3TestAdapter" Version="5.2.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
<PackageReference Include="NUnit3TestAdapter" Version="6.1.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.2" />
<PackageReference Include="coverlet.msbuild" Version="6.0.4">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" />
<PackageReference Include="OpenTelemetry" Version="1.14.0" />
<PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.14.0" />
<PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="1.14.0" />
<PackageReference Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.14.0" />
<PackageReference Include="OpenTelemetry.Instrumentation.Http" Version="1.14.0" />
<PackageReference Include="OpenTelemetry.Instrumentation.Runtime" Version="1.14.0" />
<PackageReference Include="Testcontainers.RabbitMq" Version="4.8.1" />
<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.2" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.2" />
<PackageReference Include="OpenTelemetry" Version="1.15.0" />
<PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.15.0" />
<PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="1.15.0" />
<PackageReference Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.15.0" />
<PackageReference Include="OpenTelemetry.Instrumentation.Http" Version="1.15.0" />
<PackageReference Include="OpenTelemetry.Instrumentation.Runtime" Version="1.15.0" />
<PackageReference Include="Testcontainers.RabbitMq" Version="4.10.0" />
</ItemGroup>

<ItemGroup>
Expand Down
Loading
Loading