Added Domain support for stream mirroring and sourcing and KV full support for the same.

src/NATS.Client.Core/Nuid.cs

@@ -18,7 +18,7 @@ namespace NATS.Client.Core;
 [SkipLocalsInit]
 public sealed class Nuid
 {
-    // NuidLength, PrefixLength, SequentialLength were nuint (System.UIntPtr) in the original code
+    // NuidLength, PrefixLength, SequentialLength were nuint (System.UIntPtr) in the original code,
     // however, they were changed to uint to fix the compilation error for IL2CPP Unity projects.
     // With nuint, the following error occurs in Unity Linux IL2CPP builds:
     //   Error: IL2CPP error for method 'System.Char[] NATS.Client.Core.Internal.NuidWriter::Refresh(System.UInt64&)'

src/NATS.Client.JetStream/Models/StreamConfig.cs

@@ -245,4 +245,12 @@ internal StreamConfig()
     [System.Text.Json.Serialization.JsonPropertyName("metadata")]
     [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
     public IDictionary<string, string>? Metadata { get; set; }
+
+    /// <summary>
+    /// Creates a shallow copy of the current StreamConfig instance using the MemberwiseClone method.
+    /// </summary>
+    /// <return>
+    /// A shallow copy of the current StreamConfig.
+    /// </return>
+    public StreamConfig ShallowCopy() => (StreamConfig)MemberwiseClone();
 }

src/NATS.Client.JetStream/Models/StreamSource.cs

@@ -54,4 +54,21 @@ public record StreamSource
     [System.Text.Json.Serialization.JsonPropertyName("external")]
     [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
     public ExternalStreamSource? External { get; set; }
+
+    /// <summary>
+    /// This field is a convenience for setting up an ExternalStream.
+    /// If set, the value here is used to calculate the JetStreamAPI prefix.
+    /// This field is never serialized to the server. This value cannot be set
+    /// if external is set.
+    /// </summary>
+    [System.Text.Json.Serialization.JsonIgnore]
+    public string? Domain { get; set; }
+
+    /// <summary>
+    /// Creates a shallow copy of the current StreamSource instance using the MemberwiseClone method.
+    /// </summary>
+    /// <return>
+    /// A shallow copy of the current StreamSource.
+    /// </return>
+    public StreamSource ShallowCopy() => (StreamSource)MemberwiseClone();
 }

src/NATS.Client.JetStream/NatsJSContext.Streams.cs

@@ -20,6 +20,38 @@ public async ValueTask<INatsJSStream> CreateStreamAsync(
         CancellationToken cancellationToken = default)
     {
         ThrowIfInvalidStreamName(config.Name, nameof(config.Name));
+
+        // keep caller's config intact.
+        config = config.ShallowCopy();
+
+        // 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.ShallowCopy();
+            ConvertDomain(config.Mirror);
+        }
+
+        // Check sources for the same.
+        if (config.Sources != null && config.Sources.Count > 0)
+        {
+            ICollection<StreamSource>? sources = [];
+            foreach (var ss in config.Sources)
+            {
+                if (!string.IsNullOrEmpty(ss.Domain))
+                {
+                    var remappedDomainSource = ss.ShallowCopy();
+                    ConvertDomain(remappedDomainSource);
+                    sources.Add(remappedDomainSource);
+                }
+                else
+                {
+                    sources.Add(ss);
+                }
+            }
+
+            config.Sources = sources;
+        }
+
         var response = await JSRequestResponseAsync<StreamConfig, StreamInfo>(
             subject: $"{Opts.Prefix}.STREAM.CREATE.{config.Name}",
             config,

src/NATS.Client.JetStream/NatsJSContext.cs

@@ -327,6 +327,21 @@ internal async ValueTask<NatsJSResponse<TResponse>> JSRequestAsync<TRequest, TRe
         throw new NatsJSApiNoResponseException();
     }
 
+    private static void ConvertDomain(StreamSource streamSource)
+    {
+        if (string.IsNullOrEmpty(streamSource.Domain))
+        {
+            return;
+        }
+
+        if (streamSource.External != null)
+        {
+            throw new ArgumentException("Both domain and external are set");
+        }
+
+        streamSource.External = new ExternalStreamSource { Api = $"$JS.{streamSource.Domain}.API" };
+    }
+
     [DoesNotReturn]
     private static void ThrowInvalidStreamNameException(string? paramName) =>
         throw new ArgumentException("Stream name cannot contain ' ', '.'", paramName);

src/NATS.Client.KeyValueStore/NatsKVConfig.cs

@@ -1,3 +1,5 @@
+using NATS.Client.JetStream.Models;
+
 namespace NATS.Client.KeyValueStore;
 
 /// <summary>
@@ -61,12 +63,15 @@ public record NatsKVConfig
     /// </summary>
     public bool Compression { get; init; }
 
-    // TODO: Bucket mirror configuration.
-    // pub mirror: Option<Source>,
-    // Bucket sources configuration.
-    // pub sources: Option<Vec<Source>>,
-    // Allow mirrors using direct API.
-    // pub mirror_direct: bool,
+    /// <summary>
+    /// Mirror defines the configuration for mirroring another KeyValue store
+    /// </summary>
+    public StreamSource? Mirror { get; init; }
+
+    /// <summary>
+    /// Sources defines the configuration for sources of a KeyValue store.
+    /// </summary>
+    public ICollection<StreamSource>? Sources { get; set; }
 }
 
 /// <summary>

