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

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

@@ -54,4 +54,7 @@ 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; }
+
+    [System.Text.Json.Serialization.JsonIgnore]
+    public string? Domain { get; set; }
 }

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

@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Text.RegularExpressions;
 using NATS.Client.JetStream;
@@ -173,9 +174,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 +201,60 @@ private static StreamConfig CreateStreamConfig(NatsKVConfig config)
 
         var replicas = config.NumberOfReplicas > 0 ? config.NumberOfReplicas : 1;
 
+        string[]? subjects = default;
+        StreamSource? mirror = default;
+        ICollection<StreamSource>? sources = default;
+        var mirrorDirect = false;
+        if (config.Mirror != null)
+        {
+            mirror = new StreamSource
+            {
+                Name = config.Mirror.Name.StartsWith(KvStreamNamePrefix) ? config.Mirror.Name : BucketToStream(config.Mirror.Name),
+                Domain = config.Mirror.Domain,
+                External = config.Mirror.External,
+                OptStartSeq = config.Mirror.OptStartSeq,
+                OptStartTime = config.Mirror.OptStartTime,
+                SubjectTransforms = config.Mirror.SubjectTransforms,
+                FilterSubject = config.Mirror.FilterSubject,
+            };
+            mirrorDirect = true;
+        }
+        else if (config.Sources != null && config.Sources.Count > 0)
+        {
+            sources = [];
+            foreach (var ss in config.Sources)
+            {
+                string? sourceBucketName = default;
+                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}.>"];
+        }
+        else
+        {
+            subjects = [$"$KV.{config.Bucket}.>"];
+        }
+
         var streamConfig = new StreamConfig
         {
             Name = BucketToStream(config.Bucket),
@@ -221,10 +273,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,37 @@ public async Task TestDirectMessageRepublishedSubject()
         Assert.Equal(publishSubject3, kve3.Key);
         Assert.Equal("tres", kve3.Value);
     }
+
+    [Fact]
+    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");
+
+        // ensure any async replication
+        await Task.Delay(1000);
+
+        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);
+    }
 }