From 6466d22678957d1253e428c9eecd0ee14f8d80c7 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Fri, 17 Jan 2025 15:29:04 +0000 Subject: [PATCH 01/12] Fix telemetry header readonly error --- src/NATS.Client.Core/Internal/Telemetry.cs | 6 +- .../NatsConnection.Publish.cs | 1 + src/NATS.Client.Core/NatsHeaders.cs | 18 ++ .../resources/configs/restricted-user.conf | 5 + .../NatsServerFixture.cs | 16 +- .../NATS.Client.JetStream.Tests.csproj | 7 + .../PublishRetryTest.cs | 154 ++++++++++++++++++ 7 files changed, 204 insertions(+), 3 deletions(-) create mode 100644 tests/NATS.Client.Core.Tests/resources/configs/restricted-user.conf create mode 100644 tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs diff --git a/src/NATS.Client.Core/Internal/Telemetry.cs b/src/NATS.Client.Core/Internal/Telemetry.cs index 5c5cb2fb7..fe781f5a4 100644 --- a/src/NATS.Client.Core/Internal/Telemetry.cs +++ b/src/NATS.Client.Core/Internal/Telemetry.cs @@ -87,7 +87,11 @@ internal static void AddTraceContextHeaders(Activity? activity, ref NatsHeaders? return; } - headers[fieldName] = fieldValue; + // There are cases where headers reused internally (e.g. JetStream publish retry) + // there may also be cases where application can reuse the same header + // in which case we should still be able to overwrite headers with telemetry fields + // even though headers would be set to readonly before being passed down in publish methods. + headers.SetOverrideReadOnly(fieldName, fieldValue); }); } diff --git a/src/NATS.Client.Core/NatsConnection.Publish.cs b/src/NATS.Client.Core/NatsConnection.Publish.cs index abe0fbf7b..1f5312413 100644 --- a/src/NATS.Client.Core/NatsConnection.Publish.cs +++ b/src/NATS.Client.Core/NatsConnection.Publish.cs @@ -11,6 +11,7 @@ public ValueTask PublishAsync(string subject, NatsHeaders? headers = default, st if (Telemetry.HasListeners()) { using var activity = Telemetry.StartSendActivity($"{SpanDestinationName(subject)} {Telemetry.Constants.PublishActivityName}", this, subject, replyTo); + Telemetry.AddTraceContextHeaders(activity, ref headers); try { headers?.SetReadOnly(); diff --git a/src/NATS.Client.Core/NatsHeaders.cs b/src/NATS.Client.Core/NatsHeaders.cs index d8d447edb..eda561d88 100644 --- a/src/NATS.Client.Core/NatsHeaders.cs +++ b/src/NATS.Client.Core/NatsHeaders.cs @@ -423,6 +423,24 @@ IEnumerator IEnumerable.GetEnumerator() internal void SetReadOnly() => Interlocked.Exchange(ref _readonly, 1); + internal void SetOverrideReadOnly(string key, StringValues value) + { + if (key == null) + { + throw new ArgumentNullException(nameof(key)); + } + + if (value.Count == 0) + { + Store?.Remove(key); + } + else + { + EnsureStore(1); + Store[key] = value; + } + } + private void ThrowIfReadOnly() { if (IsReadOnly) diff --git a/tests/NATS.Client.Core.Tests/resources/configs/restricted-user.conf b/tests/NATS.Client.Core.Tests/resources/configs/restricted-user.conf new file mode 100644 index 000000000..4c8a6ab04 --- /dev/null +++ b/tests/NATS.Client.Core.Tests/resources/configs/restricted-user.conf @@ -0,0 +1,5 @@ +authorization: { + users: [ + { user:u, permissions: { publish:x } } + ] +} diff --git a/tests/NATS.Client.Core2.Tests/NatsServerFixture.cs b/tests/NATS.Client.Core2.Tests/NatsServerFixture.cs index ea92dae6c..24d679005 100644 --- a/tests/NATS.Client.Core2.Tests/NatsServerFixture.cs +++ b/tests/NATS.Client.Core2.Tests/NatsServerFixture.cs @@ -8,8 +8,13 @@ public class NatsServerFixture : IDisposable private int _next; public NatsServerFixture() + : this(null) { - Server = NatsServerProcess.Start(); + } + + protected NatsServerFixture(string? config) + { + Server = NatsServerProcess.Start(config: config); } public NatsServerProcess Server { get; } @@ -27,6 +32,13 @@ public void Dispose() } [CollectionDefinition("nats-server")] -public class DatabaseCollection : ICollectionFixture +public class NatsServerCollection : ICollectionFixture +{ +} + +public class NatsServerRestrictedUserFixture() : NatsServerFixture("resources/configs/restricted-user.conf"); + +[CollectionDefinition("nats-server-restricted-user")] +public class NatsServerRestrictedUserCollection : ICollectionFixture { } diff --git a/tests/NATS.Client.JetStream.Tests/NATS.Client.JetStream.Tests.csproj b/tests/NATS.Client.JetStream.Tests/NATS.Client.JetStream.Tests.csproj index cff8c2b18..47b49207e 100644 --- a/tests/NATS.Client.JetStream.Tests/NATS.Client.JetStream.Tests.csproj +++ b/tests/NATS.Client.JetStream.Tests/NATS.Client.JetStream.Tests.csproj @@ -32,6 +32,13 @@ + + + resources\%(RecursiveDir)%(Filename)%(Extension) + PreserveNewest + + + diff --git a/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs b/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs new file mode 100644 index 000000000..6523726cc --- /dev/null +++ b/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs @@ -0,0 +1,154 @@ +using System.Diagnostics; +using NATS.Client.Core2.Tests; +using NATS.Net; + +namespace NATS.Client.JetStream.Tests; + + +[Collection("nats-server-restricted-user")] +public class PublishRetryTest +{ + private readonly ITestOutputHelper _output; + private readonly NatsServerRestrictedUserFixture _server; + + public PublishRetryTest(ITestOutputHelper output, NatsServerRestrictedUserFixture server) + { + _output = output; + _server = server; + } + + [Fact] + public async Task Publish_with_or_without_telemetry() + { + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await using var nats = new NatsConnection(new NatsOpts + { + Url = _server.Url, + AuthOpts = new NatsAuthOpts { Username = "u" }, + }); + var prefix = _server.GetNextId(); + + // With telemetry + { + using var activityListener = new ActivityListener { ShouldListenTo = _ => true, Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, }; + ActivitySource.AddActivityListener(activityListener); + var headers = new NatsHeaders(); + await nats.PublishAsync(prefix, headers: headers, cancellationToken: cts.Token); + Assert.Contains("traceparent", headers); + } + + // Without telemetry + { + var headers = new NatsHeaders(); + await nats.PublishAsync(prefix, headers: headers, cancellationToken: cts.Token); + Assert.DoesNotContain("traceparent", headers); + Assert.Empty(headers); + } + + // With telemetry and data + { + using var activityListener = new ActivityListener { ShouldListenTo = _ => true, Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, }; + ActivitySource.AddActivityListener(activityListener); + var headers = new NatsHeaders(); + await nats.PublishAsync(prefix, data: 1, headers: headers, cancellationToken: cts.Token); + Assert.Contains("traceparent", headers); + } + + // Without telemetry and data + { + var headers = new NatsHeaders(); + await nats.PublishAsync(prefix, data: 1, headers: headers, cancellationToken: cts.Token); + Assert.DoesNotContain("traceparent", headers); + Assert.Empty(headers); + } + + } + + [Fact] + public async Task Multiple_publish_with_same_headers_when_telemetry_on_should_not_throw_header_readonly_exception() + { + using var activityListener = new ActivityListener + { + ShouldListenTo = _ => true, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, + }; + ActivitySource.AddActivityListener(activityListener); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await using var nats = new NatsConnection(new NatsOpts + { + Url = _server.Url, + AuthOpts = new NatsAuthOpts { Username = "u" }, + }); + var prefix = _server.GetNextId(); + + // Publish with no data implemented in a different method + { + var headers = new NatsHeaders(); + await nats.PublishAsync(prefix, headers: headers, cancellationToken: cts.Token); + await nats.PublishAsync(prefix, headers: headers, cancellationToken: cts.Token); + Assert.Contains("traceparent", headers); + } + + // Also try publish-with-data method + { + var headers = new NatsHeaders(); + await nats.PublishAsync(prefix, data: 1, headers: headers, cancellationToken: cts.Token); + await nats.PublishAsync(prefix, data: 1, headers: headers, cancellationToken: cts.Token); + Assert.Contains("traceparent", headers); + } + } + + [Fact] + public async Task Retry_with_telemetry_on_should_not_throw_header_readonly_exception() + { + using var activityListener = new ActivityListener + { + ShouldListenTo = _ => true, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, + }; + ActivitySource.AddActivityListener(activityListener); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await using var nats = new NatsConnection(new NatsOpts + { + Url = _server.Url, + AuthOpts = new NatsAuthOpts { Username = "u" }, + RequestTimeout = TimeSpan.FromSeconds(.5), + }); + var prefix = _server.GetNextId(); + var js = nats.CreateJetStreamContext(); + + // Because of the permission error (associated user should only have permission to subject 'x' to publish) + // JetStream publish internally retry publishing again which in turn would use the same header. + // Telemetry altering headers should not cause 'readonly' exceptions. + { + var headers = new NatsHeaders(); + var exception = await Assert.ThrowsAnyAsync(async () => + { + await js.PublishAsync( + subject: $"{prefix}.foo", + data: "order 1", + headers: headers, + cancellationToken: cts.Token); + }); + Assert.IsNotType(exception); + Assert.DoesNotMatch("response headers cannot be modified", exception.Message); + Assert.Contains("traceparent", headers); + } + + // Also check for specific exception + { + var headers = new NatsHeaders(); + await Assert.ThrowsAsync(async () => + { + await js.PublishAsync( + subject: $"{prefix}.foo", + data: "order 1", + headers: headers, + cancellationToken: cts.Token); + }); + Assert.Contains("traceparent", headers); + } + } +} From d019eae3ce283aacdc8ce0b0dc7a0efae1871919 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Fri, 17 Jan 2025 15:35:53 +0000 Subject: [PATCH 02/12] Format --- tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs b/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs index 6523726cc..f1a3a68ec 100644 --- a/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs +++ b/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs @@ -1,10 +1,9 @@ -using System.Diagnostics; +using System.Diagnostics; using NATS.Client.Core2.Tests; using NATS.Net; namespace NATS.Client.JetStream.Tests; - [Collection("nats-server-restricted-user")] public class PublishRetryTest { @@ -61,7 +60,6 @@ public async Task Publish_with_or_without_telemetry() Assert.DoesNotContain("traceparent", headers); Assert.Empty(headers); } - } [Fact] From 4c2b5464377beb389386a0300215774b22796564 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 00:48:43 +0000 Subject: [PATCH 03/12] Fix missing test project in solution file --- NATS.Net.sln | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/NATS.Net.sln b/NATS.Net.sln index 9c78cc340..4d281f6a4 100644 --- a/NATS.Net.sln +++ b/NATS.Net.sln @@ -139,6 +139,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "workflows", "workflows", "{ .github\workflows\test.yml = .github\workflows\test.yml EndProjectSection EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATS.Client.TestUtilities2", "tests\NATS.Client.TestUtilities2\NATS.Client.TestUtilities2.csproj", "{25E4C70A-196C-4855-8DFC-796C08837987}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -341,6 +343,10 @@ Global {9521D9E0-642A-4C7E-BD10-372DF235CF62}.Debug|Any CPU.Build.0 = Debug|Any CPU {9521D9E0-642A-4C7E-BD10-372DF235CF62}.Release|Any CPU.ActiveCfg = Release|Any CPU {9521D9E0-642A-4C7E-BD10-372DF235CF62}.Release|Any CPU.Build.0 = Release|Any CPU + {25E4C70A-196C-4855-8DFC-796C08837987}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {25E4C70A-196C-4855-8DFC-796C08837987}.Debug|Any CPU.Build.0 = Debug|Any CPU + {25E4C70A-196C-4855-8DFC-796C08837987}.Release|Any CPU.ActiveCfg = Release|Any CPU + {25E4C70A-196C-4855-8DFC-796C08837987}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE From 7a528a526bef94234c6057cf602d15d6a977254c Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 00:55:28 +0000 Subject: [PATCH 04/12] Fix test --- tests/NATS.Client.Core.Tests/NatsConnectionTest.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/NATS.Client.Core.Tests/NatsConnectionTest.cs b/tests/NATS.Client.Core.Tests/NatsConnectionTest.cs index 70f2be0be..d3de52f6a 100644 --- a/tests/NATS.Client.Core.Tests/NatsConnectionTest.cs +++ b/tests/NATS.Client.Core.Tests/NatsConnectionTest.cs @@ -458,7 +458,7 @@ public async Task OnSocketAvailableAsync_ShouldBeInvokedOnReconnection() await nats.ConnectAsync(); - await Retry.Until("callback", () => Interlocked.CompareExchange(ref invocationCount, 0, 0) == 2); + await Retry.Until("callback", () => Interlocked.CompareExchange(ref invocationCount, 0, 0) == 2, timeout: TimeSpan.FromSeconds(30)); } [Fact] From a7f7dc6ae18e008e790522e851c4167774993b40 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 01:12:54 +0000 Subject: [PATCH 05/12] Fix missing test project in solution file --- NATS.Net.sln | 1 + 1 file changed, 1 insertion(+) diff --git a/NATS.Net.sln b/NATS.Net.sln index 4d281f6a4..b0015e8ab 100644 --- a/NATS.Net.sln +++ b/NATS.Net.sln @@ -404,6 +404,7 @@ Global {9521D9E0-642A-4C7E-BD10-372DF235CF62} = {C526E8AB-739A-48D7-8FC4-048978C9B650} {B9EF0111-6720-46DF-B11A-8F8C88C3F5C1} = {899BE3EA-C5CA-4394-9B62-C45CBFF3AF4E} {0B7F1286-4426-45DE-82C2-FE7CF13CA0DA} = {B9EF0111-6720-46DF-B11A-8F8C88C3F5C1} + {25E4C70A-196C-4855-8DFC-796C08837987} = {C526E8AB-739A-48D7-8FC4-048978C9B650} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {8CBB7278-D093-448E-B3DE-B5991209A1AA} From 7c1102ccb8e760975759730d8badcb1ab1a94d25 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 03:41:01 +0000 Subject: [PATCH 06/12] Fix JetStream test flappers --- .../ConsumerConsumeTest.cs | 193 ++++++++++-------- .../CustomSerializerTest.cs | 27 ++- .../DoubleAckNakDelayTests.cs | 38 ++-- .../ManageConsumerTest.cs | 2 +- .../OrderedConsumerTest.cs | 4 +- .../PublishRetryTest.cs | 2 +- .../PublishTest.cs | 101 +++++---- tests/NATS.Client.JetStream.Tests/Utils.cs | 19 ++ tests/NATS.Client.TestUtilities/NatsProxy.cs | 8 +- tests/NATS.Client.TestUtilities/NatsServer.cs | 18 +- tests/NATS.Client.TestUtilities/Utils.cs | 4 +- 11 files changed, 250 insertions(+), 166 deletions(-) diff --git a/tests/NATS.Client.JetStream.Tests/ConsumerConsumeTest.cs b/tests/NATS.Client.JetStream.Tests/ConsumerConsumeTest.cs index d1101f656..533a358a2 100644 --- a/tests/NATS.Client.JetStream.Tests/ConsumerConsumeTest.cs +++ b/tests/NATS.Client.JetStream.Tests/ConsumerConsumeTest.cs @@ -3,6 +3,7 @@ using NATS.Client.Core.Tests; using NATS.Client.Core2.Tests; using NATS.Client.JetStream.Models; +using NATS.Client.TestUtilities; using NATS.Client.TestUtilities2; namespace NATS.Client.JetStream.Tests; @@ -61,23 +62,26 @@ await Assert.ThrowsAnyAsync(async () => [Fact] public async Task Consume_msgs_test() { - await using var server = await NatsServer.StartJSAsync(); - var (nats, proxy) = server.CreateProxiedClientConnection(); + var proxy = _server.CreateProxy(); + await using var nats = proxy.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); - await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.*" }, cts.Token); + await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); for (var i = 0; i < 30; i++) { - var ack = await js.PublishAsync("s1.foo", new TestData { Test = i }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", new TestData { Test = i }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); ack.EnsureSuccess(); } var consumerOpts = new NatsJSConsumeOpts { MaxMsgs = 10 }; - var consumer = (NatsJSConsumer)await js.GetConsumerAsync("s1", "c1", cts.Token); + var consumer = (NatsJSConsumer)await js.GetConsumerAsync($"{prefix}s1", $"{prefix}c1", cts.Token); var count = 0; await foreach (var msg in consumer.ConsumeAsync(serializer: TestDataJsonSerializer.Default, consumerOpts, cancellationToken: cts.Token)) { @@ -90,14 +94,13 @@ public async Task Consume_msgs_test() int? PullCount() => proxy? .ClientFrames - .Count(f => f.Message.StartsWith("PUB $JS.API.CONSUMER.MSG.NEXT.s1.c1")); + .Count(f => f.Message.StartsWith($"PUB $JS.API.CONSUMER.MSG.NEXT.{prefix}s1.{prefix}c1")); await Retry.Until( reason: "received enough pulls", condition: () => PullCount() > 4, action: () => { - _output.WriteLine($"### PullCount:{PullCount()}"); return Task.CompletedTask; }, retryDelay: TimeSpan.FromSeconds(3), @@ -105,7 +108,7 @@ await Retry.Until( var msgNextRequests = proxy .ClientFrames - .Where(f => f.Message.StartsWith("PUB $JS.API.CONSUMER.MSG.NEXT.s1.c1")) + .Where(f => f.Message.StartsWith($"PUB $JS.API.CONSUMER.MSG.NEXT.{prefix}s1.{prefix}c1")) .ToList(); foreach (var frame in msgNextRequests) @@ -120,38 +123,44 @@ await Retry.Until( [Fact] public async Task Consume_idle_heartbeat_test() { - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await using var server = await NatsServer.StartJSWithTraceAsync(_output); + var signal = new WaitSignal(TimeSpan.FromSeconds(60)); + var logger = new InMemoryTestLoggerFactory(LogLevel.Debug, log => + { + if (log is { Category: "NATS.Client.JetStream.Internal.NatsJSConsume", LogLevel: LogLevel.Debug }) + { + if (log.EventId == NatsJSLogEvents.IdleTimeout) + signal.Pulse(); + } + }); - var (nats, proxy) = server.CreateProxiedClientConnection(); + var proxy = _server.CreateProxy(); + await using var nats = new NatsConnection(new NatsOpts + { + Url = $"nats://127.0.0.1:{proxy.Port}", + ConnectTimeout = TimeSpan.FromSeconds(10), + LoggerFactory = logger, + }); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); // Swallow heartbeats proxy.ServerInterceptors.Add(m => m?.Contains("Idle Heartbeat") ?? false ? null : m); - await nats.ConnectRetryAsync(); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); var js = new NatsJSContext(nats); - await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); - await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.*" }, cts.Token); + await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); - var ack = await js.PublishAsync("s1.foo", new TestData { Test = 0 }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", new TestData { Test = 0 }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); ack.EnsureSuccess(); - var signal = new WaitSignal(TimeSpan.FromSeconds(30)); - server.OnLog += log => - { - if (log is { Category: "NATS.Client.JetStream.Internal.NatsJSConsume", LogLevel: LogLevel.Debug }) - { - if (log.EventId == NatsJSLogEvents.IdleTimeout) - signal.Pulse(); - } - }; var consumerOpts = new NatsJSConsumeOpts { MaxMsgs = 10, IdleHeartbeat = TimeSpan.FromSeconds(5), }; - var consumer = (NatsJSConsumer)await js.GetConsumerAsync("s1", "c1", cts.Token); + var consumer = (NatsJSConsumer)await js.GetConsumerAsync($"{prefix}s1", $"{prefix}c1", cts.Token); var count = 0; var cc = await consumer.ConsumeInternalAsync(serializer: TestDataJsonSerializer.Default, consumerOpts, cancellationToken: cts.Token); await foreach (var msg in cc.Msgs.ReadAllAsync(cts.Token)) @@ -164,28 +173,13 @@ public async Task Consume_idle_heartbeat_test() await Retry.Until( "all pull requests are received", - () => proxy.ClientFrames.Count(f => f.Message.StartsWith("PUB $JS.API.CONSUMER.MSG.NEXT.s1.c1")) >= 2); + () => proxy.ClientFrames.Count(f => f.Message.StartsWith($"PUB $JS.API.CONSUMER.MSG.NEXT.{prefix}s1.{prefix}c1")) >= 2); var msgNextRequests = proxy .ClientFrames - .Where(f => f.Message.StartsWith("PUB $JS.API.CONSUMER.MSG.NEXT.s1.c1")) + .Where(f => f.Message.StartsWith($"PUB $JS.API.CONSUMER.MSG.NEXT.{prefix}s1.{prefix}c1")) .ToList(); - // In some cases we are receiving more than two requests which - // is possible if the tests are running in a slow container and taking - // more than the timeout? Looking at the test and the code I can't make - // sense of it, really, but I'm going to assume it's fine to receive 3 pull - // requests as well as 2 since test failure reported 3 and failed once. - if (msgNextRequests.Count > 2) - { - _output.WriteLine($"Pull request count more than expected: {msgNextRequests.Count}"); - foreach (var frame in msgNextRequests) - { - _output.WriteLine($"PULL REQUEST: {frame}"); - } - } - - // Still fail and check traces if it happens again Assert.True(msgNextRequests.Count is 2); // Pull requests @@ -198,23 +192,28 @@ await Retry.Until( [Fact] public async Task Consume_reconnect_test() { - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await using var server = await NatsServer.StartJSAsync(); + var proxy = _server.CreateProxy(); + await using var nats = proxy.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + + await using var nats2 = _server.CreateNatsConnection(); + await nats2.ConnectRetryAsync(); + + var prefix = _server.GetNextId(); - var (nats, proxy) = server.CreateProxiedClientConnection(); - await using var nats2 = await server.CreateClientConnectionAsync(); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); var js = new NatsJSContext(nats); var js2 = new NatsJSContext(nats2); - await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); - await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.*" }, cts.Token); + await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); var consumerOpts = new NatsJSConsumeOpts { MaxMsgs = 10, }; - var consumer = (NatsJSConsumer)await js.GetConsumerAsync("s1", "c1", cts.Token); + var consumer = (NatsJSConsumer)await js.GetConsumerAsync($"{prefix}s1", $"{prefix}c1", cts.Token); // Not interested in management messages sent upto this point await proxy.FlushFramesAsync(nats); @@ -238,13 +237,13 @@ public async Task Consume_reconnect_test() // Send a message before reconnect { - var ack = await js2.PublishAsync("s1.foo", new TestData { Test = 0 }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); + var ack = await js2.PublishAsync($"{prefix}s1.foo", new TestData { Test = 0 }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); ack.EnsureSuccess(); } await Retry.Until( "acked", - () => proxy.ClientFrames.Any(f => f.Message.StartsWith("PUB $JS.ACK.s1.c1"))); + () => proxy.ClientFrames.Any(f => f.Message.StartsWith($"PUB $JS.ACK.{prefix}s1.{prefix}c1"))); Assert.Contains(proxy.ClientFrames, f => f.Message.Contains("CONSUMER.MSG.NEXT")); @@ -258,7 +257,7 @@ await Retry.Until( // Send a message to be received after reconnect { - var ack = await js2.PublishAsync("s1.foo", new TestData { Test = 1 }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); + var ack = await js2.PublishAsync($"{prefix}s1.foo", new TestData { Test = 1 }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); ack.EnsureSuccess(); } @@ -348,14 +347,15 @@ await Retry.Until( [Fact] public async Task Consume_stop_test() { - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); var js = new NatsJSContext(nats); - var stream = await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); - var consumer = (NatsJSConsumer)await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + var stream = await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.*" }, cts.Token); + var consumer = (NatsJSConsumer)await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); var consumerOpts = new NatsJSConsumeOpts { @@ -366,7 +366,7 @@ public async Task Consume_stop_test() for (var i = 0; i < 10; i++) { - var ack = await js.PublishAsync("s1.foo", new TestData { Test = i }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", new TestData { Test = i }, serializer: TestDataJsonSerializer.Default, cancellationToken: cts.Token); ack.EnsureSuccess(); } @@ -398,7 +398,7 @@ await Retry.Until( "ack pending 9", async () => { - var c = await js.GetConsumerAsync("s1", "c1", cts.Token); + var c = await js.GetConsumerAsync($"{prefix}s1", $"{prefix}c1", cts.Token); return c.Info.NumAckPending == 9; }, timeout: TimeSpan.FromSeconds(20)); @@ -413,7 +413,7 @@ await Retry.Until( "ack pending 0", async () => { - var c = await js.GetConsumerAsync("s1", "c1", cts.Token); + var c = await js.GetConsumerAsync($"{prefix}s1", $"{prefix}c1", cts.Token); return c.Info.NumAckPending == 0; }, timeout: TimeSpan.FromSeconds(20)); @@ -424,18 +424,21 @@ await Retry.Until( [Fact] public async Task Serialization_errors() { - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); + await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.*" }, cts.Token); - var ack = await js.PublishAsync("s1.foo", "not an int", cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", "not an int", cancellationToken: cts.Token); ack.EnsureSuccess(); - var consumer = await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); await foreach (var msg in consumer.ConsumeAsync(cancellationToken: cts.Token)) { @@ -452,24 +455,25 @@ public async Task Serialization_errors() [Fact] public async Task Consume_right_amount_of_messages() { - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); var js = new NatsJSContext(nats); - await js.CreateStreamAsync("s1", ["s1.*"], cts.Token); + await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); var payload = new byte[1024]; for (var i = 0; i < 50; i++) { - var ack = await js.PublishAsync("s1.foo", payload, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", payload, cancellationToken: cts.Token); ack.EnsureSuccess(); } // Max messages { - var consumer = await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); var opts = new NatsJSConsumeOpts { MaxMsgs = 10, }; var count = 0; await foreach (var msg in consumer.ConsumeAsync(opts: opts, cancellationToken: cts.Token)) @@ -479,16 +483,21 @@ public async Task Consume_right_amount_of_messages() break; } - await Retry.Until("consumer stats updated", async () => - { - var info = (await js.GetConsumerAsync("s1", "c1", cts.Token)).Info; - return info is { NumAckPending: 6, NumPending: 40 }; - }); + await Retry.Until( + "consumer stats updated for Max messages", + async () => + { + var info = (await js.GetConsumerAsync($"{prefix}s1", $"{prefix}c1", cts.Token)).Info; + _output.WriteLine($"Consumer stats updated for Max messages {info.NumAckPending} of {info.NumPending}"); + return info is { NumAckPending: 6, NumPending: 40 }; + }, + retryDelay: TimeSpan.FromSeconds(3), + timeout: TimeSpan.FromSeconds(60)); } // Max bytes { - var consumer = await js.CreateOrUpdateConsumerAsync("s1", "c2", cancellationToken: cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c2", cancellationToken: cts.Token); var opts = new NatsJSConsumeOpts { MaxBytes = 10 * (1024 + 50), }; var count = 0; await foreach (var msg in consumer.ConsumeAsync(opts: opts, cancellationToken: cts.Token)) @@ -498,35 +507,39 @@ await Retry.Until("consumer stats updated", async () => break; } - await Retry.Until("consumer stats updated", async () => - { - var info = (await js.GetConsumerAsync("s1", "c2", cts.Token)).Info; - return info is { NumAckPending: 6, NumPending: 40 }; - }); + await Retry.Until( + "consumer stats updated for Max bytes", + async () => + { + var info = (await js.GetConsumerAsync($"{prefix}s1", $"{prefix}c2", cts.Token)).Info; + _output.WriteLine($"Consumer stats updated for Max bytes {info.NumAckPending} of {info.NumPending}"); + return info is { NumAckPending: 5, NumPending: 41 }; + }, + retryDelay: TimeSpan.FromSeconds(3), + timeout: TimeSpan.FromSeconds(60)); } } [Fact] public async Task Consume_right_amount_of_messages_when_ack_wait_exceeded() { - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); var js = new NatsJSContext(nats); - await js.CreateStreamAsync("email-queue", ["email.>"], cts.Token); - await js.PublishAsync("email.queue", "1", cancellationToken: cts.Token); - await js.PublishAsync("email.queue", "2", cancellationToken: cts.Token); + await js.CreateStreamAsync($"{prefix}email-queue", [$"{prefix}email.>"], cts.Token); + await js.PublishAsync($"{prefix}email.queue", "1", cancellationToken: cts.Token); + await js.PublishAsync($"{prefix}email.queue", "2", cancellationToken: cts.Token); var consumer = await js.CreateOrUpdateConsumerAsync( - stream: "email-queue", - new ConsumerConfig("email-queue-consumer") { AckWait = TimeSpan.FromSeconds(10) }, + stream: $"{prefix}email-queue", + new ConsumerConfig($"{prefix}email-queue-consumer") { AckWait = TimeSpan.FromSeconds(10) }, cancellationToken: cts.Token); var count = 0; await foreach (var msg in consumer.ConsumeAsync(opts: new NatsJSConsumeOpts { MaxMsgs = 1 }, cancellationToken: cts.Token)) { - _output.WriteLine($"Received: {msg.Data}"); - // Only wait for the first couple of messages // to get close to the ack wait time if (count < 2) diff --git a/tests/NATS.Client.JetStream.Tests/CustomSerializerTest.cs b/tests/NATS.Client.JetStream.Tests/CustomSerializerTest.cs index e910b7096..a865e4867 100644 --- a/tests/NATS.Client.JetStream.Tests/CustomSerializerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/CustomSerializerTest.cs @@ -1,29 +1,44 @@ using System.Buffers; using NATS.Client.Core.Tests; +using NATS.Client.Core2.Tests; +using NATS.Client.TestUtilities2; namespace NATS.Client.JetStream.Tests; +[Collection("nats-server")] public class CustomSerializerTest { + private readonly ITestOutputHelper _output; + private readonly NatsServerFixture _server; + + public CustomSerializerTest(ITestOutputHelper output, NatsServerFixture server) + { + _output = output; + _server = server; + } + [Fact] public async Task When_consuming_ack_should_be_serialized_normally_if_custom_serializer_used() { - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(new NatsOpts + await using var nats = new NatsConnection(new NatsOpts { + Url = _server.Url, SerializerRegistry = new Level42SerializerRegistry(), RequestTimeout = TimeSpan.FromSeconds(10), }); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); + await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.*" }, cts.Token); - await js.PublishAsync("s1.1", new byte[] { 0 }, cancellationToken: cts.Token); - await js.PublishAsync("s1.2", new byte[] { 0 }, cancellationToken: cts.Token); + await js.PublishAsync($"{prefix}s1.1", new byte[] { 0 }, cancellationToken: cts.Token); + await js.PublishAsync($"{prefix}s1.2", new byte[] { 0 }, cancellationToken: cts.Token); - var consumer = await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); // single ack { diff --git a/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs b/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs index d7d388b82..c7d3cd3b9 100644 --- a/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs +++ b/tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs @@ -1,28 +1,38 @@ using System.Text.RegularExpressions; using NATS.Client.Core.Tests; +using NATS.Client.Core2.Tests; +using NATS.Client.TestUtilities2; namespace NATS.Client.JetStream.Tests; +[Collection("nats-server")] public class DoubleAckNakDelayTests { private readonly ITestOutputHelper _output; + private readonly NatsServerFixture _server; - public DoubleAckNakDelayTests(ITestOutputHelper output) => _output = output; + public DoubleAckNakDelayTests(ITestOutputHelper output, NatsServerFixture server) + { + _output = output; + _server = server; + } [Fact] public async Task Double_ack_received_messages() { - await using var server = await NatsServer.StartJSAsync(); - var (nats1, proxy) = server.CreateProxiedClientConnection(new NatsOpts { RequestTimeout = TimeSpan.FromSeconds(10) }); - await using var nats = nats1; + var proxy = _server.CreateProxy(); + await using var nats = proxy.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); - var consumer = await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); - var ack = await js.PublishAsync("s1.foo", 42, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", 42, cancellationToken: cts.Token); ack.EnsureSuccess(); var next = await consumer.NextAsync(cancellationToken: cts.Token); if (next is { } msg) @@ -46,17 +56,19 @@ public async Task Double_ack_received_messages() [Fact] public async Task Delay_nak_received_messages() { - await using var server = await NatsServer.StartJSAsync(); - var (nats1, proxy) = server.CreateProxiedClientConnection(new NatsOpts { RequestTimeout = TimeSpan.FromSeconds(10) }); - await using var nats = nats1; + var proxy = _server.CreateProxy(); + await using var nats = proxy.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); - var consumer = await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.*" }, cts.Token); + var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); - var ack = await js.PublishAsync("s1.foo", 42, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", 42, cancellationToken: cts.Token); ack.EnsureSuccess(); var next = await consumer.NextAsync(cancellationToken: cts.Token); if (next is { } msg) diff --git a/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs b/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs index 7038aae62..bc239aa23 100644 --- a/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs @@ -41,7 +41,7 @@ public async Task Create_get_consumer() [Fact] public async Task List_delete_consumer() { - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); await using var server = await NatsServer.StartJSAsync(); var nats = await server.CreateClientConnectionAsync(); diff --git a/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs b/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs index 79f7cbd7a..8b2bd8965 100644 --- a/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs @@ -143,7 +143,7 @@ public async Task Fetch_no_wait_test() _output.WriteLine($"stopwatch.Elapsed: {stopwatch.Elapsed}"); Assert.Equal(0, count); - Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(3)); + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(10)); // Where there is less than we want to fetch, we should get all the messages // without waiting for the timeout. @@ -178,7 +178,7 @@ public async Task Fetch_no_wait_test() Assert.Equal(2, iterationCount); Assert.Equal(10, count); - Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(3)); + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(10)); } [Fact] diff --git a/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs b/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs index f1a3a68ec..e76eb3611 100644 --- a/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs +++ b/tests/NATS.Client.JetStream.Tests/PublishRetryTest.cs @@ -107,7 +107,7 @@ public async Task Retry_with_telemetry_on_should_not_throw_header_readonly_excep }; ActivitySource.AddActivityListener(activityListener); - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url, diff --git a/tests/NATS.Client.JetStream.Tests/PublishTest.cs b/tests/NATS.Client.JetStream.Tests/PublishTest.cs index a067554b4..fae93e94a 100644 --- a/tests/NATS.Client.JetStream.Tests/PublishTest.cs +++ b/tests/NATS.Client.JetStream.Tests/PublishTest.cs @@ -1,32 +1,42 @@ using System.Text.RegularExpressions; using Microsoft.Extensions.Logging; using NATS.Client.Core.Tests; +using NATS.Client.Core2.Tests; +using NATS.Client.TestUtilities; using NATS.Client.TestUtilities2; namespace NATS.Client.JetStream.Tests; +[Collection("nats-server")] public class PublishTest { private readonly ITestOutputHelper _output; + private readonly NatsServerFixture _server; - public PublishTest(ITestOutputHelper output) => _output = output; + public PublishTest(ITestOutputHelper output, NatsServerFixture server) + { + _output = output; + _server = server; + } [Fact] public async Task Publish_test() { - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await js.CreateStreamAsync("s1", new[] { "s1.>" }, cts.Token); - await js.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + await js.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.>" }, cts.Token); + await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); // Publish { var ack = await js.PublishAsync( - "s1.foo", + $"{prefix}s1.foo", new TestData { Test = 1, @@ -35,7 +45,7 @@ public async Task Publish_test() cancellationToken: cts.Token); Assert.Null(ack.Error); Assert.Equal(1, (int)ack.Seq); - Assert.Equal("s1", ack.Stream); + Assert.Equal($"{prefix}s1", ack.Stream); Assert.False(ack.Duplicate); Assert.True(ack.IsSuccess()); } @@ -43,7 +53,7 @@ public async Task Publish_test() // Duplicate { var ack1 = await js.PublishAsync( - subject: "s1.foo", + subject: $"{prefix}s1.foo", data: new TestData { Test = 2 }, serializer: TestDataJsonSerializer.Default, opts: new NatsJSPubOpts { MsgId = "2" }, @@ -54,7 +64,7 @@ public async Task Publish_test() Assert.True(ack1.IsSuccess()); var ack2 = await js.PublishAsync( - subject: "s1.foo", + subject: $"{prefix}s1.foo", data: new TestData { Test = 2 }, serializer: TestDataJsonSerializer.Default, opts: new NatsJSPubOpts { MsgId = "2" }, @@ -67,14 +77,14 @@ public async Task Publish_test() // ExpectedStream { var ack1 = await js.PublishAsync( - subject: "s1.foo", + subject: $"{prefix}s1.foo", data: 1, - opts: new NatsJSPubOpts { ExpectedStream = "s1" }, + opts: new NatsJSPubOpts { ExpectedStream = $"{prefix}s1" }, cancellationToken: cts.Token); Assert.Null(ack1.Error); var ack2 = await js.PublishAsync( - subject: "s1.foo", + subject: $"{prefix}s1.foo", data: 2, opts: new NatsJSPubOpts { ExpectedStream = "non-existent-stream" }, cancellationToken: cts.Token); @@ -87,20 +97,20 @@ public async Task Publish_test() // ExpectedLastSequence { var ack1 = await js.PublishAsync( - subject: "s1.foo", + subject: $"{prefix}s1.foo", data: 1, cancellationToken: cts.Token); Assert.Null(ack1.Error); var ack2 = await js.PublishAsync( - subject: "s1.foo", + subject: $"{prefix}s1.foo", data: 2, opts: new NatsJSPubOpts { ExpectedLastSequence = ack1.Seq }, cancellationToken: cts.Token); Assert.Null(ack2.Error); var ack3 = await js.PublishAsync( - subject: "s1.foo", + subject: $"{prefix}s1.foo", data: 3, opts: new NatsJSPubOpts { ExpectedLastSequence = ack1.Seq }, cancellationToken: cts.Token); @@ -112,20 +122,20 @@ public async Task Publish_test() // ExpectedLastSubjectSequence { var ack1 = await js.PublishAsync( - subject: "s1.foo.ExpectedLastSubjectSequence", + subject: $"{prefix}s1.foo.ExpectedLastSubjectSequence", data: 1, cancellationToken: cts.Token); Assert.Null(ack1.Error); var ack2 = await js.PublishAsync( - subject: "s1.foo.ExpectedLastSubjectSequence", + subject: $"{prefix}s1.foo.ExpectedLastSubjectSequence", data: 2, opts: new NatsJSPubOpts { ExpectedLastSubjectSequence = ack1.Seq }, cancellationToken: cts.Token); Assert.Null(ack2.Error); var ack3 = await js.PublishAsync( - subject: "s1.foo.ExpectedLastSubjectSequence", + subject: $"{prefix}s1.foo.ExpectedLastSubjectSequence", data: 3, opts: new NatsJSPubOpts { ExpectedLastSubjectSequence = ack1.Seq }, cancellationToken: cts.Token); @@ -137,21 +147,21 @@ public async Task Publish_test() // ExpectedLastMsgId { var ack1 = await js.PublishAsync( - subject: "s1.foo.ExpectedLastSubjectSequence", + subject: $"{prefix}s1.foo.ExpectedLastSubjectSequence", data: 1, opts: new NatsJSPubOpts { MsgId = "ExpectedLastMsgId-1" }, cancellationToken: cts.Token); Assert.Null(ack1.Error); var ack2 = await js.PublishAsync( - subject: "s1.foo.ExpectedLastSubjectSequence", + subject: $"{prefix}s1.foo.ExpectedLastSubjectSequence", data: 2, opts: new NatsJSPubOpts { MsgId = "ExpectedLastMsgId-2", ExpectedLastMsgId = "ExpectedLastMsgId-1" }, cancellationToken: cts.Token); Assert.Null(ack2.Error); var ack3 = await js.PublishAsync( - subject: "s1.foo.ExpectedLastSubjectSequence", + subject: $"{prefix}s1.foo.ExpectedLastSubjectSequence", data: 3, opts: new NatsJSPubOpts { MsgId = "ExpectedLastMsgId-3", ExpectedLastMsgId = "unexpected-msg-id" }, cancellationToken: cts.Token); @@ -164,14 +174,26 @@ public async Task Publish_test() [Fact] public async Task Publish_retry_test() { - var ackRegex = new Regex(@"{""stream"":""s1"",\s*""seq"":\d+}"); + var retryCount = 0; + var logger = new InMemoryTestLoggerFactory(LogLevel.Debug, log => + { + if (log is { LogLevel: LogLevel.Debug } && log.EventId == NatsJSLogEvents.PublishNoResponseRetry) + { + Interlocked.Increment(ref retryCount); + } + }); - // give enough time for retries to avoid NatsJSPublishNoResponseExceptions - var natsOpts = NatsOpts.Default with { RequestTimeout = TimeSpan.FromSeconds(3) }; + var proxy = _server.CreateProxy(); + await using var nats = new NatsConnection(new NatsOpts + { + Url = $"nats://127.0.0.1:{proxy.Port}", + ConnectTimeout = TimeSpan.FromSeconds(10), + RequestTimeout = TimeSpan.FromSeconds(3), // give enough time for retries to avoid NatsJSPublishNoResponseExceptions + LoggerFactory = logger, + }); + var prefix = _server.GetNextId(); - await using var server = await NatsServer.StartJSAsync(); - var (nats1, proxy) = server.CreateProxiedClientConnection(natsOpts); - await using var nats = nats1; + var ackRegex = new Regex($$"""{"stream":"{{prefix}}s1",\s*"seq":\s*\d+}"""); var swallowAcksCount = 0; proxy.ServerInterceptors.Add(m => @@ -187,32 +209,23 @@ public async Task Publish_retry_test() return m; }); - var retryCount = 0; - server.OnLog += log => - { - if (log is { LogLevel: LogLevel.Debug } && log.EventId == NatsJSLogEvents.PublishNoResponseRetry) - { - Interlocked.Increment(ref retryCount); - } - }; - - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(45)); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60)); // use different connection to create stream and consumer to avoid request timeouts - await using var nats0 = await server.CreateClientConnectionAsync(); + await using var nats0 = _server.CreateNatsConnection(); await nats.ConnectRetryAsync(); await nats0.ConnectRetryAsync(); var js0 = new NatsJSContext(nats0); - await js0.CreateStreamAsync("s1", new[] { "s1.>" }, cts.Token); - await js0.CreateOrUpdateConsumerAsync("s1", "c1", cancellationToken: cts.Token); + await js0.CreateStreamAsync($"{prefix}s1", new[] { $"{prefix}s1.>" }, cts.Token); + await js0.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); var js = new NatsJSContext(nats); // Publish succeeds without retry { - var ack = await js.PublishAsync("s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token); ack.EnsureSuccess(); Assert.Equal(0, Volatile.Read(ref retryCount)); @@ -226,7 +239,7 @@ public async Task Publish_retry_test() Interlocked.Exchange(ref retryCount, 0); Interlocked.Exchange(ref swallowAcksCount, 1); - var ack = await js.PublishAsync("s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token); ack.EnsureSuccess(); Assert.Equal(1, Volatile.Read(ref retryCount)); @@ -239,7 +252,7 @@ public async Task Publish_retry_test() Interlocked.Exchange(ref retryCount, 0); Interlocked.Exchange(ref swallowAcksCount, 2); - var ack = await js.PublishAsync("s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 3 }, cancellationToken: cts.Token); + var ack = await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 3 }, cancellationToken: cts.Token); ack.EnsureSuccess(); Assert.Equal(2, Volatile.Read(ref retryCount)); @@ -253,7 +266,7 @@ public async Task Publish_retry_test() Interlocked.Exchange(ref swallowAcksCount, 2); await Assert.ThrowsAsync(async () => - await js.PublishAsync("s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token)); + await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token)); Assert.Equal(2, Volatile.Read(ref retryCount)); await Retry.Until("ack received", () => proxy.Frames.Count(f => ackRegex.IsMatch(f.Message)) == 2, timeout: TimeSpan.FromSeconds(20)); diff --git a/tests/NATS.Client.JetStream.Tests/Utils.cs b/tests/NATS.Client.JetStream.Tests/Utils.cs index 57f010fe7..d72e65a0f 100644 --- a/tests/NATS.Client.JetStream.Tests/Utils.cs +++ b/tests/NATS.Client.JetStream.Tests/Utils.cs @@ -1,3 +1,5 @@ +using NATS.Client.Core.Tests; +using NATS.Client.Core2.Tests; using NATS.Client.JetStream.Models; namespace NATS.Client.JetStream.Tests; @@ -9,4 +11,21 @@ public static ValueTask CreateOrUpdateConsumerAsync(this NatsJS public static ValueTask CreateStreamAsync(this NatsJSContext context, string stream, string[] subjects, CancellationToken cancellationToken = default) => context.CreateStreamAsync(new StreamConfig { Name = stream, Subjects = subjects }, cancellationToken); + + public static NatsProxy CreateProxy(this NatsServerFixture server) + => new(new Uri(server.Url).Port); + + public static NatsConnection CreateNatsConnection(this NatsProxy proxy) + => new(new NatsOpts + { + Url = $"nats://127.0.0.1:{proxy.Port}", + ConnectTimeout = TimeSpan.FromSeconds(10), + }); + + public static NatsConnection CreateNatsConnection(this NatsServerFixture server) + => new(new NatsOpts + { + Url = server.Url, + ConnectTimeout = TimeSpan.FromSeconds(10), + }); } diff --git a/tests/NATS.Client.TestUtilities/NatsProxy.cs b/tests/NATS.Client.TestUtilities/NatsProxy.cs index 62a21c85b..e82ebeb92 100644 --- a/tests/NATS.Client.TestUtilities/NatsProxy.cs +++ b/tests/NATS.Client.TestUtilities/NatsProxy.cs @@ -158,7 +158,13 @@ public async Task FlushFramesAsync(NatsConnection nats) await Retry.Until( "flush sync frame", - () => AllFrames.Any(f => f.Message == $"PUB {subject} 0␍␊")); + async () => + { + await nats.PublishAsync(subject); + return AllFrames.Any(f => f.Message == $"PUB {subject} 0␍␊"); + }, + retryDelay: TimeSpan.FromSeconds(1), + timeout: TimeSpan.FromSeconds(60)); lock (_frames) _frames.Clear(); diff --git a/tests/NATS.Client.TestUtilities/NatsServer.cs b/tests/NATS.Client.TestUtilities/NatsServer.cs index 86cbf4caf..ba8cc305c 100644 --- a/tests/NATS.Client.TestUtilities/NatsServer.cs +++ b/tests/NATS.Client.TestUtilities/NatsServer.cs @@ -308,9 +308,12 @@ public async ValueTask StopAsync() { _cancellationTokenSource?.Cancel(); // trigger of process kill. _cancellationTokenSource?.Dispose(); - ServerProcess!.Kill(); - using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await ServerProcess!.WaitForExitAsync(cts.Token); + if (ServerProcess != null) + { + ServerProcess.Kill(); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await ServerProcess.WaitForExitAsync(cts.Token); + } } catch (OperationCanceledException) { @@ -333,9 +336,12 @@ public async ValueTask DisposeAsync() { _cancellationTokenSource?.Cancel(); // trigger of process kill. _cancellationTokenSource?.Dispose(); - ServerProcess!.Kill(); - using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await ServerProcess!.WaitForExitAsync(cts.Token); + if (ServerProcess != null) + { + ServerProcess.Kill(); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await ServerProcess.WaitForExitAsync(cts.Token)!; + } } catch (OperationCanceledException) { diff --git a/tests/NATS.Client.TestUtilities/Utils.cs b/tests/NATS.Client.TestUtilities/Utils.cs index f55f9e901..835f362b0 100644 --- a/tests/NATS.Client.TestUtilities/Utils.cs +++ b/tests/NATS.Client.TestUtilities/Utils.cs @@ -15,7 +15,7 @@ public static class Retry { public static async Task Until(string reason, Func condition, Func? action = null, TimeSpan? timeout = null, TimeSpan? retryDelay = null) { - timeout ??= TimeSpan.FromSeconds(10); + timeout ??= TimeSpan.FromSeconds(30); var delay1 = retryDelay ?? TimeSpan.FromSeconds(.1); var stopwatch = Stopwatch.StartNew(); @@ -33,7 +33,7 @@ public static async Task Until(string reason, Func condition, Func? public static async Task Until(string reason, Func> condition, Func? action = null, TimeSpan? timeout = null, TimeSpan? retryDelay = null) { - timeout ??= TimeSpan.FromSeconds(10); + timeout ??= TimeSpan.FromSeconds(30); var delay1 = retryDelay ?? TimeSpan.FromSeconds(.1); var stopwatch = Stopwatch.StartNew(); From 727092f68929df3b10f699bbfa9ca11c39724954 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 04:24:00 +0000 Subject: [PATCH 07/12] Fix test proxy signal subject --- tests/NATS.Client.TestUtilities/NatsProxy.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/NATS.Client.TestUtilities/NatsProxy.cs b/tests/NATS.Client.TestUtilities/NatsProxy.cs index e82ebeb92..62909b511 100644 --- a/tests/NATS.Client.TestUtilities/NatsProxy.cs +++ b/tests/NATS.Client.TestUtilities/NatsProxy.cs @@ -121,7 +121,7 @@ public IReadOnlyList Frames } } - public IReadOnlyList ClientFrames => Frames.Where(f => f.Origin == "C").ToList(); + public IReadOnlyList ClientFrames => Frames.Where(f => f.Origin == "C" && !f.Message.Contains("__PROXY_SIGNAL_SYNC__")).ToList(); public IReadOnlyList ServerFrames => Frames.Where(f => f.Origin == "S").ToList(); @@ -152,7 +152,7 @@ public void Reset() public async Task FlushFramesAsync(NatsConnection nats) { - var subject = $"_SIGNAL_SYNC_{Interlocked.Increment(ref _syncCount)}"; + var subject = $"__PROXY_SIGNAL_SYNC__{Interlocked.Increment(ref _syncCount)}"; await nats.PublishAsync(subject); From cd60b7b4aa9f986752079c69c3fd337fa1a18f5a Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 04:34:19 +0000 Subject: [PATCH 08/12] Fix test connection retry --- tests/NATS.Client.Core.Tests/ClusterTests.cs | 3 +++ tests/NATS.Client.Core.Tests/TlsClientTest.cs | 2 ++ 2 files changed, 5 insertions(+) diff --git a/tests/NATS.Client.Core.Tests/ClusterTests.cs b/tests/NATS.Client.Core.Tests/ClusterTests.cs index 33cb324ec..c6acd099b 100644 --- a/tests/NATS.Client.Core.Tests/ClusterTests.cs +++ b/tests/NATS.Client.Core.Tests/ClusterTests.cs @@ -1,3 +1,5 @@ +using NATS.Client.TestUtilities2; + namespace NATS.Client.Core.Tests; public class ClusterTests(ITestOutputHelper output) @@ -47,6 +49,7 @@ public async Task Seed_urls_on_retry(bool userAuthInUrl) NoRandomize = true, Url = $"{url1},{url2}", }); + await nats.ConnectRetryAsync(); using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60)); diff --git a/tests/NATS.Client.Core.Tests/TlsClientTest.cs b/tests/NATS.Client.Core.Tests/TlsClientTest.cs index 830ef7091..7eb8bba90 100644 --- a/tests/NATS.Client.Core.Tests/TlsClientTest.cs +++ b/tests/NATS.Client.Core.Tests/TlsClientTest.cs @@ -2,6 +2,7 @@ using System.Net.Sockets; using System.Security.Cryptography.X509Certificates; using System.Text; +using NATS.Client.TestUtilities2; namespace NATS.Client.Core.Tests; @@ -78,6 +79,7 @@ public async Task Client_cannot_connect_without_certificate(TransportType transp var clientOpts = server.ClientOpts(NatsOpts.Default); clientOpts = clientOpts with { TlsOpts = clientOpts.TlsOpts with { CertFile = null, KeyFile = null } }; await using var nats = new NatsConnection(clientOpts); + await nats.ConnectRetryAsync(); var exceptionTask = Assert.ThrowsAsync(async () => await nats.ConnectAsync()); From eee43aae9123f8585901fe638d2e5f3f9c85d14a Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 04:47:27 +0000 Subject: [PATCH 09/12] Fix JS Test --- tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs b/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs index 54daee928..4894f499d 100644 --- a/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs @@ -245,7 +245,7 @@ public async Task Ordered_consumer_consume_handling() var js = new NatsJSContext(nats); - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60)); var stream = await js.CreateStreamAsync(new StreamConfig($"{prefix}s1", new[] { $"{prefix}s1.*" }), cts.Token); var consumer = (NatsJSOrderedConsumer)await stream.CreateOrderedConsumerAsync(cancellationToken: cts.Token); @@ -281,7 +281,7 @@ public async Task Ordered_consumer_consume_handling() // Swallow heartbeats proxy.ServerInterceptors.Add(m => m?.Contains("Idle Heartbeat") ?? false ? null : m); - var consumeCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); + var consumeCts = new CancellationTokenSource(); var consume = Task.Run( async () => { @@ -308,7 +308,7 @@ public async Task Ordered_consumer_consume_handling() }, cts.Token); - await Retry.Until("timed out", () => Volatile.Read(ref timeoutNotifications) > 0, timeout: TimeSpan.FromSeconds(20)); + await Retry.Until("timed out", () => Volatile.Read(ref timeoutNotifications) > 0, timeout: TimeSpan.FromSeconds(30)); consumeCts.Cancel(); await consume; From 60d9f9e9ca5d206af7939c0a3ce4292387c9e55a Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 04:50:45 +0000 Subject: [PATCH 10/12] Fix test revert --- tests/NATS.Client.Core.Tests/TlsClientTest.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/NATS.Client.Core.Tests/TlsClientTest.cs b/tests/NATS.Client.Core.Tests/TlsClientTest.cs index 7eb8bba90..5655df56b 100644 --- a/tests/NATS.Client.Core.Tests/TlsClientTest.cs +++ b/tests/NATS.Client.Core.Tests/TlsClientTest.cs @@ -79,7 +79,6 @@ public async Task Client_cannot_connect_without_certificate(TransportType transp var clientOpts = server.ClientOpts(NatsOpts.Default); clientOpts = clientOpts with { TlsOpts = clientOpts.TlsOpts with { CertFile = null, KeyFile = null } }; await using var nats = new NatsConnection(clientOpts); - await nats.ConnectRetryAsync(); var exceptionTask = Assert.ThrowsAsync(async () => await nats.ConnectAsync()); From 96db87c65d68b8600a425e55c0f484bc540cff14 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 04:34:19 +0000 Subject: [PATCH 11/12] Fix test --- .../ClusterTests.cs | 3 +- .../OrderedConsumerTest.cs | 73 +++++++++++-------- 2 files changed, 44 insertions(+), 32 deletions(-) diff --git a/tests/NATS.Client.JetStream.Tests/ClusterTests.cs b/tests/NATS.Client.JetStream.Tests/ClusterTests.cs index 47d3cf301..bd7738465 100644 --- a/tests/NATS.Client.JetStream.Tests/ClusterTests.cs +++ b/tests/NATS.Client.JetStream.Tests/ClusterTests.cs @@ -1,4 +1,5 @@ using NATS.Client.Core.Tests; +using NATS.Client.TestUtilities2; namespace NATS.Client.JetStream.Tests; @@ -15,7 +16,7 @@ public async Task Form_a_local_cluster() await cluster.StartAsync(); await using var nats = await cluster.Server1.CreateClientConnectionAsync(); - await nats.PingAsync(); + await nats.ConnectRetryAsync(); var urls = nats.ServerInfo!.ClientConnectUrls!.ToList(); diff --git a/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs b/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs index 8b2bd8965..29ef2b249 100644 --- a/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/OrderedConsumerTest.cs @@ -1,33 +1,43 @@ using System.Diagnostics; using NATS.Client.Core.Tests; +using NATS.Client.Core2.Tests; +using NATS.Client.TestUtilities2; namespace NATS.Client.JetStream.Tests; +[Collection("nats-server")] public class OrderedConsumerTest { private readonly ITestOutputHelper _output; + private readonly NatsServerFixture _server; - public OrderedConsumerTest(ITestOutputHelper output) => _output = output; + public OrderedConsumerTest(ITestOutputHelper output, NatsServerFixture server) + { + _output = output; + _server = server; + } [Fact] public async Task Consume_test() { - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); - var stream = await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + var stream = await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); for (var i = 0; i < 10; i++) { - await js.PublishAsync("s1.foo", i, cancellationToken: cts.Token); + await js.PublishAsync($"{prefix}s1.foo", i, cancellationToken: cts.Token); } var consumer = await stream.CreateOrderedConsumerAsync(cancellationToken: cts.Token); var count = 0; - _output.WriteLine("Consuming..."); var consumeOpts = new NatsJSConsumeOpts { MaxMsgs = 3, @@ -35,7 +45,6 @@ public async Task Consume_test() }; await foreach (var msg in consumer.ConsumeAsync(opts: consumeOpts, cancellationToken: cts.Token)) { - _output.WriteLine($"[RCV] {msg.Data}"); Assert.Equal(count, msg.Data); if (++count == 10) break; @@ -51,7 +60,7 @@ public async Task Consume_reconnect_publish() var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - var stream = await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); + var stream = await js.CreateStreamAsync("s1", ["s1.*"], cts.Token); for (var i = 0; i < 50; i++) { @@ -87,23 +96,25 @@ public async Task Consume_reconnect_publish() [Fact] public async Task Fetch_test() { - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); - var stream = await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + var stream = await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); for (var i = 0; i < 10; i++) { - await js.PublishAsync("s1.foo", i, cancellationToken: cts.Token); + await js.PublishAsync($"{prefix}s1.foo", i, cancellationToken: cts.Token); } var consumer = await stream.CreateOrderedConsumerAsync(cancellationToken: cts.Token); for (var i = 0; i < 10;) { - _output.WriteLine("Fetching..."); var fetchOpts = new NatsJSFetchOpts { MaxMsgs = 3, @@ -111,7 +122,6 @@ public async Task Fetch_test() }; await foreach (var msg in consumer.FetchAsync(opts: fetchOpts, cancellationToken: cts.Token)) { - _output.WriteLine($"[RCV] {msg.Data}"); Assert.Equal(i, msg.Data); i++; } @@ -121,12 +131,15 @@ public async Task Fetch_test() [Fact] public async Task Fetch_no_wait_test() { - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); - var stream = await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + var stream = await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); var consumer = await stream.CreateOrderedConsumerAsync(cancellationToken: cts.Token); @@ -140,8 +153,6 @@ public async Task Fetch_no_wait_test() stopwatch.Stop(); - _output.WriteLine($"stopwatch.Elapsed: {stopwatch.Elapsed}"); - Assert.Equal(0, count); Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(10)); @@ -149,7 +160,7 @@ public async Task Fetch_no_wait_test() // without waiting for the timeout. for (var i = 0; i < 10; i++) { - await js.PublishAsync("s1.foo", i, cancellationToken: cts.Token); + await js.PublishAsync($"{prefix}s1.foo", i, cancellationToken: cts.Token); } stopwatch.Restart(); @@ -159,7 +170,6 @@ public async Task Fetch_no_wait_test() var currentCount = 0; await foreach (var msg in consumer.FetchNoWaitAsync(opts: new NatsJSFetchOpts { MaxMsgs = 6 }, cancellationToken: cts.Token)) { - _output.WriteLine($"[RCV][{iterationCount}] {msg.Data}"); Assert.Equal(count, msg.Data); count++; currentCount++; @@ -184,23 +194,25 @@ public async Task Fetch_no_wait_test() [Fact] public async Task Next_test() { - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); - await using var server = await NatsServer.StartJSAsync(); - await using var nats = await server.CreateClientConnectionAsync(); + await using var nats = _server.CreateNatsConnection(); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); - var stream = await js.CreateStreamAsync("s1", new[] { "s1.*" }, cts.Token); + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + var stream = await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token); for (var i = 0; i < 10; i++) { - await js.PublishAsync("s1.foo", i, cancellationToken: cts.Token); + await js.PublishAsync($"{prefix}s1.foo", i, cancellationToken: cts.Token); } var consumer = await stream.CreateOrderedConsumerAsync(cancellationToken: cts.Token); for (var i = 0; i < 10;) { - _output.WriteLine("Next..."); var nextOpts = new NatsJSNextOpts { Expires = TimeSpan.FromSeconds(3), @@ -209,7 +221,6 @@ public async Task Next_test() if (next is { } msg) { - _output.WriteLine($"[RCV] {msg.Data}"); Assert.Equal(i, msg.Data); i++; } From 7047d84e014966fec21a2636631cb9d3c470129c Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Sat, 25 Jan 2025 06:30:20 +0000 Subject: [PATCH 12/12] Fix test cancellation --- tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs b/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs index 4894f499d..abf07a358 100644 --- a/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/ErrorHandlerTest.cs @@ -310,9 +310,16 @@ public async Task Ordered_consumer_consume_handling() await Retry.Until("timed out", () => Volatile.Read(ref timeoutNotifications) > 0, timeout: TimeSpan.FromSeconds(30)); consumeCts.Cancel(); - await consume; Assert.True(Volatile.Read(ref timeoutNotifications) > 0); + + try + { + await consume; + } + catch (OperationCanceledException) + { + } } [Fact]