[Host.Nats] Queues support via QueueGroup

[steel97]

[Host.Nats] Add queue support

As stated in library description:

Supports multiple messaging patterns: pub/sub, request-response, queues

This PR adds missing queue pattern to Nats transport via QueueGroup.
PR highly based on Redis transport code

[zarusz]

Hey @steel97, thanks for the contribution to the NATS transport — this looks solid overall 👍

A couple of follow-ups before we merge:

• Please update provider_nats.t.md, then run build/md-preprocessor.ps1 to regenerate the final .md file.
• Take a look at the SonarCloud warnings: see if coverage can be increased and whether there’s an opportunity to reduce duplication (it may be over-reporting in this case).

[steel97]

Hi @zarusz! Thanks for quick reply. I looked over Sonar warnings and added tests which cover configs. To pass code duplication issue I did some variable rename (I think that this is not good solution, but I didn't found better way to fix that).

Regarding provider_nats.t.md, I'm not sure what should be added there, I just updated example to include Queue and regenerated docs, hope that would be enough.

Few notes:

• I did minor refactoring inside nats example because of addition of new consumer and message.
• I removed ex from LogInformation at this place because it was too verbose on production, prints stack trace each time I stop my app. Unfortunately seems like sonar doesn't like it... I thought exception usually passed to LogError, please correct me if I'm wrong: https://github.com/zarusz/SlimMessageBus/pull/453/changes#diff-7276bcff705b2d698203cf1cb84a496d1ae7b5dcea2cb82a8b144dd1a269b440L61
• Just minor wish: it would be cool if default localhost inside appsettings all over the project will be replaced with 127.0.0.1, ip addresses more stable than hostnames (especially in some weird broken environments like mine, where localhost never resolves correctly...)

UPD:
About When_Stop_Given_ConcurrentCalls_Then_ItOnlyStopsConsumersOnce test failure, this is not related to PR and quite hard to debug. I was unable to reproduce it locally, it fails only on github actions very randomly. After debugging I noticed that start count is incremented from somewhere outside of test method.
Looks like concurrency problem.

After digging a little bit more (with lots of Console.WriteLine's) I find out next:
[xUnit.net 00:00:14.07] Output:
Begin of test
[xUnit.net 00:00:14.07] start test

Pre AutoStart call

SlimMessageBus/src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs

Line 68 in 41343b8

var bus = (T)BusBuilder.Build();

[xUnit.net 00:00:14.07] creating bus and starting #

Incremented started var, source: test method
[xUnit.net 00:00:14.07] inc When_Stop_Given_ConcurrentCalls_Then_ItOnlyStopsConsumersOnce
[xUnit.net 00:00:14.07] stopped: 0
[xUnit.net 00:00:14.07] started: 1
[xUnit.net 00:00:14.07] dec When_Stop_Given_ConcurrentCalls_Then_ItOnlyStopsConsumersOnce

Incremented started var, source: AutoStart
[xUnit.net 00:00:14.07] inc AutoStart
[xUnit.net 00:00:14.07] dec When_Stop_Given_ConcurrentCalls_Then_ItOnlyStopsConsumersOnce
[xUnit.net 00:00:14.07] test end
[xUnit.net 00:00:14.07] stopped: 2
[xUnit.net 00:00:14.07] started: 2

Based on this, looks like this failure can be seen when:

1. test started
2. var bus = Bus; initializes lazy var with CreateMessageBus
3. CreateMessageBus starts thread (Start still not called, thread not scheduled or waiting for lock to be free)
4. test calls Start method which increments started counter, then calls Stop which not only increment _stopped var but also set IsStarted variable to false
5. And finally, if thread created on step 3 just entered initial Start lock it will see: IsStarting = false (Start method already completed), and IsStarted also false, because Stop was called and handled before AutoStart, so another Start will be handled and counter incremented. test fails.

UPD 2 TL;DR
To reproduce:
Add Task.Delay(500) to AutoStart

public async virtual Task AutoStart(CancellationToken cancellationToken)
{
        await Task.Delay(500);
        ...

Add Task.Delay(1000) to When_Stop_Given_ConcurrentCalls_Then_ItOnlyStopsConsumersOnce before asserts

for (var i = 0; i < 10; i++)
{
     await Task.WhenAll(Enumerable.Range(0, 10000).Select(x => bus.Stop()).AsParallel());
}
await Task.Delay(1000);

UPD 3
Simple fix is: bus.Settings.AutoStartConsumers = false;
Looks like consumers anyway not required at that test. It might be better to move fix to another PR, but I included it here for now. Probably other concurrency related tests should be checked for this bug aswell

[zarusz]

hey @steel97 , thanks for the changes. Few more asks before we can merge:

• can you squash all your commits into one?
• apply the changes to provider_nats.t.md, as I have this post processing script mentioned above that will inject all the referrenced samples and override the provider_nats.md file.

[steel97]

can you squash all your commits into one?

Done

apply the changes to provider_nats.t.md, as I have this post processing script mentioned above that will inject all the referrenced samples and override the provider_nats.md file.

Sure, I did it already at: 4f96fa5

[sonarqubecloud]

Quality Gate passed

Issues
1 New issue
0 Accepted issues

Measures
0 Security Hotspots
100.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[zarusz]

Thanks for the contribution, @steel97! This has now been merged. I’m planning to include it in the 3.4.0 release, and there will be a release candidate (-rc) build available in the meantime for preview.