diff --git a/docs/guide/messaging/transports/nats.md b/docs/guide/messaging/transports/nats.md index 64b4a85bc..5b70adab7 100644 --- a/docs/guide/messaging/transports/nats.md +++ b/docs/guide/messaging/transports/nats.md @@ -478,6 +478,43 @@ opts.PublishMessage() 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() + .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 diff --git a/src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs b/src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs new file mode 100644 index 000000000..59e9aee2f --- /dev/null +++ b/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; + +/// +/// 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 whose Protocol (and therefore +/// its endpoints' URI scheme) is the broker name, so its endpoints never collide with the default +/// nats:// broker. +/// +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().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().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().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().Single(x => x.Protocol == "secondary"); + named.Configuration.ConnectionString.ShouldBe("nats://localhost:5222"); + + var @default = options.Transports.OfType().Single(x => x.Protocol == "nats"); + @default.Configuration.ConnectionString.ShouldBe("nats://localhost:4222"); + } +} + +/// +/// End-to-end coverage that a named NATS broker actually talks to a different server than the +/// default one. Mirrors : the default broker is server A (from +/// NATS_URL / docker-compose) and the named broker is a second Testcontainers broker (server B). Proves: +/// +/// a message published on the named broker lands on server B and not server A, +/// a default publish lands on server A and not server B, +/// a message published and consumed on the named broker round-trips and arrives stamped with +/// the broker's own URI scheme (the receive pipeline sets Destination from the listener endpoint's +/// URI), and +/// 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. +/// +/// +[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().ToNatsSubjectOnNamedBroker(theName, subject).SendInline(); + opts.ListenToNatsSubjectOnNamedBroker(theName, subject); + }) + .StartAsync(); + + var session = await host + .TrackActivity() + .Timeout(30.Seconds()) + .WaitForMessageToBeReceivedAt(host) + .ExecuteAndWaitAsync(c => c.SendAsync(new OrderPlaced("round-trip"))); + + var received = session.Received.SingleEnvelope(); + received.Message.ShouldBeOfType().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().ToNatsSubjectOnNamedBroker(theName, subject); + opts.ListenToNatsSubjectOnNamedBroker(theName, subject); + }) + .StartAsync(); + + var (_, response) = await host + .TrackActivity() + .Timeout(30.Seconds()) + .InvokeAndWaitAsync(new NamedPing("named-rr")); + + response.ShouldNotBeNull(); + response.Name.ShouldBe("named-rr"); + } + + /// + /// Both brokers are always registered and connected (server A default, server B named), so "not on the + /// other server" is a meaningful assertion. Only one publish rule is registered per host — to the + /// named broker or the default — so a single OrderPlaced send targets exactly one server. + /// + private Task 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().ToNatsSubjectOnNamedBroker(theName, subject).SendInline(); + } + else + { + opts.PublishMessage().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); +} diff --git a/src/Transports/NATS/Wolverine.Nats/Extensions/NatsTransportExtensions.cs b/src/Transports/NATS/Wolverine.Nats/Extensions/NatsTransportExtensions.cs index 8cf50e99b..7b5d12e7c 100644 --- a/src/Transports/NATS/Wolverine.Nats/Extensions/NatsTransportExtensions.cs +++ b/src/Transports/NATS/Wolverine.Nats/Extensions/NatsTransportExtensions.cs @@ -11,11 +11,12 @@ namespace Wolverine.Nats; public static class NatsTransportExtensions { /// - /// Get access to the NATS transport for advanced configuration + /// Get access to the NATS transport for advanced configuration. Pass a to + /// resolve an additional, named NATS broker (see ). /// - internal static NatsTransport NatsTransport(this WolverineOptions options) + internal static NatsTransport NatsTransport(this WolverineOptions options, BrokerName? name = null) { - return options.Transports.GetOrCreate(); + return options.Transports.GetOrCreate(name); } /// @@ -79,6 +80,46 @@ IConfiguration configuration return new NatsTransportExpression(transport, options); } + /// + /// Configure connection and authentication information for a secondary NATS broker used by this + /// application. Only use this overload if your Wolverine application needs to talk to two or more + /// NATS brokers. The doubles as the URI scheme for the additional broker's + /// endpoints, so pin publishing/listening to it with / + /// . + /// + /// + /// Identity of the additional NATS broker + /// The NATS connection string for the additional broker + public static NatsTransportExpression AddNamedNatsBroker( + this WolverineOptions options, + BrokerName name, + string connectionString + ) + { + var transport = options.NatsTransport(name); + transport.Configuration.ConnectionString = connectionString; + return new NatsTransportExpression(transport, options); + } + + /// + /// Configure connection and authentication information for a secondary NATS broker used by this + /// application with full access to the (auth, TLS, JetStream). + /// Only use this overload if your Wolverine application needs to talk to two or more NATS brokers. + /// + /// + /// Identity of the additional NATS broker + /// Configuration for the additional broker's connection + public static NatsTransportExpression AddNamedNatsBroker( + this WolverineOptions options, + BrokerName name, + Action configure + ) + { + var transport = options.NatsTransport(name); + configure(transport.Configuration); + return new NatsTransportExpression(transport, options); + } + /// /// Publish messages to a NATS subject /// @@ -98,6 +139,30 @@ string subject return new NatsSubscriberConfiguration(endpoint); } + /// + /// Publish messages to a NATS subject on an additional, named broker registered via + /// . + /// + /// + /// Identity of the additional NATS broker + /// The NATS subject + public static NatsSubscriberConfiguration ToNatsSubjectOnNamedBroker( + this IPublishToExpression publishing, + BrokerName name, + string subject + ) + { + var transports = publishing.As().Parent.Transports; + var transport = transports.GetOrCreate(name); + + var endpoint = transport.EndpointForSubject(subject); + + // This is necessary to hook up the subscription rules + publishing.To(endpoint.Uri); + + return new NatsSubscriberConfiguration(endpoint); + } + /// /// Publish messages to a NATS subject /// @@ -150,6 +215,26 @@ string subject return new NatsListenerConfiguration(endpoint); } + /// + /// Listen to messages from a NATS subject on an additional, named broker registered via + /// . + /// + /// + /// Identity of the additional NATS broker + /// The NATS subject + public static NatsListenerConfiguration ListenToNatsSubjectOnNamedBroker( + this WolverineOptions options, + BrokerName name, + string subject + ) + { + var transport = options.NatsTransport(name); + var endpoint = transport.EndpointForSubject(subject); + endpoint.IsListener = true; + + return new NatsListenerConfiguration(endpoint); + } + /// /// Access the NATS transport configuration for advanced scenarios. /// This is useful for adding policies or modifying configuration after initial setup. diff --git a/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs index 1c116287f..83b079c66 100644 --- a/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs +++ b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs @@ -19,7 +19,7 @@ public class NatsEndpoint : Endpoint, IBrokerEndpoint private NatsEnvelopeMapper? _mapper; public NatsEndpoint(string subject, NatsTransport transport, EndpointRole role) - : base(new Uri($"nats://subject/{subject}"), role) + : base(new Uri($"{transport.Protocol}://subject/{subject}"), role) { Subject = subject; _transport = transport; diff --git a/src/Transports/NATS/Wolverine.Nats/Internal/NatsHealthCheck.cs b/src/Transports/NATS/Wolverine.Nats/Internal/NatsHealthCheck.cs index 57cb2d811..e07d28c39 100644 --- a/src/Transports/NATS/Wolverine.Nats/Internal/NatsHealthCheck.cs +++ b/src/Transports/NATS/Wolverine.Nats/Internal/NatsHealthCheck.cs @@ -9,7 +9,7 @@ internal class NatsHealthCheck : WolverineTransportHealthCheck public NatsHealthCheck(NatsTransport transport) => _transport = transport; public override string TransportName => "NATS"; - public override string Protocol => "nats"; + public override string Protocol => _transport.Protocol; public override Task CheckHealthAsync(CancellationToken cancellationToken = default) { diff --git a/src/Transports/NATS/Wolverine.Nats/Internal/NatsTransport.cs b/src/Transports/NATS/Wolverine.Nats/Internal/NatsTransport.cs index 0e1db2683..4cec00085 100644 --- a/src/Transports/NATS/Wolverine.Nats/Internal/NatsTransport.cs +++ b/src/Transports/NATS/Wolverine.Nats/Internal/NatsTransport.cs @@ -33,8 +33,18 @@ public class NatsTransport : BrokerTransport, IAsyncDisposable internal JasperFx.Core.LightweightCache Tenants { get; } = new(); internal ITenantSubjectMapper TenantSubjectMapper { get; set; } = new DefaultTenantSubjectMapper(); - public NatsTransport() - : base(ProtocolName, "NATS Transport", ["nats.io"]) + public NatsTransport() : this(ProtocolName) + { + } + + /// + /// Constructor used when connecting to more than one NATS broker from a single application. The + /// doubles as the additional broker's URI scheme so its endpoints don't + /// collide with the default nats:// broker. Reached through + /// when a is supplied. + /// + public NatsTransport(string protocol) + : base(protocol, "NATS Transport", ["nats.io"]) { _endpoints.OnMissing = subject => { @@ -252,13 +262,14 @@ private static NatsOpts buildTenantNatsOpts(NatsTenant tenant) return opts with { Name = $"{opts.Name}-tenant-{tenant.TenantId}" }; } + /// + /// Extract the NATS subject from a Wolverine NATS endpoint URI of the form + /// {scheme}://subject/{subject}. The scheme is intentionally not validated against a fixed + /// literal: named brokers (see AddNamedNatsBroker) carry the broker name as the scheme, and + /// routing to the correct transport instance has already happened by scheme before this is reached. + /// public static string ExtractSubjectFromUri(Uri uri) { - if (uri.Scheme != "nats") - { - throw new ArgumentException($"Invalid URI scheme. Expected 'nats', got '{uri.Scheme}'"); - } - var path = uri.LocalPath.Trim('/'); return string.IsNullOrEmpty(path) ? uri.Host : path; }