diff --git a/src/NATS.Client.JetStream/NatsJSContext.Streams.cs b/src/NATS.Client.JetStream/NatsJSContext.Streams.cs index 135d33457..d80d7d497 100644 --- a/src/NATS.Client.JetStream/NatsJSContext.Streams.cs +++ b/src/NATS.Client.JetStream/NatsJSContext.Streams.cs @@ -19,38 +19,7 @@ public async ValueTask CreateStreamAsync( StreamConfig config, CancellationToken cancellationToken = default) { - ThrowIfInvalidStreamName(config.Name, nameof(config.Name)); - - // keep caller's config intact. - config = config with { }; - - // If we have a mirror and an external domain, convert to ext.APIPrefix. - if (config.Mirror != null && !string.IsNullOrEmpty(config.Mirror.Domain)) - { - config.Mirror = config.Mirror with { }; - ConvertDomain(config.Mirror); - } - - // Check sources for the same. - if (config.Sources != null && config.Sources.Count > 0) - { - ICollection? sources = []; - foreach (var ss in config.Sources) - { - if (!string.IsNullOrEmpty(ss.Domain)) - { - var remappedDomainSource = ss with { }; - ConvertDomain(remappedDomainSource); - sources.Add(remappedDomainSource); - } - else - { - sources.Add(ss); - } - } - - config.Sources = sources; - } + config = AdjustStreamConfigForDomain(config); var response = await JSRequestResponseAsync( subject: $"{Opts.Prefix}.STREAM.CREATE.{config.Name}", @@ -71,7 +40,7 @@ public async ValueTask CreateStreamAsync( /// The name in is null. public async ValueTask CreateOrUpdateStreamAsync(StreamConfig config, CancellationToken cancellationToken = default) { - ThrowIfInvalidStreamName(config.Name, nameof(config.Name)); + config = AdjustStreamConfigForDomain(config); var response = await JSRequestAsync( subject: $"{Opts.Prefix}.STREAM.UPDATE.{config.Name}", request: config, @@ -194,7 +163,7 @@ public async ValueTask UpdateStreamAsync( StreamConfig request, CancellationToken cancellationToken = default) { - ThrowIfInvalidStreamName(request.Name, nameof(request.Name)); + request = AdjustStreamConfigForDomain(request); var response = await JSRequestResponseAsync( subject: $"{Opts.Prefix}.STREAM.UPDATE.{request.Name}", request: request, @@ -267,4 +236,42 @@ public async IAsyncEnumerable ListStreamNamesAsync(string? subject = def offset += response.Streams.Count; } } + + private static StreamConfig AdjustStreamConfigForDomain(StreamConfig config) + { + ThrowIfInvalidStreamName(config.Name, nameof(config.Name)); + + // keep caller's config intact. + config = config with { }; + + // If we have a mirror and an external domain, convert to ext.APIPrefix. + if (config.Mirror != null && !string.IsNullOrEmpty(config.Mirror.Domain)) + { + config.Mirror = config.Mirror with { }; + ConvertDomain(config.Mirror); + } + + // Check sources for the same. + if (config.Sources != null && config.Sources.Count > 0) + { + ICollection? sources = []; + foreach (var ss in config.Sources) + { + if (!string.IsNullOrEmpty(ss.Domain)) + { + var remappedDomainSource = ss with { }; + ConvertDomain(remappedDomainSource); + sources.Add(remappedDomainSource); + } + else + { + sources.Add(ss); + } + } + + config.Sources = sources; + } + + return config; + } } diff --git a/src/NATS.Client.JetStream/NatsJSContext.cs b/src/NATS.Client.JetStream/NatsJSContext.cs index 79cb77ee1..edf941522 100644 --- a/src/NATS.Client.JetStream/NatsJSContext.cs +++ b/src/NATS.Client.JetStream/NatsJSContext.cs @@ -500,7 +500,7 @@ private static void ConvertDomain(StreamSource streamSource) return; } - if (streamSource.External != null) + if (streamSource.External?.Deliver is { Length: > 0 }) { throw new ArgumentException("Both domain and external are set"); } diff --git a/tests/NATS.Client.JetStream.Tests/StreamDomainAdjustmentTest.cs b/tests/NATS.Client.JetStream.Tests/StreamDomainAdjustmentTest.cs new file mode 100644 index 000000000..5da808331 --- /dev/null +++ b/tests/NATS.Client.JetStream.Tests/StreamDomainAdjustmentTest.cs @@ -0,0 +1,226 @@ +using System.Text.Json; +using NATS.Client.Core2.Tests; +using NATS.Client.JetStream.Models; +using NATS.Client.TestUtilities; +using NATS.Client.TestUtilities2; + +namespace NATS.Client.JetStream.Tests; + +[Collection("nats-server")] +public class StreamDomainAdjustmentTest(NatsServerFixture server) +{ + [Fact] + public async Task Stream_operations_should_convert_domain_to_external_api() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var cancellationToken = cts.Token; + + // Track captured payloads for each operation + var capturedPayloads = new List(); + + await using var ms = new MockServer((_, cmd) => + { + if (cmd.Name == "PUB") + { + // Capture the payload sent by the client + var payload = new string(cmd.Buffer!); + + if (cmd.Subject.Contains("STREAM.")) + { + capturedPayloads.Add(payload); + + // Return minimal valid response + cmd.Reply(payload: """{"config":{"name":"s1"},"state":{"messages":0}}"""); + } + } + + return Task.CompletedTask; + }); + + await using var nats = new NatsConnection(new NatsOpts { Url = ms.Url }); + var js = new NatsJSContext(nats); + + // Configure a stream with Source that has Domain set + var streamConfig = new StreamConfig + { + Name = "s1", + Subjects = ["s1"], + Sources = + [ + new StreamSource + { + Name = "x1", + Domain = "TEST_DOMAIN", + } + ], + }; + + // Test all the calls + await js.CreateStreamAsync(streamConfig, cancellationToken); + await js.UpdateStreamAsync(streamConfig, cancellationToken); + await js.CreateOrUpdateStreamAsync(streamConfig, cancellationToken); + + // Verify all payloads were captured + Assert.Equal(3, capturedPayloads.Count); + + // Verify each payload has the correct domain conversion + foreach (var payload in capturedPayloads) + { + // Parse the JSON payload + var doc = JsonDocument.Parse(payload); + var root = doc.RootElement; + + // Verify sources array exists + Assert.True(root.TryGetProperty("sources", out var sources)); + Assert.True(sources.GetArrayLength() > 0); + + var firstSource = sources[0]; + + // Verify domain field does NOT exist (it's client-side only, JsonIgnore) + Assert.False(firstSource.TryGetProperty("domain", out _)); + + // Verify the external.api field exists and has the correct value + Assert.True(firstSource.TryGetProperty("external", out var external)); + Assert.True(external.TryGetProperty("api", out var api)); + Assert.Equal("$JS.TEST_DOMAIN.API", api.GetString()); + } + } + + [Fact] + public async Task Create_stream_with_domain_on_sources_should_populate_external_api() + { + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + var prefix = server.GetNextId(); + await nats.ConnectRetryAsync(); + + var js = new NatsJSContext(nats); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + // Create an aggregate stream with Domain on Sources + var streamConfig = new StreamConfig($"{prefix}aggregate", [$"{prefix}aggregate.*"]) + { + Sources = + [ + new StreamSource + { + Name = $"{prefix}source", + Domain = "TEST_DOMAIN", + } + ], + }; + + var stream = await js.CreateStreamAsync(streamConfig, cts.Token); + + // Verify the returned config has External.Api set, Domain is not preserved + Assert.NotNull(stream.Info.Config.Sources); + Assert.Single(stream.Info.Config.Sources); + var source = stream.Info.Config.Sources.First(); + Assert.NotNull(source.External); + Assert.Equal("$JS.TEST_DOMAIN.API", source.External.Api); + + // Get stream back and verify again + var retrievedStream = await js.GetStreamAsync($"{prefix}aggregate", cancellationToken: cts.Token); + Assert.NotNull(retrievedStream.Info.Config.Sources); + var retrievedSource = retrievedStream.Info.Config.Sources.First(); + Assert.NotNull(retrievedSource.External); + Assert.Equal("$JS.TEST_DOMAIN.API", retrievedSource.External.Api); + } + + [Fact] + public async Task Update_stream_to_add_domain_on_sources_should_populate_external_api() + { + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + var prefix = server.GetNextId(); + await nats.ConnectRetryAsync(); + + var js = new NatsJSContext(nats); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + // Create aggregate stream WITHOUT Domain + var streamConfig = new StreamConfig($"{prefix}aggregate", [$"{prefix}aggregate.*"]) + { + Sources = + [ + new StreamSource + { + Name = $"{prefix}source", + } + ], + }; + + var stream = await js.CreateStreamAsync(streamConfig, cts.Token); + + // Verify initially no External + Assert.NotNull(stream.Info.Config.Sources); + var source = stream.Info.Config.Sources.First(); + Assert.Null(source.External); + + // Update to add Domain + var updatedConfig = streamConfig with + { + Sources = + [ + new StreamSource + { + Name = $"{prefix}source", + Domain = "TEST_DOMAIN", + } + ], + }; + + var updatedStream = await js.UpdateStreamAsync(updatedConfig, cts.Token); + + // Verify External.Api is now set + Assert.NotNull(updatedStream.Info.Config.Sources); + var updatedSource = updatedStream.Info.Config.Sources.First(); + Assert.NotNull(updatedSource.External); + Assert.Equal("$JS.TEST_DOMAIN.API", updatedSource.External.Api); + + // Get stream back and verify + var retrievedStream = await js.GetStreamAsync($"{prefix}aggregate", cancellationToken: cts.Token); + Assert.NotNull(retrievedStream.Info.Config.Sources); + var retrievedSource = retrievedStream.Info.Config.Sources.First(); + Assert.NotNull(retrievedSource.External); + Assert.Equal("$JS.TEST_DOMAIN.API", retrievedSource.External.Api); + } + + [Fact] + public async Task CreateOrUpdate_stream_with_domain_on_sources_should_populate_external_api() + { + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + var prefix = server.GetNextId(); + await nats.ConnectRetryAsync(); + + var js = new NatsJSContext(nats); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + // CreateOrUpdate aggregate stream with Domain on Sources + var streamConfig = new StreamConfig($"{prefix}aggregate", [$"{prefix}aggregate.*"]) + { + Sources = + [ + new StreamSource + { + Name = $"{prefix}source", + Domain = "TEST_DOMAIN", + } + ], + }; + + var stream = await js.CreateOrUpdateStreamAsync(streamConfig, cts.Token); + + // Verify the config has External.Api set + Assert.NotNull(stream.Info.Config.Sources); + Assert.Single(stream.Info.Config.Sources); + var source = stream.Info.Config.Sources.First(); + Assert.NotNull(source.External); + Assert.Equal("$JS.TEST_DOMAIN.API", source.External.Api); + + // Get stream back and verify + var retrievedStream = await js.GetStreamAsync($"{prefix}aggregate", cancellationToken: cts.Token); + Assert.NotNull(retrievedStream.Info.Config.Sources); + var retrievedSource = retrievedStream.Info.Config.Sources.First(); + Assert.NotNull(retrievedSource.External); + Assert.Equal("$JS.TEST_DOMAIN.API", retrievedSource.External.Api); + } +}