feat(nats): add named-broker support (GH-3310)

docs/guide/messaging/transports/nats.md

@@ -478,6 +478,43 @@ opts.PublishMessage<OrderCreated>()
 
 When native scheduled send is not available (server < 2.12 or stream not configured), Wolverine falls back to its database-backed scheduled message persistence.
 
+## Connecting to Multiple NATS Brokers
+
+If a single Wolverine application needs to talk to more than one NATS broker, register the additional
+broker(s) with `AddNamedNatsBroker` using a `BrokerName`, then pin publishing or listening to a specific
+broker with the `*OnNamedBroker` overloads:
+
+```csharp
+opts.UseNats("nats://localhost:4222");
+
+// An additional, independent NATS broker identified by name
+opts.AddNamedNatsBroker(new BrokerName("secondary"), "nats://secondary-nats:4222");
+
+// Or configure the additional broker with the full connection/auth surface
+opts.AddNamedNatsBroker(new BrokerName("eu"), cfg =>
+{
+    cfg.ConnectionString = "nats://eu-nats:4222";
+    cfg.EnableJetStream = true;
+});
+
+// Publish a message type to a subject on a named broker
+opts.PublishMessage<OrderPlaced>()
+    .ToNatsSubjectOnNamedBroker(new BrokerName("secondary"), "orders");
+
+// Listen to a subject on a named broker
+opts.ListenToNatsSubjectOnNamedBroker(new BrokerName("secondary"), "orders");
+```
+
+::: info
+The Wolverine `Uri` scheme for any endpoint on a named broker is the broker name itself, so in the example
+above you would see endpoint URIs like `secondary://subject/orders`. The default broker keeps the canonical
+`nats://` scheme, which keeps the two brokers' endpoints from colliding.
+:::
+
+Connecting to multiple named brokers is distinct from [Multi-Tenancy](#multi-tenancy): a named broker is a
+statically-addressed second connection that you target explicitly, whereas per-tenant connections are
+selected at runtime from each message's tenant id.
+
 ## Multi-Tenancy
 
 ::: tip

src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs

@@ -0,0 +1,272 @@
+using IntegrationTests;
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Shouldly;
+using Testcontainers.Nats;
+using Wolverine.Nats.Internal;
+using Wolverine.Tracking;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Wolverine.Nats.Tests;
+
+/// <summary>
+/// Registration-level coverage for named NATS brokers that needs no running broker, so it always runs in CI.
+/// A named broker is a second, independent <see cref="NatsTransport"/> whose <c>Protocol</c> (and therefore
+/// its endpoints' URI scheme) is the broker name, so its endpoints never collide with the default
+/// <c>nats://</c> broker.
+/// </summary>
+public class NatsNamedBrokerRegistrationTests
+{
+    private readonly BrokerName theName = new("secondary");
+
+    [Fact]
+    public void adds_a_distinct_transport_instance_per_broker()
+    {
+        var options = new WolverineOptions();
+        options.UseNats("nats://localhost:4222");
+        options.AddNamedNatsBroker(theName, "nats://localhost:5222");
+
+        var transports = options.Transports.OfType<NatsTransport>().ToList();
+        transports.Count.ShouldBe(2);
+        transports.Select(x => x.Protocol).OrderBy(x => x).ShouldBe(["nats", "secondary"]);
+    }
+
+    [Fact]
+    public void named_broker_endpoints_use_the_broker_name_as_their_uri_scheme()
+    {
+        var options = new WolverineOptions();
+        options.UseNats("nats://localhost:4222");
+        options.AddNamedNatsBroker(theName, "nats://localhost:5222");
+
+        var named = options.Transports.OfType<NatsTransport>().Single(x => x.Protocol == "secondary");
+        named.EndpointForSubject("orders.created").Uri.ShouldBe(new Uri("secondary://subject/orders.created"));
+
+        // ...and the default broker keeps the canonical nats:// scheme.
+        var @default = options.Transports.OfType<NatsTransport>().Single(x => x.Protocol == "nats");
+        @default.EndpointForSubject("orders.created").Uri.ShouldBe(new Uri("nats://subject/orders.created"));
+    }
+
+    [Fact]
+    public void configuration_action_overload_applies_to_the_named_broker_only()
+    {
+        var options = new WolverineOptions();
+        options.UseNats("nats://localhost:4222");
+        options.AddNamedNatsBroker(theName, cfg =>
+        {
+            cfg.ConnectionString = "nats://localhost:5222";
+            cfg.EnableJetStream = false;
+        });
+
+        var named = options.Transports.OfType<NatsTransport>().Single(x => x.Protocol == "secondary");
+        named.Configuration.ConnectionString.ShouldBe("nats://localhost:5222");
+
+        var @default = options.Transports.OfType<NatsTransport>().Single(x => x.Protocol == "nats");
+        @default.Configuration.ConnectionString.ShouldBe("nats://localhost:4222");
+    }
+}
+
+/// <summary>
+/// End-to-end coverage that a named NATS broker actually talks to a <em>different</em> server than the
+/// default one. Mirrors <see cref="NatsPerTenantConnectionTests"/>: the default broker is server A (from
+/// NATS_URL / docker-compose) and the named broker is a second Testcontainers broker (server B). Proves:
+/// <list type="bullet">
+/// <item>a message published on the named broker lands on <b>server B</b> and not server A,</item>
+/// <item>a default publish lands on <b>server A</b> and not server B,</item>
+/// <item>a message published <b>and consumed</b> on the named broker round-trips and arrives stamped with
+/// the broker's own URI scheme (the receive pipeline sets <c>Destination</c> from the listener endpoint's
+/// URI), and</item>
+/// <item>a full request/reply round-trips over the named broker, proving the reply is routed back to
+/// server B rather than the default server A.</item>
+/// </list>
+/// </summary>
+[Collection("NATS Integration")]
+[Trait("Category", "Integration")]
+public class NatsNamedBrokerTests : IAsyncLifetime
+{
+    private readonly ITestOutputHelper _output;
+    private readonly BrokerName theName = new("secondary");
+    private NatsContainer? _serverB;
+    private string _serverAUrl = null!;
+    private string _serverBUrl = null!;
+    private bool _skip;
+
+    public NatsNamedBrokerTests(ITestOutputHelper output) => _output = output;
+
+    public async Task InitializeAsync()
+    {
+        _serverAUrl = NatsTestHelpers.ResolveUrl();
+
+        if (!await NatsTestHelpers.IsNatsAvailable(_serverAUrl))
+        {
+            _skip = true;
+            return;
+        }
+
+        _serverB = new NatsBuilder().WithImage("nats:latest").Build();
+        await _serverB.StartAsync();
+        _serverBUrl = _serverB.GetConnectionString();
+
+        _output.WriteLine($"Server A (default): {_serverAUrl}");
+        _output.WriteLine($"Server B (named):   {_serverBUrl}");
+    }
+
+    public async Task DisposeAsync()
+    {
+        if (_serverB != null)
+        {
+            await _serverB.DisposeAsync();
+        }
+    }
+
+    [Fact]
+    public async Task named_broker_send_lands_on_the_named_broker()
+    {
+        if (_skip) return;
+
+        var subject = $"named.{Guid.NewGuid():N}";
+
+        await using var subOnB = await NatsTestHelpers.SubscribeRawAsync(_serverBUrl, subject);
+        await using var subOnA = await NatsTestHelpers.SubscribeRawAsync(_serverAUrl, subject);
+
+        using var host = await BuildSenderAsync(subject, useNamedBroker: true);
+
+        await host.MessageBus().SendAsync(new OrderPlaced("on-named-broker"));
+
+        // Landed on server B (the named broker's connection)...
+        var received = await subOnB.ReadAsync(15.Seconds());
+        received.ShouldNotBeNull();
+        received!.Value.Subject.ShouldBe(subject);
+
+        // ...and NOT on the default server A.
+        (await subOnA.ReadAsync(2.Seconds())).ShouldBeNull();
+    }
+
+    [Fact]
+    public async Task default_broker_send_lands_on_the_default_broker()
+    {
+        if (_skip) return;
+
+        var subject = $"named.{Guid.NewGuid():N}";
+
+        await using var subOnA = await NatsTestHelpers.SubscribeRawAsync(_serverAUrl, subject);
+        await using var subOnB = await NatsTestHelpers.SubscribeRawAsync(_serverBUrl, subject);
+
+        using var host = await BuildSenderAsync(subject, useNamedBroker: false);
+
+        await host.MessageBus().SendAsync(new OrderPlaced("on-default-broker"));
+
+        var received = await subOnA.ReadAsync(15.Seconds());
+        received.ShouldNotBeNull();
+        received!.Value.Subject.ShouldBe(subject);
+
+        (await subOnB.ReadAsync(2.Seconds())).ShouldBeNull();
+    }
+
+    [Fact]
+    public async Task round_trips_a_message_over_the_named_broker()
+    {
+        if (_skip) return;
+
+        var subject = $"named.{Guid.NewGuid():N}";
+
+        // A single host both publishes and listens on the named broker (server B). On receipt the pipeline
+        // stamps Destination from the listener endpoint's URI, so the consumed envelope carries the named
+        // broker's "secondary" scheme rather than the default "nats".
+        using var host = await Host.CreateDefaultBuilder()
+            .ConfigureLogging(l => l.AddXunitLogging(_output))
+            .UseWolverine(opts =>
+            {
+                opts.ServiceName = "NamedBrokerInbound";
+                opts.UseNats(_serverAUrl);
+                opts.AddNamedNatsBroker(theName, _serverBUrl);
+
+                opts.PublishMessage<OrderPlaced>().ToNatsSubjectOnNamedBroker(theName, subject).SendInline();
+                opts.ListenToNatsSubjectOnNamedBroker(theName, subject);
+            })
+            .StartAsync();
+
+        var session = await host
+            .TrackActivity()
+            .Timeout(30.Seconds())
+            .WaitForMessageToBeReceivedAt<OrderPlaced>(host)
+            .ExecuteAndWaitAsync(c => c.SendAsync(new OrderPlaced("round-trip")));
+
+        var received = session.Received.SingleEnvelope<OrderPlaced>();
+        received.Message.ShouldBeOfType<OrderPlaced>().OrderId.ShouldBe("round-trip");
+        received.Destination!.Scheme.ShouldBe("secondary");
+    }
+
+    [Fact]
+    public async Task request_reply_round_trips_over_the_named_broker()
+    {
+        if (_skip) return;
+
+        var subject = $"named.rr.{Guid.NewGuid():N}";
+
+        // Request/reply entirely over the named broker (server B). The reply is routed by the reply-uri's
+        // scheme, which the pipeline corrects to the receiving endpoint's scheme ("secondary") — so the
+        // response travels back over server B, not the default server A. If reply routing resolved to the
+        // default broker, InvokeAndWaitAsync would time out.
+        using var host = await Host.CreateDefaultBuilder()
+            .ConfigureLogging(l => l.AddXunitLogging(_output))
+            .UseWolverine(opts =>
+            {
+                opts.ServiceName = "NamedBrokerRequestReply";
+                opts.UseNats(_serverAUrl);
+                opts.AddNamedNatsBroker(theName, _serverBUrl);
+
+                opts.Policies.DisableConventionalLocalRouting();
+                opts.PublishMessage<NamedPing>().ToNatsSubjectOnNamedBroker(theName, subject);
+                opts.ListenToNatsSubjectOnNamedBroker(theName, subject);
+            })
+            .StartAsync();
+
+        var (_, response) = await host
+            .TrackActivity()
+            .Timeout(30.Seconds())
+            .InvokeAndWaitAsync<NamedPong>(new NamedPing("named-rr"));
+
+        response.ShouldNotBeNull();
+        response.Name.ShouldBe("named-rr");
+    }
+
+    /// <summary>
+    /// Both brokers are always registered and connected (server A default, server B named), so "not on the
+    /// other server" is a meaningful assertion. Only <b>one</b> publish rule is registered per host — to the
+    /// named broker or the default — so a single <c>OrderPlaced</c> send targets exactly one server.
+    /// </summary>
+    private Task<IHost> BuildSenderAsync(string subject, bool useNamedBroker)
+    {
+        return Host.CreateDefaultBuilder()
+            .ConfigureLogging(l => l.AddXunitLogging(_output))
+            .UseWolverine(opts =>
+            {
+                opts.ServiceName = "NamedBrokerSender";
+                opts.UseNats(_serverAUrl);
+                opts.AddNamedNatsBroker(theName, _serverBUrl);
+
+                opts.Policies.DisableConventionalLocalRouting();
+
+                if (useNamedBroker)
+                {
+                    opts.PublishMessage<OrderPlaced>().ToNatsSubjectOnNamedBroker(theName, subject).SendInline();
+                }
+                else
+                {
+                    opts.PublishMessage<OrderPlaced>().ToNatsSubject(subject).SendInline();
+                }
+            })
+            .StartAsync();
+    }
+}
+
+public record NamedPing(string Name);
+
+public record NamedPong(string Name);
+
+public class NamedPingHandler
+{
+    public NamedPong Handle(NamedPing ping) => new(ping.Name);
+}