src/NATS.Client.KeyValueStore/NatsKVContext.cs

@@ -173,9 +173,6 @@ private static string ExtractBucketName(string streamName)
 
     private static StreamConfig CreateStreamConfig(NatsKVConfig config)
     {
-        // TODO: KV Mirrors
-        var subjects = new[] { $"$KV.{config.Bucket}.>" };
-
         long history;
         if (config.History > 0)
         {
@@ -203,6 +200,64 @@ private static StreamConfig CreateStreamConfig(NatsKVConfig config)
 
         var replicas = config.NumberOfReplicas > 0 ? config.NumberOfReplicas : 1;
 
+        string[]? subjects;
+        StreamSource? mirror;
+        ICollection<StreamSource>? sources;
+        bool mirrorDirect;
+
+        if (config.Mirror != null)
+        {
+            mirror = config.Mirror.ShallowCopy();
+            mirror.Name = config.Mirror.Name.StartsWith(KvStreamNamePrefix)
+                ? config.Mirror.Name
+                : BucketToStream(config.Mirror.Name);
+            mirrorDirect = true;
+            subjects = default;
+            sources = default;
+        }
+        else if (config.Sources is { Count: > 0 })
+        {
+            sources = [];
+            foreach (var ss in config.Sources)
+            {
+                string? sourceBucketName;
+                if (ss.Name.StartsWith(KvStreamNamePrefix))
+                {
+                    sourceBucketName = ss.Name.Substring(KvStreamNamePrefixLen);
+                }
+                else
+                {
+                    sourceBucketName = ss.Name;
+                    ss.Name = BucketToStream(ss.Name);
+                }
+
+                if (ss.External == null || sourceBucketName != config.Bucket)
+                {
+                    ss.SubjectTransforms =
+                    [
+                        new SubjectTransform
+                        {
+                            Src = $"$KV.{sourceBucketName}.>",
+                            Dest = $"$KV.{config.Bucket}.>",
+                        }
+                    ];
+                }
+
+                sources.Add(ss);
+            }
+
+            subjects = [$"$KV.{config.Bucket}.>"];
+            mirror = default;
+            mirrorDirect = false;
+        }
+        else
+        {
+            subjects = [$"$KV.{config.Bucket}.>"];
+            mirror = default;
+            sources = default;
+            mirrorDirect = false;
+        }
+
         var streamConfig = new StreamConfig
         {
             Name = BucketToStream(config.Bucket),
@@ -221,10 +276,9 @@ private static StreamConfig CreateStreamConfig(NatsKVConfig config)
             AllowDirect = true,
             NumReplicas = replicas,
             Discard = StreamConfigDiscard.New,
-
-            // TODO: KV mirrors
-            // MirrorDirect =
-            // Mirror =
+            Mirror = mirror,
+            MirrorDirect = mirrorDirect,
+            Sources = sources,
             Retention = StreamConfigRetention.Limits, // from ADR-8
         };
 

tests/NATS.Client.KeyValueStore.Tests/KeyValueStoreTest.cs

@@ -648,4 +648,51 @@ public async Task TestDirectMessageRepublishedSubject()
         Assert.Equal(publishSubject3, kve3.Key);
         Assert.Equal("tres", kve3.Value);
     }
+
+    [SkipIfNatsServer(versionEarlierThan: "2.10")]
+    public async Task Test_CombinedSources()
+    {
+        await using var server = NatsServer.StartJS();
+        await using var nats = server.CreateClientConnection();
+
+        var js = new NatsJSContext(nats);
+        var kv = new NatsKVContext(js);
+
+        var storeSource1 = await kv.CreateStoreAsync("source1");
+        var storeSource2 = await kv.CreateStoreAsync("source2");
+
+        var storeCombined = await kv.CreateStoreAsync(new NatsKVConfig("combined")
+        {
+            Sources = [
+                new StreamSource { Name = "source1" },
+                new StreamSource { Name = "source2" }
+            ],
+        });
+
+        await storeSource1.PutAsync("ss1_a", "a_fromStore1");
+        await storeSource2.PutAsync("ss2_b", "b_fromStore2");
+
+        await Retry.Until(
+            "async replication is completed",
+            async () =>
+            {
+                try
+                {
+                    await storeCombined.GetEntryAsync<string>("ss1_a");
+                    await storeCombined.GetEntryAsync<string>("ss2_b");
+                }
+                catch (NatsKVKeyNotFoundException)
+                {
+                    return false;
+                }
+
+                return true;
+            });
+
+        var entryA = await storeCombined.GetEntryAsync<string>("ss1_a");
+        var entryB = await storeCombined.GetEntryAsync<string>("ss2_b");
+
+        Assert.Equal("a_fromStore1", entryA.Value);
+        Assert.Equal("b_fromStore2", entryB.Value);
+    }
 }