From c4f6779cd457e97cdee6177b5fda5a6232380083 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 5 Jul 2026 17:19:23 -0500 Subject: [PATCH 1/2] feat(nats): add named-broker support (GH-3310) Bring the NATS transport to parity with RabbitMQ/Kafka/SQS/Azure on the "named broker" feature so a single application can address multiple independent NATS brokers, each pinnable via *OnNamedBroker overloads. - Add a NatsTransport(string protocol) constructor so the transport can be created through TransportCollection.GetOrCreate(BrokerName). A named broker's Protocol doubles as its endpoints' URI scheme. - Build endpoint URIs and inbound envelope Destination/ReplyUri from the transport's Protocol instead of a hard-coded "nats" literal, so a named broker's endpoints never collide with the default nats:// broker and reply/tracking routing resolves back to the right transport instance. - Relax NatsTransport.ExtractSubjectFromUri: the scheme is no longer validated against a fixed literal (named brokers carry the broker name as the scheme; routing by scheme already happened upstream). - NatsHealthCheck.Protocol now reflects the transport's Protocol. - New extension methods: AddNamedNatsBroker (connection-string and Action overloads), ToNatsSubjectOnNamedBroker, ListenToNatsSubjectOnNamedBroker. Tests: 3 CI-safe registration/URI tests + 3 integration tests that spin up a second broker via Testcontainers proving a named send lands on the named broker (not the default), a default send lands on the default (not the named), and a round-trip over the named broker stamps the broker's scheme. Full Wolverine.Nats suite green (144). Docs: new "Connecting to Multiple NATS Brokers" section in nats.md. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/guide/messaging/transports/nats.md | 37 +++ .../NatsNamedBrokerTests.cs | 226 ++++++++++++++++++ .../Extensions/NatsTransportExtensions.cs | 91 ++++++- .../Wolverine.Nats/Internal/NatsEndpoint.cs | 2 +- .../Internal/NatsEnvelopeMapper.cs | 22 +- .../Internal/NatsHealthCheck.cs | 2 +- .../Wolverine.Nats/Internal/NatsTransport.cs | 25 +- 7 files changed, 387 insertions(+), 18 deletions(-) create mode 100644 src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs 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..10ef0a252 --- /dev/null +++ b/src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs @@ -0,0 +1,226 @@ +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, and +/// a message published and consumed on the named broker round-trips (exercising the named +/// broker's inbound envelope mapping, which stamps the broker's own URI scheme). +/// +/// +[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). The round-trip exercises + // the named broker's inbound envelope mapping, which must stamp the "secondary" scheme (not "nats") + // so Destination/reply routing resolves back to the right transport instance. + 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"); + } + + /// + /// 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(); + } +} 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/NatsEnvelopeMapper.cs b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEnvelopeMapper.cs index 6bd91365b..e9e985465 100644 --- a/src/Transports/NATS/Wolverine.Nats/Internal/NatsEnvelopeMapper.cs +++ b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEnvelopeMapper.cs @@ -8,11 +8,17 @@ namespace Wolverine.Nats.Internal; public class NatsEnvelopeMapper : EnvelopeMapper, NatsHeaders> { private readonly ITenantSubjectMapper? _tenantMapper; - + + // Named brokers carry the broker name as their URI scheme, so incoming envelopes must be stamped with + // this endpoint's scheme (not a hard-coded "nats") for reply/tracking routing to resolve back to the + // right transport instance. See AddNamedNatsBroker. + private readonly string _scheme; + public NatsEnvelopeMapper(NatsEndpoint endpoint, ITenantSubjectMapper? tenantMapper = null) : base(endpoint) { _tenantMapper = tenantMapper; + _scheme = endpoint.Uri.Scheme; } protected override void writeOutgoingHeader(NatsHeaders headers, string key, string value) @@ -45,7 +51,7 @@ out string? value protected override void writeIncomingHeaders(NatsMsg incoming, Envelope envelope) { envelope.Data = incoming.Data; - envelope.Destination = new Uri($"nats://subject/{incoming.Subject}"); + envelope.Destination = new Uri($"{_scheme}://subject/{incoming.Subject}"); if (_tenantMapper != null) { @@ -61,7 +67,7 @@ protected override void writeIncomingHeaders(NatsMsg incoming, Envelope EnvelopeSerializer.ReadDataElement( envelope, EnvelopeConstants.ReplyUriKey, - $"nats://subject/{incoming.ReplyTo}" + $"{_scheme}://subject/{incoming.ReplyTo}" ); } @@ -78,11 +84,15 @@ protected override void writeIncomingHeaders(NatsMsg incoming, Envelope public class JetStreamEnvelopeMapper : EnvelopeMapper, NatsHeaders> { private readonly ITenantSubjectMapper? _tenantMapper; - + + // See NatsEnvelopeMapper._scheme — named brokers carry the broker name as their URI scheme. + private readonly string _scheme; + public JetStreamEnvelopeMapper(NatsEndpoint endpoint, ITenantSubjectMapper? tenantMapper = null) : base(endpoint) { _tenantMapper = tenantMapper; + _scheme = endpoint.Uri.Scheme; } protected override void writeOutgoingHeader(NatsHeaders headers, string key, string value) @@ -115,7 +125,7 @@ out string? value protected override void writeIncomingHeaders(INatsJSMsg incoming, Envelope envelope) { envelope.Data = incoming.Data; - envelope.Destination = new Uri($"nats://subject/{incoming.Subject}"); + envelope.Destination = new Uri($"{_scheme}://subject/{incoming.Subject}"); if (_tenantMapper != null) { @@ -131,7 +141,7 @@ protected override void writeIncomingHeaders(INatsJSMsg incoming, Envelo EnvelopeSerializer.ReadDataElement( envelope, EnvelopeConstants.ReplyUriKey, - $"nats://subject/{incoming.ReplyTo}" + $"{_scheme}://subject/{incoming.ReplyTo}" ); } 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; } From 82a967d64db06a821e31f1c0e4a928d584d10875 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 5 Jul 2026 17:36:22 -0500 Subject: [PATCH 2/2] review(nats): drop unnecessary inbound-mapper scheme change; add request/reply test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses review feedback questioning whether the NatsEnvelopeMapper scheme change was necessary. It was not: on a received message the pipeline already sets the correct scheme without touching the mapper. - Envelope.MarkReceived sets Destination from the listener endpoint's URI (NatsListener.Address = NatsEndpoint.Uri), which already carries the named broker's scheme via the NatsEndpoint(protocol) constructor change. - MessageContext.ReadEnvelope calls MaybeCorrectReplyUri, which rewrites the reply-uri's scheme to match Destination.Scheme (JasperFx MaybeCorrectScheme unconditionally swaps the scheme). MarkReceived runs before ReadEnvelope, so by then Destination already has the named broker's scheme. So reverting NatsEnvelopeMapper/JetStreamEnvelopeMapper back to the canonical nats:// form is correct — the endpoint-URI scheme is the single source of truth and the pipeline propagates it to Destination and the reply-uri. Adds request_reply_round_trips_over_the_named_broker: a full InvokeAndWaitAsync over the named broker. If reply routing resolved to the default broker instead of the named one, this would time out — so it proves the reply travels back over the named broker with the mapper untouched. --- .../NatsNamedBrokerTests.cs | 58 +++++++++++++++++-- .../Internal/NatsEnvelopeMapper.cs | 22 ++----- 2 files changed, 58 insertions(+), 22 deletions(-) diff --git a/src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs b/src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs index 10ef0a252..59e9aee2f 100644 --- a/src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs +++ b/src/Transports/NATS/Wolverine.Nats.Tests/NatsNamedBrokerTests.cs @@ -73,9 +73,12 @@ public void configuration_action_overload_applies_to_the_named_broker_only() /// 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, and -/// a message published and consumed on the named broker round-trips (exercising the named -/// broker's inbound envelope mapping, which stamps the broker's own URI scheme). +/// 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")] @@ -168,9 +171,9 @@ public async Task round_trips_a_message_over_the_named_broker() var subject = $"named.{Guid.NewGuid():N}"; - // A single host both publishes and listens on the named broker (server B). The round-trip exercises - // the named broker's inbound envelope mapping, which must stamp the "secondary" scheme (not "nats") - // so Destination/reply routing resolves back to the right transport instance. + // 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 => @@ -195,6 +198,40 @@ public async Task round_trips_a_message_over_the_named_broker() 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 @@ -224,3 +261,12 @@ private Task BuildSenderAsync(string subject, bool useNamedBroker) .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/Internal/NatsEnvelopeMapper.cs b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEnvelopeMapper.cs index e9e985465..6bd91365b 100644 --- a/src/Transports/NATS/Wolverine.Nats/Internal/NatsEnvelopeMapper.cs +++ b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEnvelopeMapper.cs @@ -8,17 +8,11 @@ namespace Wolverine.Nats.Internal; public class NatsEnvelopeMapper : EnvelopeMapper, NatsHeaders> { private readonly ITenantSubjectMapper? _tenantMapper; - - // Named brokers carry the broker name as their URI scheme, so incoming envelopes must be stamped with - // this endpoint's scheme (not a hard-coded "nats") for reply/tracking routing to resolve back to the - // right transport instance. See AddNamedNatsBroker. - private readonly string _scheme; - + public NatsEnvelopeMapper(NatsEndpoint endpoint, ITenantSubjectMapper? tenantMapper = null) : base(endpoint) { _tenantMapper = tenantMapper; - _scheme = endpoint.Uri.Scheme; } protected override void writeOutgoingHeader(NatsHeaders headers, string key, string value) @@ -51,7 +45,7 @@ out string? value protected override void writeIncomingHeaders(NatsMsg incoming, Envelope envelope) { envelope.Data = incoming.Data; - envelope.Destination = new Uri($"{_scheme}://subject/{incoming.Subject}"); + envelope.Destination = new Uri($"nats://subject/{incoming.Subject}"); if (_tenantMapper != null) { @@ -67,7 +61,7 @@ protected override void writeIncomingHeaders(NatsMsg incoming, Envelope EnvelopeSerializer.ReadDataElement( envelope, EnvelopeConstants.ReplyUriKey, - $"{_scheme}://subject/{incoming.ReplyTo}" + $"nats://subject/{incoming.ReplyTo}" ); } @@ -84,15 +78,11 @@ protected override void writeIncomingHeaders(NatsMsg incoming, Envelope public class JetStreamEnvelopeMapper : EnvelopeMapper, NatsHeaders> { private readonly ITenantSubjectMapper? _tenantMapper; - - // See NatsEnvelopeMapper._scheme — named brokers carry the broker name as their URI scheme. - private readonly string _scheme; - + public JetStreamEnvelopeMapper(NatsEndpoint endpoint, ITenantSubjectMapper? tenantMapper = null) : base(endpoint) { _tenantMapper = tenantMapper; - _scheme = endpoint.Uri.Scheme; } protected override void writeOutgoingHeader(NatsHeaders headers, string key, string value) @@ -125,7 +115,7 @@ out string? value protected override void writeIncomingHeaders(INatsJSMsg incoming, Envelope envelope) { envelope.Data = incoming.Data; - envelope.Destination = new Uri($"{_scheme}://subject/{incoming.Subject}"); + envelope.Destination = new Uri($"nats://subject/{incoming.Subject}"); if (_tenantMapper != null) { @@ -141,7 +131,7 @@ protected override void writeIncomingHeaders(INatsJSMsg incoming, Envelo EnvelopeSerializer.ReadDataElement( envelope, EnvelopeConstants.ReplyUriKey, - $"{_scheme}://subject/{incoming.ReplyTo}" + $"nats://subject/{incoming.ReplyTo}" ); }