Added Domain support for stream mirroring and sourcing and KV full su…

…pport for the same. (#631)

* Added Sources and Mirror to KV Store Creation

* Removed the need for StreamSource.Domain

* Fixed trailing whitespace

* Whitespace fix for KeyValueStoreTest

* Added Domain support for stream sourcing.

* Test fix

* Keep caller's config intact.

* Test fix

* Remove unused import in StreamSource.cs

Deleted the 'System.Diagnostics.Metrics' import as it was not being used anywhere in the file. This change helps in maintaining a clean codebase with only necessary dependencies.

* Remove unused using directive

Deleted an unnecessary directive for System.Xml.Schema in NatsJSContext.cs. This cleanup aids in maintaining clean and efficient code.

* Refactor initialization logic for stream configuration.

Clean up redundant initializations and streamline the handling of `subjects`, `mirror`, and `sources` to improve code clarity. Ensure default assignments are explicitly defined within conditional branches for better code maintainability.

* Refactor mirror assignment with ShallowCopy method

Simplify the mirror object instantiation by using the ShallowCopy method, reducing code redundancy. This change improves readability and maintenance by encapsulating the cloning logic within the method.

* Refactor null and count check for config.Sources

Updated the conditional check for `config.Sources` to use pattern matching, improving readability and adhering to modern C# conventions. This change ensures cleaner and more maintainable code.

* Skip specific test for NATS servers earlier than v2.10

This commit updates the KeyValueStoreTest to skip the Test_CombinedSources test for NATS server versions earlier than 2.10 since some of the mirroring features are introduced with 2.10. It also removes unnecessary retry delay and timeout parameters from the test configuration.

* Use with syntax to clone

* Fix build warnings

* Fix build warnings and add test

* Fix test

---------

Co-authored-by: Ziya Suzen <[email protected]>

.github/workflows/test.yml

@@ -94,6 +94,13 @@ jobs:
           cd tests/NATS.Client.Services.Tests
           dotnet test -c Release --no-build
 
+      - name: Test Simplified
+        run: |
+            killall nats-server 2> /dev/null | echo -n
+            nats-server -v
+            cd tests/NATS.Client.Simplified.Tests
+            dotnet test -c Release --no-build
+
       - name: Test OpenTelemetry
         run: |
           killall nats-server 2> /dev/null | echo -n

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&)'
@@ -88,7 +88,7 @@ private static bool TryWriteNuidCore(Span<char> buffer, Span<char> prefix, ulong
         ref var digitsPtr = ref MemoryMarshal.GetReference(Digits);
 
         // write backwards so the last two characters change the fastest
-        for (var i = NuidLength; i > PrefixLength;)
+        for (nuint i = NuidLength; i > PrefixLength;)
         {
             i--;
             var digitIndex = (nuint)(sequential % Base);

src/NATS.Client.JetStream/Internal/netstandard.cs

@@ -5,10 +5,6 @@
 
 namespace System.Runtime.CompilerServices
 {
-    internal class ExtensionAttribute : Attribute
-    {
-    }
-
     internal sealed class CompilerFeatureRequiredAttribute : Attribute
     {
         public CompilerFeatureRequiredAttribute(string featureName)

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

@@ -54,4 +54,13 @@ 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; }
 }

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 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<StreamSource>? 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;
+        }
+
         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,66 @@ 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 with
+            {
+                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 +278,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);
+    }
 }

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

@@ -50,6 +50,8 @@ public class MockJsContext : INatsJSContext
     {
         public INatsConnection Connection { get; } = new NatsConnection();
 
+        public NatsJSOpts Opts { get; } = new(new NatsOpts());
+
         public ValueTask<INatsJSConsumer> CreateOrderedConsumerAsync(string stream, NatsJSOrderedConsumerOpts? opts = default, CancellationToken cancellationToken = default) => throw new NotImplementedException();
 
         public ValueTask<INatsJSConsumer> CreateOrUpdateConsumerAsync(string stream, ConsumerConfig config, CancellationToken cancellationToken = default) => throw new NotImplementedException();

tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs

@@ -50,6 +50,8 @@ public class MockJsContext : INatsJSContext
     {
         public INatsConnection Connection { get; } = new NatsConnection();
 
+        public NatsJSOpts Opts { get; } = new(new NatsOpts());
+
         public ValueTask<INatsJSConsumer> CreateOrderedConsumerAsync(string stream, NatsJSOrderedConsumerOpts? opts = default, CancellationToken cancellationToken = default) => throw new NotImplementedException();
 
         public ValueTask<INatsJSConsumer> CreateOrUpdateConsumerAsync(string stream, ConsumerConfig config, CancellationToken cancellationToken = default) => throw new NotImplementedException();

tests/NATS.Client.Platform.Windows.Tests/NATS.Client.Platform.Windows.Tests.csproj

@@ -16,6 +16,8 @@
         <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.8.0"/>
         <PackageReference Include="xunit" Version="2.5.3"/>
         <PackageReference Include="xunit.runner.visualstudio" Version="2.5.3"/>
+        <PackageReference Include="System.Net.Http" Version="4.3.4" />
+        <PackageReference Include="System.Text.RegularExpressions" Version="4.3.1" />
     </ItemGroup>
 
     <ItemGroup>

tests/NATS.Client.Simplified.Tests/test.runsettings

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<RunSettings>
+    <RunConfiguration>
+        <MaxCpuCount>1</MaxCpuCount>
+        <TestSessionTimeout>300000</TestSessionTimeout>
+    </RunConfiguration>
+</RunSettings>