Skip to content

Commit bb2537a

Browse files
darkwatchukmtmk
darkwatchuk
and
mtmk
authored
Added Domain support for stream mirroring and sourcing and KV full support 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 <ziya@suzen.net>
1 parent 06156f9 commit bb2537a

File tree

13 files changed

+199
-19
lines changed

13 files changed

+199
-19
lines changed

‎.github/workflows/test.yml

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,13 @@ jobs:
9494
cd tests/NATS.Client.Services.Tests
9595
dotnet test -c Release --no-build
9696
97+
- name: Test Simplified
98+
run: |
99+
killall nats-server 2> /dev/null | echo -n
100+
nats-server -v
101+
cd tests/NATS.Client.Simplified.Tests
102+
dotnet test -c Release --no-build
103+
97104
- name: Test OpenTelemetry
98105
run: |
99106
killall nats-server 2> /dev/null | echo -n

‎src/NATS.Client.Core/Nuid.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ namespace NATS.Client.Core;
1818
[SkipLocalsInit]
1919
public sealed class Nuid
2020
{
21-
// NuidLength, PrefixLength, SequentialLength were nuint (System.UIntPtr) in the original code
21+
// NuidLength, PrefixLength, SequentialLength were nuint (System.UIntPtr) in the original code,
2222
// however, they were changed to uint to fix the compilation error for IL2CPP Unity projects.
2323
// With nuint, the following error occurs in Unity Linux IL2CPP builds:
2424
// 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
8888
ref var digitsPtr = ref MemoryMarshal.GetReference(Digits);
8989

9090
// write backwards so the last two characters change the fastest
91-
for (var i = NuidLength; i > PrefixLength;)
91+
for (nuint i = NuidLength; i > PrefixLength;)
9292
{
9393
i--;
9494
var digitIndex = (nuint)(sequential % Base);

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

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,6 @@
55

66
namespace System.Runtime.CompilerServices
77
{
8-
internal class ExtensionAttribute : Attribute
9-
{
10-
}
11-
128
internal sealed class CompilerFeatureRequiredAttribute : Attribute
139
{
1410
public CompilerFeatureRequiredAttribute(string featureName)

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,4 +54,13 @@ public record StreamSource
5454
[System.Text.Json.Serialization.JsonPropertyName("external")]
5555
[System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)]
5656
public ExternalStreamSource? External { get; set; }
57+
58+
/// <summary>
59+
/// This field is a convenience for setting up an ExternalStream.
60+
/// If set, the value here is used to calculate the JetStreamAPI prefix.
61+
/// This field is never serialized to the server. This value cannot be set
62+
/// if external is set.
63+
/// </summary>
64+
[System.Text.Json.Serialization.JsonIgnore]
65+
public string? Domain { get; set; }
5766
}

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

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,38 @@ public async ValueTask<INatsJSStream> CreateStreamAsync(
2020
CancellationToken cancellationToken = default)
2121
{
2222
ThrowIfInvalidStreamName(config.Name, nameof(config.Name));
23+
24+
// keep caller's config intact.
25+
config = config with { };
26+
27+
// If we have a mirror and an external domain, convert to ext.APIPrefix.
28+
if (config.Mirror != null && !string.IsNullOrEmpty(config.Mirror.Domain))
29+
{
30+
config.Mirror = config.Mirror with { };
31+
ConvertDomain(config.Mirror);
32+
}
33+
34+
// Check sources for the same.
35+
if (config.Sources != null && config.Sources.Count > 0)
36+
{
37+
ICollection<StreamSource>? sources = [];
38+
foreach (var ss in config.Sources)
39+
{
40+
if (!string.IsNullOrEmpty(ss.Domain))
41+
{
42+
var remappedDomainSource = ss with { };
43+
ConvertDomain(remappedDomainSource);
44+
sources.Add(remappedDomainSource);
45+
}
46+
else
47+
{
48+
sources.Add(ss);
49+
}
50+
}
51+
52+
config.Sources = sources;
53+
}
54+
2355
var response = await JSRequestResponseAsync<StreamConfig, StreamInfo>(
2456
subject: $"{Opts.Prefix}.STREAM.CREATE.{config.Name}",
2557
config,

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -327,6 +327,21 @@ internal async ValueTask<NatsJSResponse<TResponse>> JSRequestAsync<TRequest, TRe
327327
throw new NatsJSApiNoResponseException();
328328
}
329329

330+
private static void ConvertDomain(StreamSource streamSource)
331+
{
332+
if (string.IsNullOrEmpty(streamSource.Domain))
333+
{
334+
return;
335+
}
336+
337+
if (streamSource.External != null)
338+
{
339+
throw new ArgumentException("Both domain and external are set");
340+
}
341+
342+
streamSource.External = new ExternalStreamSource { Api = $"$JS.{streamSource.Domain}.API" };
343+
}
344+
330345
[DoesNotReturn]
331346
private static void ThrowInvalidStreamNameException(string? paramName) =>
332347
throw new ArgumentException("Stream name cannot contain ' ', '.'", paramName);

‎src/NATS.Client.KeyValueStore/NatsKVConfig.cs

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
using NATS.Client.JetStream.Models;
2+
13
namespace NATS.Client.KeyValueStore;
24

35
/// <summary>
@@ -61,12 +63,15 @@ public record NatsKVConfig
6163
/// </summary>
6264
public bool Compression { get; init; }
6365

64-
// TODO: Bucket mirror configuration.
65-
// pub mirror: Option<Source>,
66-
// Bucket sources configuration.
67-
// pub sources: Option<Vec<Source>>,
68-
// Allow mirrors using direct API.
69-
// pub mirror_direct: bool,
66+
/// <summary>
67+
/// Mirror defines the configuration for mirroring another KeyValue store
68+
/// </summary>
69+
public StreamSource? Mirror { get; init; }
70+
71+
/// <summary>
72+
/// Sources defines the configuration for sources of a KeyValue store.
73+
/// </summary>
74+
public ICollection<StreamSource>? Sources { get; set; }
7075
}
7176

7277
/// <summary>

‎src/NATS.Client.KeyValueStore/NatsKVContext.cs

Lines changed: 63 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -173,9 +173,6 @@ private static string ExtractBucketName(string streamName)
173173

174174
private static StreamConfig CreateStreamConfig(NatsKVConfig config)
175175
{
176-
// TODO: KV Mirrors
177-
var subjects = new[] { $"$KV.{config.Bucket}.>" };
178-
179176
long history;
180177
if (config.History > 0)
181178
{
@@ -203,6 +200,66 @@ private static StreamConfig CreateStreamConfig(NatsKVConfig config)
203200

204201
var replicas = config.NumberOfReplicas > 0 ? config.NumberOfReplicas : 1;
205202

203+
string[]? subjects;
204+
StreamSource? mirror;
205+
ICollection<StreamSource>? sources;
206+
bool mirrorDirect;
207+
208+
if (config.Mirror != null)
209+
{
210+
mirror = config.Mirror with
211+
{
212+
Name = config.Mirror.Name.StartsWith(KvStreamNamePrefix)
213+
? config.Mirror.Name
214+
: BucketToStream(config.Mirror.Name),
215+
};
216+
mirrorDirect = true;
217+
subjects = default;
218+
sources = default;
219+
}
220+
else if (config.Sources is { Count: > 0 })
221+
{
222+
sources = [];
223+
foreach (var ss in config.Sources)
224+
{
225+
string? sourceBucketName;
226+
if (ss.Name.StartsWith(KvStreamNamePrefix))
227+
{
228+
sourceBucketName = ss.Name.Substring(KvStreamNamePrefixLen);
229+
}
230+
else
231+
{
232+
sourceBucketName = ss.Name;
233+
ss.Name = BucketToStream(ss.Name);
234+
}
235+
236+
if (ss.External == null || sourceBucketName != config.Bucket)
237+
{
238+
ss.SubjectTransforms =
239+
[
240+
new SubjectTransform
241+
{
242+
Src = $"$KV.{sourceBucketName}.>",
243+
Dest = $"$KV.{config.Bucket}.>",
244+
}
245+
];
246+
}
247+
248+
sources.Add(ss);
249+
}
250+
251+
subjects = [$"$KV.{config.Bucket}.>"];
252+
mirror = default;
253+
mirrorDirect = false;
254+
}
255+
else
256+
{
257+
subjects = [$"$KV.{config.Bucket}.>"];
258+
mirror = default;
259+
sources = default;
260+
mirrorDirect = false;
261+
}
262+
206263
var streamConfig = new StreamConfig
207264
{
208265
Name = BucketToStream(config.Bucket),
@@ -221,10 +278,9 @@ private static StreamConfig CreateStreamConfig(NatsKVConfig config)
221278
AllowDirect = true,
222279
NumReplicas = replicas,
223280
Discard = StreamConfigDiscard.New,
224-
225-
// TODO: KV mirrors
226-
// MirrorDirect =
227-
// Mirror =
281+
Mirror = mirror,
282+
MirrorDirect = mirrorDirect,
283+
Sources = sources,
228284
Retention = StreamConfigRetention.Limits, // from ADR-8
229285
};
230286

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

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -648,4 +648,51 @@ public async Task TestDirectMessageRepublishedSubject()
648648
Assert.Equal(publishSubject3, kve3.Key);
649649
Assert.Equal("tres", kve3.Value);
650650
}
651+
652+
[SkipIfNatsServer(versionEarlierThan: "2.10")]
653+
public async Task Test_CombinedSources()
654+
{
655+
await using var server = NatsServer.StartJS();
656+
await using var nats = server.CreateClientConnection();
657+
658+
var js = new NatsJSContext(nats);
659+
var kv = new NatsKVContext(js);
660+
661+
var storeSource1 = await kv.CreateStoreAsync("source1");
662+
var storeSource2 = await kv.CreateStoreAsync("source2");
663+
664+
var storeCombined = await kv.CreateStoreAsync(new NatsKVConfig("combined")
665+
{
666+
Sources = [
667+
new StreamSource { Name = "source1" },
668+
new StreamSource { Name = "source2" }
669+
],
670+
});
671+
672+
await storeSource1.PutAsync("ss1_a", "a_fromStore1");
673+
await storeSource2.PutAsync("ss2_b", "b_fromStore2");
674+
675+
await Retry.Until(
676+
"async replication is completed",
677+
async () =>
678+
{
679+
try
680+
{
681+
await storeCombined.GetEntryAsync<string>("ss1_a");
682+
await storeCombined.GetEntryAsync<string>("ss2_b");
683+
}
684+
catch (NatsKVKeyNotFoundException)
685+
{
686+
return false;
687+
}
688+
689+
return true;
690+
});
691+
692+
var entryA = await storeCombined.GetEntryAsync<string>("ss1_a");
693+
var entryB = await storeCombined.GetEntryAsync<string>("ss2_b");
694+
695+
Assert.Equal("a_fromStore1", entryA.Value);
696+
Assert.Equal("b_fromStore2", entryB.Value);
697+
}
651698
}

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ public class MockJsContext : INatsJSContext
5050
{
5151
public INatsConnection Connection { get; } = new NatsConnection();
5252

53+
public NatsJSOpts Opts { get; } = new(new NatsOpts());
54+
5355
public ValueTask<INatsJSConsumer> CreateOrderedConsumerAsync(string stream, NatsJSOrderedConsumerOpts? opts = default, CancellationToken cancellationToken = default) => throw new NotImplementedException();
5456

5557
public ValueTask<INatsJSConsumer> CreateOrUpdateConsumerAsync(string stream, ConsumerConfig config, CancellationToken cancellationToken = default) => throw new NotImplementedException();

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ public class MockJsContext : INatsJSContext
5050
{
5151
public INatsConnection Connection { get; } = new NatsConnection();
5252

53+
public NatsJSOpts Opts { get; } = new(new NatsOpts());
54+
5355
public ValueTask<INatsJSConsumer> CreateOrderedConsumerAsync(string stream, NatsJSOrderedConsumerOpts? opts = default, CancellationToken cancellationToken = default) => throw new NotImplementedException();
5456

5557
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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.8.0"/>
1717
<PackageReference Include="xunit" Version="2.5.3"/>
1818
<PackageReference Include="xunit.runner.visualstudio" Version="2.5.3"/>
19+
<PackageReference Include="System.Net.Http" Version="4.3.4" />
20+
<PackageReference Include="System.Text.RegularExpressions" Version="4.3.1" />
1921
</ItemGroup>
2022

2123
<ItemGroup>

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
<?xml version="1.0" encoding="utf-8" ?>
2+
<RunSettings>
3+
<RunConfiguration>
4+
<MaxCpuCount>1</MaxCpuCount>
5+
<TestSessionTimeout>300000</TestSessionTimeout>
6+
</RunConfiguration>
7+
</RunSettings>

0 commit comments

Comments
 (0)