From dc061c14348d3f7205a5c42ba1d422a833162627 Mon Sep 17 00:00:00 2001 From: monty Date: Fri, 8 May 2026 17:05:32 +0100 Subject: [PATCH 1/6] Add Example Docs (#1119) * examples: add Example.NatsIODocs project * ci: add Example.NatsIODocs build workflow * examples: address review feedback - replace placeholder GUID with proper random one - add Ctrl+C cancellation to subscribe examples so they exit cleanly - restrict workflow GITHUB_TOKEN to contents:read * ci: temporarily skip non-essential workflows on this branch Will be reverted before merge. * examples: remove cancellation plumbing; add timeout to harness Subscribe examples no longer have CTS/CancelKeyPress plumbing, keeping doc snippets focused on the API. The Program.cs harness uses Task.WhenAny with a 5s timeout so subscribe examples can run unattended in CI without hanging. * examples: add 13 more NatsIODocs examples translated from java Covers subjects (single/multi wildcards, monitoring), queue groups (basic, dynamic scaling, request/reply, mixed subscribers), and request/reply variants (basic, timeout, multiple responders, no responders, headers, calculator). Each uses idiomatic .NET patterns: NatsClient, await foreach over SubscribeAsync, and Task.Run for background subscribers. Harness timeout bumped to 10s and CI runs all 17 examples. * examples: split getting-started into standalone console projects The getting-started examples are what new users will copy into a fresh 'dotnet new console' project, so they should be standalone rather than called via a static-class dispatcher. Each is now its own .csproj with its own Program.cs as the actual entry point. CI builds and runs both, using timeout(1) (or gtimeout on macOS) to bound the subscribe example. * examples: tighten NATS-DOC marker placement - Move markers off class scope into RunAsync body for QueueGroupsBasic, QueueGroupsDynamicScaling, SubjectsMultiWildcard so the doc snippet doesn't drag in the static-class harness scaffolding. - Stop running standalone getting-started projects in CI; build is enough since they target demo.nats.io which isn't reliably reachable from the macOS runner. * examples: use NatsSubOpts.Timeout in RequestReplyTimeout, tighten SubjectsMonitoring snippet RequestReplyTimeout now sets the per-request timeout via NatsSubOpts and catches NatsNoReplyException, which is the idiomatic NATS .NET pattern. SubjectsMonitoring's NATS-DOC markers now wrap just the background subscribe pattern (matching the Java original), with a count-based break replacing the cts.Token + try/catch noise. * Revert "ci: temporarily skip non-essential workflows on this branch" This reverts commit 21b97cd7deeb7a9ade2723b47d1b055618e0cb15. * examples: address review feedback * examples: convert NatsIODocs to xunit test project * examples: simplify NatsIODocs samples to doc-snippet style * examples: refine NatsIODocs sample tweaks --- .github/workflows/test_natsiodocs.yml | 46 +++++++++++++++ NATS.Net.slnx | 1 + examples/Example.NatsIODocs/BasicsPublish.cs | 23 ++++++++ .../Example.NatsIODocs/BasicsSubscribe.cs | 27 +++++++++ .../Example.NatsIODocs.csproj | 29 ++++++++++ .../GettingStartedPublish.cs | 18 ++++++ .../GettingStartedSubscribe.cs | 21 +++++++ .../Example.NatsIODocs/NatsServerFixture.cs | 20 +++++++ .../Example.NatsIODocs/QueueGroupsBasic.cs | 51 ++++++++++++++++ .../QueueGroupsDynamicScaling.cs | 38 ++++++++++++ .../QueueGroupsMixedSubscribers.cs | 58 +++++++++++++++++++ .../QueueGroupsRequestReply.cs | 41 +++++++++++++ .../Example.NatsIODocs/RequestReplyBasic.cs | 33 +++++++++++ .../RequestReplyCalculator.cs | 37 ++++++++++++ .../Example.NatsIODocs/RequestReplyHeaders.cs | 44 ++++++++++++++ .../RequestReplyMultipleResponders.cs | 39 +++++++++++++ .../RequestReplyNoResponders.cs | 28 +++++++++ .../Example.NatsIODocs/RequestReplyTimeout.cs | 42 ++++++++++++++ .../Example.NatsIODocs/SubjectsMonitoring.cs | 33 +++++++++++ .../SubjectsMultiWildcard.cs | 53 +++++++++++++++++ .../SubjectsSingleWildcard.cs | 52 +++++++++++++++++ 21 files changed, 734 insertions(+) create mode 100644 .github/workflows/test_natsiodocs.yml create mode 100644 examples/Example.NatsIODocs/BasicsPublish.cs create mode 100644 examples/Example.NatsIODocs/BasicsSubscribe.cs create mode 100644 examples/Example.NatsIODocs/Example.NatsIODocs.csproj create mode 100644 examples/Example.NatsIODocs/GettingStartedPublish.cs create mode 100644 examples/Example.NatsIODocs/GettingStartedSubscribe.cs create mode 100644 examples/Example.NatsIODocs/NatsServerFixture.cs create mode 100644 examples/Example.NatsIODocs/QueueGroupsBasic.cs create mode 100644 examples/Example.NatsIODocs/QueueGroupsDynamicScaling.cs create mode 100644 examples/Example.NatsIODocs/QueueGroupsMixedSubscribers.cs create mode 100644 examples/Example.NatsIODocs/QueueGroupsRequestReply.cs create mode 100644 examples/Example.NatsIODocs/RequestReplyBasic.cs create mode 100644 examples/Example.NatsIODocs/RequestReplyCalculator.cs create mode 100644 examples/Example.NatsIODocs/RequestReplyHeaders.cs create mode 100644 examples/Example.NatsIODocs/RequestReplyMultipleResponders.cs create mode 100644 examples/Example.NatsIODocs/RequestReplyNoResponders.cs create mode 100644 examples/Example.NatsIODocs/RequestReplyTimeout.cs create mode 100644 examples/Example.NatsIODocs/SubjectsMonitoring.cs create mode 100644 examples/Example.NatsIODocs/SubjectsMultiWildcard.cs create mode 100644 examples/Example.NatsIODocs/SubjectsSingleWildcard.cs diff --git a/.github/workflows/test_natsiodocs.yml b/.github/workflows/test_natsiodocs.yml new file mode 100644 index 000000000..ce9a33590 --- /dev/null +++ b/.github/workflows/test_natsiodocs.yml @@ -0,0 +1,46 @@ +name: Test Examples NatsIODocs + +on: + pull_request: {} + push: + branches: + - main + +permissions: + contents: read + +jobs: + build: + name: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest, macos-latest] + runs-on: ${{ matrix.os }} + env: + DOTNET_CLI_TELEMETRY_OPTOUT: 1 + DOTNET_SKIP_FIRST_TIME_EXPERIENCE: 1 + NUGET_XMLDOC_MODE: skip + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup dotnet + uses: actions/setup-dotnet@v4 + with: + dotnet-version: 8.x + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: 'stable' + cache: true + + - name: Install nats-server + run: go install github.com/nats-io/nats-server/v2@latest + + - name: Check nats-server + run: nats-server -v + + - name: Test + run: dotnet test examples/Example.NatsIODocs/Example.NatsIODocs.csproj -c Release diff --git a/NATS.Net.slnx b/NATS.Net.slnx index fe0bd79fc..53b3565f4 100644 --- a/NATS.Net.slnx +++ b/NATS.Net.slnx @@ -10,6 +10,7 @@ + diff --git a/examples/Example.NatsIODocs/BasicsPublish.cs b/examples/Example.NatsIODocs/BasicsPublish.cs new file mode 100644 index 000000000..f84fbe414 --- /dev/null +++ b/examples/Example.NatsIODocs/BasicsPublish.cs @@ -0,0 +1,23 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class BasicsPublish(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + var sub = await client.Connection.SubscribeCoreAsync("weather.updates"); + + // NATS-DOC-START + // Publish a message to the subject "weather.updates" + await client.PublishAsync("weather.updates", "Temperature: 72F"); + + // NATS-DOC-END + var msg = await sub.Msgs.ReadAsync(); + output.WriteLine($"Received: {msg.Data}"); + } +} diff --git a/examples/Example.NatsIODocs/BasicsSubscribe.cs b/examples/Example.NatsIODocs/BasicsSubscribe.cs new file mode 100644 index 000000000..3dfc502bd --- /dev/null +++ b/examples/Example.NatsIODocs/BasicsSubscribe.cs @@ -0,0 +1,27 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class BasicsSubscribe(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + _ = Task.Run(async () => + { + // NATS-DOC-START + // Subscribe to 'weather.updates' and process messages + await foreach (var msg in client.SubscribeAsync("weather.updates")) + { + output.WriteLine($"Received: {msg.Data}"); + } + + // NATS-DOC-END + }); + + await Task.Delay(1000); + await client.PublishAsync("weather.updates", "sunny"); + } +} diff --git a/examples/Example.NatsIODocs/Example.NatsIODocs.csproj b/examples/Example.NatsIODocs/Example.NatsIODocs.csproj new file mode 100644 index 000000000..05d5613a4 --- /dev/null +++ b/examples/Example.NatsIODocs/Example.NatsIODocs.csproj @@ -0,0 +1,29 @@ + + + + net8.0 + enable + enable + false + true + + + + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + + + + + + + diff --git a/examples/Example.NatsIODocs/GettingStartedPublish.cs b/examples/Example.NatsIODocs/GettingStartedPublish.cs new file mode 100644 index 000000000..8be1bf198 --- /dev/null +++ b/examples/Example.NatsIODocs/GettingStartedPublish.cs @@ -0,0 +1,18 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +public class GettingStartedPublish +{ + [Fact(Skip = "Targets demo.nats.io; not run in CI.")] + public async Task RunAsync() + { + // NATS-DOC-START + await using var client = new NatsClient("demo.nats.io"); + + // Publish a message to the subject "hello" + await client.PublishAsync("hello", "Hello NATS!"); + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/GettingStartedSubscribe.cs b/examples/Example.NatsIODocs/GettingStartedSubscribe.cs new file mode 100644 index 000000000..f8ad625b5 --- /dev/null +++ b/examples/Example.NatsIODocs/GettingStartedSubscribe.cs @@ -0,0 +1,21 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +public class GettingStartedSubscribe(ITestOutputHelper output) +{ + [Fact(Skip = "Targets demo.nats.io and blocks waiting for messages; not run in CI.")] + public async Task RunAsync() + { + // NATS-DOC-START + await using var client = new NatsClient("demo.nats.io"); + + // Subscribe to 'hello' and process messages + await foreach (var msg in client.SubscribeAsync("hello")) + { + output.WriteLine($"Received: {msg.Data}"); + } + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/NatsServerFixture.cs b/examples/Example.NatsIODocs/NatsServerFixture.cs new file mode 100644 index 000000000..d841c1b72 --- /dev/null +++ b/examples/Example.NatsIODocs/NatsServerFixture.cs @@ -0,0 +1,20 @@ +using Synadia.Orbit.Testing.NatsServerProcessManager; + +namespace Example.NatsIODocs; + +public sealed class NatsServerFixture : IDisposable +{ + public NatsServerFixture() + { + Server = NatsServerProcess.Start(); + } + + public NatsServerProcess Server { get; } + + public void Dispose() => Server.Dispose(); +} + +[CollectionDefinition("nats-server")] +public class NatsServerCollection : ICollectionFixture +{ +} diff --git a/examples/Example.NatsIODocs/QueueGroupsBasic.cs b/examples/Example.NatsIODocs/QueueGroupsBasic.cs new file mode 100644 index 000000000..01e48d8c1 --- /dev/null +++ b/examples/Example.NatsIODocs/QueueGroupsBasic.cs @@ -0,0 +1,51 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class QueueGroupsBasic(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Create three workers in the same queue group + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.new", queueGroup: "new-orders-queue")) + { + output.WriteLine($"Worker A processed: {msg.Data}"); + } + }); + + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.new", queueGroup: "new-orders-queue")) + { + output.WriteLine($"Worker B processed: {msg.Data}"); + } + }); + + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.new", queueGroup: "new-orders-queue")) + { + output.WriteLine($"Worker C processed: {msg.Data}"); + } + }); + + // Let subscription tasks start + await Task.Delay(1000); + + // Publish messages once all subscriptions are set up + for (var i = 1; i <= 10; i++) + { + await client.PublishAsync("orders.new", $"Order Number: {i}"); + } + + // NATS-DOC-END + await Task.Delay(1000); + } +} diff --git a/examples/Example.NatsIODocs/QueueGroupsDynamicScaling.cs b/examples/Example.NatsIODocs/QueueGroupsDynamicScaling.cs new file mode 100644 index 000000000..2d50be52e --- /dev/null +++ b/examples/Example.NatsIODocs/QueueGroupsDynamicScaling.cs @@ -0,0 +1,38 @@ +using NATS.Client.Core; +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class QueueGroupsDynamicScaling(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact(Skip = "Subscriber lifecycle (unsubscribe) plumbing isn't a clean doc snippet; not run in CI.")] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Start workers in the same queue group; track them so we can scale down later + var workers = new List>(); + + for (var i = 1; i <= 5; i++) + { + var id = i; + var sub = await client.Connection.SubscribeCoreAsync("tasks", queueGroup: "workers"); + _ = Task.Run(async () => + { + await foreach (var msg in sub.Msgs.ReadAllAsync()) + { + output.WriteLine($"Worker {id} processing: {msg.Data}"); + } + }); + workers.Add(sub); + } + + // Scale down: drop the first worker + await workers[0].DisposeAsync(); + workers.RemoveAt(0); + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/QueueGroupsMixedSubscribers.cs b/examples/Example.NatsIODocs/QueueGroupsMixedSubscribers.cs new file mode 100644 index 000000000..1a774fd62 --- /dev/null +++ b/examples/Example.NatsIODocs/QueueGroupsMixedSubscribers.cs @@ -0,0 +1,58 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class QueueGroupsMixedSubscribers(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Audit logger - receives all order messages + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.>")) + { + output.WriteLine($"[AUDIT] {msg.Subject}: {msg.Data}"); + } + }); + + // Metrics collector - receives all order messages + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.>")) + { + output.WriteLine($"[METRICS] {msg.Subject}: {msg.Data}"); + } + }); + + // Workers - share load via a queue group + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.new", queueGroup: "workers")) + { + output.WriteLine($"[WORKER A] Processing: {msg.Data}"); + } + }); + + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.new", queueGroup: "workers")) + { + output.WriteLine($"[WORKER B] Processing: {msg.Data}"); + } + }); + + // Let subscription tasks start + await Task.Delay(1000); + + // Audit and metrics see every message; one worker processes each + await client.PublishAsync("orders.new", "Order 123"); + + // NATS-DOC-END + await Task.Delay(1000); + } +} diff --git a/examples/Example.NatsIODocs/QueueGroupsRequestReply.cs b/examples/Example.NatsIODocs/QueueGroupsRequestReply.cs new file mode 100644 index 000000000..cc7347487 --- /dev/null +++ b/examples/Example.NatsIODocs/QueueGroupsRequestReply.cs @@ -0,0 +1,41 @@ +using NATS.Client.Core; +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class QueueGroupsRequestReply(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Start three service instances sharing the same queue group + for (var i = 1; i <= 3; i++) + { + var id = i; + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("api.calculate", queueGroup: "api-workers")) + { + await msg.ReplyAsync($"Result from instance {id}"); + } + }); + } + + // Let subscription tasks start + await Task.Delay(1000); + + // Make 10 requests; the queue group balances them across instances + var replyOpts = new NatsSubOpts { Timeout = TimeSpan.FromSeconds(1) }; + for (var i = 0; i < 10; i++) + { + var reply = await client.RequestAsync("api.calculate", replyOpts: replyOpts); + output.WriteLine($"{i}) {reply.Data}"); + } + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/RequestReplyBasic.cs b/examples/Example.NatsIODocs/RequestReplyBasic.cs new file mode 100644 index 000000000..b97986b24 --- /dev/null +++ b/examples/Example.NatsIODocs/RequestReplyBasic.cs @@ -0,0 +1,33 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class RequestReplyBasic(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Set up a service that replies with the current time + // DateTime would be serialized as an ISO-formatted string, just like all primitive types. + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("time")) + { + await msg.ReplyAsync(DateTime.Now); + } + }); + + // Let the subscription task start + await Task.Delay(1000); + + // Make a request + var reply = await client.RequestAsync("time"); + output.WriteLine($"Time is {reply.Data:O}"); + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/RequestReplyCalculator.cs b/examples/Example.NatsIODocs/RequestReplyCalculator.cs new file mode 100644 index 000000000..e619a58ed --- /dev/null +++ b/examples/Example.NatsIODocs/RequestReplyCalculator.cs @@ -0,0 +1,37 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class RequestReplyCalculator(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Calculator service: takes in an array of integers and replies with the sum + // An integer array would be serialized as a JSON array, while a single integer + // would be serialized as a string, just like all other primitive types. + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("calc.sum")) + { + var sum = msg.Data!.Sum(); + await msg.ReplyAsync(sum); + } + }); + + // Let the subscription register + await Task.Delay(1000); + + var reply1 = await client.RequestAsync("calc.sum", [5, 3, 1]); + output.WriteLine($"5 + 3 + 1 = {reply1.Data}"); + + var reply2 = await client.RequestAsync("calc.sum", [10, 7]); + output.WriteLine($"10 + 7 = {reply2.Data}"); + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/RequestReplyHeaders.cs b/examples/Example.NatsIODocs/RequestReplyHeaders.cs new file mode 100644 index 000000000..c5b8ccb4a --- /dev/null +++ b/examples/Example.NatsIODocs/RequestReplyHeaders.cs @@ -0,0 +1,44 @@ +using NATS.Client.Core; +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class RequestReplyHeaders(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // Echo service: reflects the request id back as a response id + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("service")) + { + var responseHeaders = new NatsHeaders + { + ["X-Response-ID"] = msg.Headers?["X-Request-ID"] ?? string.Empty, + }; + await msg.ReplyAsync(msg.Data, headers: responseHeaders); + } + }); + + // Let the subscription register + await Task.Delay(1000); + + // NATS-DOC-START + // Send a request with headers + var requestHeaders = new NatsHeaders + { + ["X-Request-ID"] = "123", + ["X-Priority"] = "high", + }; + + var reply = await client.RequestAsync(subject: "service", data: "data", headers: requestHeaders); + output.WriteLine($"Response: {reply.Data}"); + output.WriteLine($"Response ID: {reply.Headers?["X-Response-ID"]}"); + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/RequestReplyMultipleResponders.cs b/examples/Example.NatsIODocs/RequestReplyMultipleResponders.cs new file mode 100644 index 000000000..b8e87ce93 --- /dev/null +++ b/examples/Example.NatsIODocs/RequestReplyMultipleResponders.cs @@ -0,0 +1,39 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class RequestReplyMultipleResponders(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Two service instances reply on the same subject; the first reply wins + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("calc.add")) + { + await msg.ReplyAsync("calculated result from A"); + } + }); + + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("calc.add")) + { + await msg.ReplyAsync("calculated result from B"); + } + }); + + // Let the subscriptions register + await Task.Delay(1000); + + var reply = await client.RequestAsync("calc.add"); + output.WriteLine($"Got response: {reply.Data}"); + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/RequestReplyNoResponders.cs b/examples/Example.NatsIODocs/RequestReplyNoResponders.cs new file mode 100644 index 000000000..f0aa71206 --- /dev/null +++ b/examples/Example.NatsIODocs/RequestReplyNoResponders.cs @@ -0,0 +1,28 @@ +using NATS.Client.Core; +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class RequestReplyNoResponders(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // RequestAsync throws NatsNoRespondersException by default when nobody is listening + try + { + var reply = await client.RequestAsync(subject: "no.such.service", data: "test"); + output.WriteLine($"Response: {reply.Data}"); + } + catch (NatsNoRespondersException) + { + output.WriteLine("No services available to handle the request"); + } + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/RequestReplyTimeout.cs b/examples/Example.NatsIODocs/RequestReplyTimeout.cs new file mode 100644 index 000000000..dd1979793 --- /dev/null +++ b/examples/Example.NatsIODocs/RequestReplyTimeout.cs @@ -0,0 +1,42 @@ +using NATS.Client.Core; +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class RequestReplyTimeout(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // Slow service: receives the request but delays longer than the caller's timeout + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("service")) + { + await Task.Delay(TimeSpan.FromSeconds(5)); + await msg.ReplyAsync("late reply"); + } + }); + + // Let the subscription register + await Task.Delay(1000); + + // NATS-DOC-START + // Set the per-request timeout via reply options + var replyOpts = new NatsSubOpts { Timeout = TimeSpan.FromSeconds(1) }; + try + { + var reply = await client.RequestAsync("service", replyOpts: replyOpts); + output.WriteLine($"Response: {reply.Data}"); + } + catch (NatsNoReplyException) + { + output.WriteLine("No Response: timed out"); + } + + // NATS-DOC-END + } +} diff --git a/examples/Example.NatsIODocs/SubjectsMonitoring.cs b/examples/Example.NatsIODocs/SubjectsMonitoring.cs new file mode 100644 index 000000000..9879a4391 --- /dev/null +++ b/examples/Example.NatsIODocs/SubjectsMonitoring.cs @@ -0,0 +1,33 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class SubjectsMonitoring(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + _ = Task.Run(async () => + { + // NATS-DOC-START + // Wire tap: subscribe to everything for monitoring + await foreach (var msg in client.SubscribeAsync(">")) + { + output.WriteLine($"[MONITOR] {msg.Subject}: {msg.Data}"); + } + + // NATS-DOC-END + }); + + await Task.Delay(1000); + + await client.PublishAsync("hello", "Hello NATS!"); + await client.PublishAsync("event.new", "click"); + await client.PublishAsync("weather.north.fr", "Temperature: 11C"); + + await Task.Delay(1000); + } +} diff --git a/examples/Example.NatsIODocs/SubjectsMultiWildcard.cs b/examples/Example.NatsIODocs/SubjectsMultiWildcard.cs new file mode 100644 index 000000000..4c8132871 --- /dev/null +++ b/examples/Example.NatsIODocs/SubjectsMultiWildcard.cs @@ -0,0 +1,53 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class SubjectsMultiWildcard(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Subscribe to all non-critical alarms + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("sensor.alarm.*")) + { + output.WriteLine($"[sensor.alarm.*] {msg.Data,-15} ({msg.Subject})"); + } + }); + + // Subscribe to all critical + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("sensor.*.*.critical")) + { + output.WriteLine($"[sensor.*.*.critical] {msg.Data,-15} ({msg.Subject})"); + } + }); + + // Subscribe to everything + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("sensor.>")) + { + output.WriteLine($"[sensor.>] {msg.Data,-15} ({msg.Subject})"); + } + }); + + // Let subscription tasks start + await Task.Delay(1000); + + // Publish to specific subjects + await client.PublishAsync("sensor.alarm.smoke", "kitchen,14:22"); + await client.PublishAsync("sensor.alarm.smoke.critical", "kitchen,14:23"); + await client.PublishAsync("sensor.alarm.water", "basement,16:42"); + await client.PublishAsync("sensor.alarm.water.critical", "basement,16:43"); + + // NATS-DOC-END + await Task.Delay(1000); + } +} diff --git a/examples/Example.NatsIODocs/SubjectsSingleWildcard.cs b/examples/Example.NatsIODocs/SubjectsSingleWildcard.cs new file mode 100644 index 000000000..3cc972622 --- /dev/null +++ b/examples/Example.NatsIODocs/SubjectsSingleWildcard.cs @@ -0,0 +1,52 @@ +using NATS.Net; + +namespace Example.NatsIODocs; + +[Collection("nats-server")] +public class SubjectsSingleWildcard(NatsServerFixture fixture, ITestOutputHelper output) +{ + [Fact] + public async Task RunAsync() + { + await using var client = new NatsClient(fixture.Server.Url); + + // NATS-DOC-START + // Subscribe to the shipped orders + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.*.shipped")) + { + output.WriteLine($"[orders.*.shipped] {msg.Data,-12} ({msg.Subject})"); + } + }); + + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.*.placed")) + { + output.WriteLine($"[orders.*.placed] {msg.Data,-12} ({msg.Subject})"); + } + }); + + // Subscribe to the retail orders + _ = Task.Run(async () => + { + await foreach (var msg in client.SubscribeAsync("orders.retail.*")) + { + output.WriteLine($"[orders.retail.*] {msg.Data,-12} ({msg.Subject})"); + } + }); + + // Let subscription tasks start + await Task.Delay(1000); + + // Publish to specific subjects + await client.PublishAsync("orders.wholesale.placed", "Order W73737"); + await client.PublishAsync("orders.retail.placed", "Order R65432"); + await client.PublishAsync("orders.wholesale.shipped", "Order W73001"); + await client.PublishAsync("orders.retail.shipped", "Order R65321"); + + // NATS-DOC-END + await Task.Delay(1000); + } +} From 998e74efdb59bb608fccc8dee0b971344c560fd5 Mon Sep 17 00:00:00 2001 From: Tomasz Pietrek Date: Mon, 11 May 2026 14:56:47 +0200 Subject: [PATCH 2/6] docs: add Client and Orbit section to README (#1133) Explains the split between the core NATS.Net client and the Orbit utility packages (https://github.com/synadia-io/orbit.net): what each is for, what goes where, and a small layering diagram. Mirrors the section added in nats.rs#1580. Signed-off-by: Tomasz Pietrek --- README.md | 68 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/README.md b/README.md index 37587dcca..ba13d029c 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,74 @@ Head over to [NATS documentation](https://docs.nats.io/nats-concepts/overview) f - **NATS.Client.Serializers.Json**: JSON serializer for ad-hoc types - **NATS.Extensions.Microsoft.DependencyInjection**: extension to configure DI container +## Client and Orbit + +NATS client functionality is split across two layers: the **core client** +(`NATS.Net`, this repo) and **[Orbit](https://github.com/synadia-io/orbit.net)**, +a separate set of packages with higher-level utilities. + +The split exists so the core can stay small, stable, and consistent across +NATS clients in every language, while Orbit can iterate quickly on +opinionated abstractions without dragging the core API along for the ride. + +### Core client (`NATS.Net`) + +- Direct API over Core NATS and JetStream as exposed by `nats-server`. +- Lightweight, unopinionated, performance-oriented. +- API surface kept in **parity** with other official NATS clients + (Rust, Go, Java, JS, Python, C). A feature shipped here should look + the same shape everywhere. +- Stable, conservative versioning. Breaking changes are rare and deliberate. + +### Orbit (`orbit.net`) + +- Higher-level, opinionated abstractions built **on top of** the core client. +- Per-package versioning, so an experimental utility can iterate + without bumping every other piece. +- Free to be language-specific: a .NET-idiomatic API does not need to match + the equivalent in other languages. +- May lag, omit, or extend cross-client parity items. + +### What goes where? + +| Concern | Core (`NATS.Net`) | Orbit | +|----------------------------------------------------|:-------------------:|:-----:| +| Connect, publish, subscribe, request/reply | ✅ | | +| JetStream publish, consumers, streams, KV, OS | ✅ | | +| Service API (request/reply micro-services) | ✅ | | +| Wire-protocol coverage, auth, TLS, reconnection | ✅ | | +| Cross-client parity, conservative semver | ✅ | | +| Opinionated helpers / sugar over core APIs | | ✅ | +| New experimental patterns (e.g. partitioned groups)| | ✅ | +| KV codecs, distributed counters, NATS contexts | | ✅ | +| .NET-idiomatic abstractions with no parity mandate | | ✅ | +| Per-utility versioning, faster API churn allowed | | ✅ | + +> **Rule of thumb:** if it is a thin mapping of something `nats-server` +> already speaks and every official client must expose it, it belongs in +> core. If it is a pattern, helper, or abstraction layered on top, it +> belongs in Orbit. + +### Layering + +```text + ┌──────────────────────────────────────────────────────┐ + │ Application code │ + └──────────────┬───────────────────────────┬───────────┘ + │ │ + ▼ ▼ + ┌───────────────────┐ ┌───────────────────┐ + │ Orbit packages │ uses │ NATS.Net (core) │ + │ (opinionated, │──────▶│ (parity, stable, │ + │ per-pkg semver) │ │ protocol-level) │ + └───────────────────┘ └─────────┬─────────┘ + │ + ▼ + ┌─────────────┐ + │ nats-server │ + └─────────────┘ +``` + ## Contributing You are welcome to contribute to this project. Here are some steps to get you started: From 3e1912d4a14fbb7b97910934e1d978875d845fc1 Mon Sep 17 00:00:00 2001 From: monty Date: Wed, 13 May 2026 08:14:55 +0100 Subject: [PATCH 3/6] Add server error event (#745) * Add server error event * Bump test session timeout to 600s * Increase test ConnectTimeout to match other proxy-based tests * Simplify ErrorHandlerTest publish-and-wait helper * Route NatsConnection events through PushEvent * Parse server error kind on NatsServerErrorEventArgs * Bump auth timeout in restricted-user fixture * Cover null input in NatsServerErrorEventArgs tests * Add server-errors docs page --- src/NATS.Client.Core/INatsConnection.cs | 5 + .../Internal/NatsReadProtocolProcessor.cs | 2 + src/NATS.Client.Core/NatsConnection.cs | 23 +++-- src/NATS.Client.Core/NatsEventArgs.cs | 59 +++++++++++ .../ErrorHandlerTest.cs | 98 +++++++++++++++++++ .../NatsServerErrorEventArgsTest.cs | 43 ++++++++ .../NatsJsContextFactoryTest.cs | 2 + .../resources/configs/restricted-user.conf | 2 + .../Advanced/ServerErrorsPage.cs | 59 +++++++++++ .../site_src/documentation/advanced/intro.md | 1 + .../documentation/advanced/server-errors.md | 29 ++++++ tools/site_src/documentation/toc.yml | 2 + 12 files changed, 318 insertions(+), 7 deletions(-) create mode 100644 tests/NATS.Client.Core2.Tests/ErrorHandlerTest.cs create mode 100644 tests/NATS.Client.CoreUnit.Tests/NatsServerErrorEventArgsTest.cs create mode 100644 tests/NATS.Net.DocsExamples/Advanced/ServerErrorsPage.cs create mode 100644 tools/site_src/documentation/advanced/server-errors.md diff --git a/src/NATS.Client.Core/INatsConnection.cs b/src/NATS.Client.Core/INatsConnection.cs index 45b859e41..1e0bc7959 100644 --- a/src/NATS.Client.Core/INatsConnection.cs +++ b/src/NATS.Client.Core/INatsConnection.cs @@ -37,6 +37,11 @@ public interface INatsConnection : INatsClient /// public event AsyncEventHandler? LameDuckModeActivated; + /// + /// Event that is raised when server sends an error message ('-ERR'). + /// + public event AsyncEventHandler? ServerError; + /// /// Server information received from the NATS server. /// diff --git a/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs b/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs index 213db7aec..c33cb6c09 100644 --- a/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs +++ b/src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs @@ -404,6 +404,7 @@ private async ValueTask> DispatchCommandAsync(int code, R var newPosition = newBuffer.PositionOf((byte)'\n'); var error = ParseError(newBuffer.Slice(0, newBuffer.GetOffset(newPosition!.Value) - 1)); _logger.LogError(NatsLogEvents.Protocol, "Server error {Error}", error); + _connection.PushEvent(NatsEvent.ServerError, new NatsServerErrorEventArgs(error)); _waitForPongOrErrorSignal.TrySetObservedException(new NatsServerException(error)); return newBuffer.Slice(newBuffer.GetPosition(1, newPosition!.Value)); } @@ -411,6 +412,7 @@ private async ValueTask> DispatchCommandAsync(int code, R { var error = ParseError(buffer.Slice(0, buffer.GetOffset(position.Value) - 1)); _logger.LogError(NatsLogEvents.Protocol, "Server error {Error}", error); + _connection.PushEvent(NatsEvent.ServerError, new NatsServerErrorEventArgs(error)); _waitForPongOrErrorSignal.TrySetObservedException(new NatsServerException(error)); return buffer.Slice(buffer.GetPosition(1, position.Value)); } diff --git a/src/NATS.Client.Core/NatsConnection.cs b/src/NATS.Client.Core/NatsConnection.cs index ae2aaad2e..d97564d24 100644 --- a/src/NATS.Client.Core/NatsConnection.cs +++ b/src/NATS.Client.Core/NatsConnection.cs @@ -29,6 +29,7 @@ internal enum NatsEvent LameDuckModeActivated, ConnectionFailed, SlowConsumerDetected, + ServerError, } public partial class NatsConnection : INatsConnection @@ -120,6 +121,8 @@ public NatsConnection(NatsOpts opts) public event AsyncEventHandler? LameDuckModeActivated; + public event AsyncEventHandler? ServerError; + public INatsConnection Connection => this; public NatsOpts Opts { get; } @@ -152,7 +155,7 @@ internal ServerInfo? WritableServerInfo { if (value?.LameDuckMode == true) { - _eventChannel.Writer.TryWrite((NatsEvent.LameDuckModeActivated, new NatsLameDuckModeActivatedEventArgs(_currentConnectUri!.Uri))); + PushEvent(NatsEvent.LameDuckModeActivated, new NatsLameDuckModeActivatedEventArgs(_currentConnectUri!.Uri)); } Interlocked.Exchange(ref _writableServerInfo, value); @@ -240,11 +243,11 @@ public async ValueTask ConnectAsync() public void OnMessageDropped(NatsSubBase natsSub, int pending, NatsMsg msg) { var subject = msg.Subject; - _eventChannel.Writer.TryWrite((NatsEvent.MessageDropped, new NatsMessageDroppedEventArgs(natsSub, pending, subject, msg.ReplyTo, msg.Headers, msg.Data))); + PushEvent(NatsEvent.MessageDropped, new NatsMessageDroppedEventArgs(natsSub, pending, subject, msg.ReplyTo, msg.Headers, msg.Data)); if (natsSub.TryMarkSlowConsumer()) { - _eventChannel.Writer.TryWrite((NatsEvent.SlowConsumerDetected, new NatsSlowConsumerEventArgs(natsSub))); + PushEvent(NatsEvent.SlowConsumerDetected, new NatsSlowConsumerEventArgs(natsSub)); if (!Opts.SuppressSlowConsumerWarnings) { @@ -415,6 +418,9 @@ internal ValueTask UnsubscribeAsync(int sid) return default; } + internal void PushEvent(NatsEvent @event, NatsEventArgs args) + => _eventChannel.Writer.TryWrite((@event, args)); + private async ValueTask InitialConnectAsync() { Debug.Assert(ConnectionState == NatsConnectionState.Connecting, "Connection state"); @@ -514,7 +520,7 @@ private async ValueTask InitialConnectAsync() _initialConnectCts.Cancel(); #pragma warning restore VSTHRD103 _reconnectLoopTask = Task.Run(ReconnectLoop); - _eventChannel.Writer.TryWrite((NatsEvent.ConnectionOpened, new NatsEventArgs(url?.ToString() ?? string.Empty))); + PushEvent(NatsEvent.ConnectionOpened, new NatsEventArgs(url?.ToString() ?? string.Empty)); } } @@ -745,7 +751,7 @@ private async void ReconnectLoop() } // Invoke event after state changed - _eventChannel.Writer.TryWrite((NatsEvent.ConnectionDisconnected, new NatsEventArgs(_currentConnectUri?.ToString() ?? string.Empty))); + PushEvent(NatsEvent.ConnectionDisconnected, new NatsEventArgs(_currentConnectUri?.ToString() ?? string.Empty)); // Cleanup current socket await DisposeSocketAsync(true).ConfigureAwait(false); @@ -810,7 +816,7 @@ private async void ReconnectLoop() var attempted = _currentConnectUri ?? url; _logger.LogWarning(NatsLogEvents.Connection, ex, "Failed to connect NATS {Url} [{ReconnectCount}]", attempted, reconnectCount); - _eventChannel.Writer.TryWrite((NatsEvent.ReconnectFailed, new NatsEventArgs(attempted?.ToString() ?? string.Empty))); + PushEvent(NatsEvent.ReconnectFailed, new NatsEventArgs(attempted?.ToString() ?? string.Empty)); if (debug) { @@ -839,7 +845,7 @@ private async void ReconnectLoop() StartPingTimer(_pingTimerCancellationTokenSource.Token); _waitForOpenConnection.TrySetResult(); _reconnectLoopTask = Task.Run(ReconnectLoop); - _eventChannel.Writer.TryWrite((NatsEvent.ConnectionOpened, new NatsEventArgs(url.ToString()))); + PushEvent(NatsEvent.ConnectionOpened, new NatsEventArgs(url.ToString())); } } catch (Exception ex) @@ -923,6 +929,9 @@ private async Task PublishEventsAsync() case NatsEvent.LameDuckModeActivated when LameDuckModeActivated != null && args is NatsLameDuckModeActivatedEventArgs uri: await LameDuckModeActivated.InvokeAsync(this, uri).ConfigureAwait(false); break; + case NatsEvent.ServerError when ServerError != null && args is NatsServerErrorEventArgs error: + await ServerError.InvokeAsync(this, error).ConfigureAwait(false); + break; } } } diff --git a/src/NATS.Client.Core/NatsEventArgs.cs b/src/NATS.Client.Core/NatsEventArgs.cs index 46a3a6e77..c6754cba1 100644 --- a/src/NATS.Client.Core/NatsEventArgs.cs +++ b/src/NATS.Client.Core/NatsEventArgs.cs @@ -1,6 +1,20 @@ // ReSharper disable UnusedAutoPropertyAccessor.Global - properties are used by consumers outside of this library namespace NATS.Client.Core; +public enum NatsServerErrorKind +{ + Unknown = 0, + AuthorizationViolation, + AuthenticationExpired, + AuthenticationRevoked, + AccountAuthenticationExpired, + PermissionsViolation, + StaleConnection, + MaximumConnectionsExceeded, + MaximumAccountConnectionsExceeded, + MaximumSubscriptionsExceeded, +} + public class NatsEventArgs : EventArgs { public NatsEventArgs(string message) => Message = message; @@ -52,3 +66,48 @@ public NatsSlowConsumerEventArgs(NatsSubBase subscription) public NatsSubBase Subscription { get; } } + +public class NatsServerErrorEventArgs : NatsEventArgs +{ + public NatsServerErrorEventArgs(string error) + : base($"Server error {error}") + { + Error = error; + Kind = ParseKind(error); + } + + public string Error { get; } + + public NatsServerErrorKind Kind { get; } + + // Match the literal -ERR strings the server emits to clients (see + // nats-server/server/client.go and errors.go). Server casing is fixed + // in source, so Ordinal is enough. Anything unrecognised stays as + // Unknown; the raw string is always available on Error. + private static NatsServerErrorKind ParseKind(string error) + { + if (string.IsNullOrEmpty(error)) + return NatsServerErrorKind.Unknown; + + if (error.StartsWith("Permissions Violation", StringComparison.Ordinal)) + return NatsServerErrorKind.PermissionsViolation; + if (error.StartsWith("Authorization Violation", StringComparison.Ordinal)) + return NatsServerErrorKind.AuthorizationViolation; + if (error.StartsWith("User Authentication Expired", StringComparison.Ordinal)) + return NatsServerErrorKind.AuthenticationExpired; + if (error.StartsWith("User Authentication Revoked", StringComparison.Ordinal)) + return NatsServerErrorKind.AuthenticationRevoked; + if (error.StartsWith("Account Authentication Expired", StringComparison.Ordinal)) + return NatsServerErrorKind.AccountAuthenticationExpired; + if (error.StartsWith("Stale Connection", StringComparison.Ordinal)) + return NatsServerErrorKind.StaleConnection; + if (error.StartsWith("Maximum Subscriptions", StringComparison.Ordinal)) + return NatsServerErrorKind.MaximumSubscriptionsExceeded; + if (error.StartsWith("Maximum Account Active Connections", StringComparison.Ordinal)) + return NatsServerErrorKind.MaximumAccountConnectionsExceeded; + if (error.StartsWith("Maximum Connections", StringComparison.Ordinal)) + return NatsServerErrorKind.MaximumConnectionsExceeded; + + return NatsServerErrorKind.Unknown; + } +} diff --git a/tests/NATS.Client.Core2.Tests/ErrorHandlerTest.cs b/tests/NATS.Client.Core2.Tests/ErrorHandlerTest.cs new file mode 100644 index 000000000..ef9a19f6c --- /dev/null +++ b/tests/NATS.Client.Core2.Tests/ErrorHandlerTest.cs @@ -0,0 +1,98 @@ +using Microsoft.Extensions.Logging; +using NATS.Client.Core.Tests; +using NATS.Client.TestUtilities; + +namespace NATS.Client.Core2.Tests; + +[Collection("nats-server-restricted-user")] +public class ErrorHandlerTest +{ + private readonly ITestOutputHelper _output; + private readonly NatsServerRestrictedUserFixture _server; + + public ErrorHandlerTest(ITestOutputHelper output, NatsServerRestrictedUserFixture server) + { + _output = output; + _server = server; + } + + [Fact] + public async Task Handle_permissions_violation() + { + var logger = new InMemoryTestLoggerFactory(LogLevel.Error); + + var proxy = new NatsProxy(_server.Port); + await using var nats = new NatsConnection(new NatsOpts + { + Url = $"nats://127.0.0.1:{proxy.Port}", + ConnectTimeout = TimeSpan.FromSeconds(10), + LoggerFactory = logger, + AuthOpts = new NatsAuthOpts { Username = "u" }, + }); + + var errors = new List(); + + nats.ServerError += (_, args) => + { + lock (errors) + { + errors.Add(args); + } + + return default; + }; + + var prefix = _server.GetNextId(); + + async Task PublishAndWaitForPong(string subject, string marker) + { + await nats.PublishAsync(subject, $"_{prefix}_{marker}_"); + await nats.PingAsync(); + await Retry.Until( + $"published {marker} and pinged", + () => + { + var published = false; + foreach (var frame in proxy.AllFrames) + { + if (frame.Origin == "C" && frame.Message.Contains($"_{prefix}_{marker}_")) + { + published = true; + continue; + } + + if (published && frame.Origin == "S" && frame.Message == "PONG") + { + return true; + } + } + + return false; + }); + } + + await PublishAndWaitForPong("x", "published_1"); + await PublishAndWaitForPong("y", "published_2"); + + Assert.Contains(proxy.AllFrames, f => f.Origin == "S" && f.Message == "-ERR 'Permissions Violation for Publish to \"y\"'"); + + await Retry.Until( + "error is logged", + () => + { + return logger.Logs.Any(x => x.LogLevel == LogLevel.Error && x.Message == "Server error Permissions Violation for Publish to \"y\""); + }); + + await Retry.Until( + "server error event", + () => + { + lock (errors) + { + return errors.Any(e => + e.Error == "Permissions Violation for Publish to \"y\"" + && e.Kind == NatsServerErrorKind.PermissionsViolation); + } + }); + } +} diff --git a/tests/NATS.Client.CoreUnit.Tests/NatsServerErrorEventArgsTest.cs b/tests/NATS.Client.CoreUnit.Tests/NatsServerErrorEventArgsTest.cs new file mode 100644 index 000000000..349451595 --- /dev/null +++ b/tests/NATS.Client.CoreUnit.Tests/NatsServerErrorEventArgsTest.cs @@ -0,0 +1,43 @@ +namespace NATS.Client.Core.Tests; + +public class NatsServerErrorEventArgsTest +{ + // Pairs of (raw -ERR text, expected Kind). Casing matches what the + // server actually emits; mismatches must fall through to Unknown. + [Theory] + [InlineData("Permissions Violation for Publish to \"foo\"", NatsServerErrorKind.PermissionsViolation)] + [InlineData("Permissions Violation for Subscription to \"foo\"", NatsServerErrorKind.PermissionsViolation)] + [InlineData("Permissions Violation for Subscription to \"foo\" using queue \"q\"", NatsServerErrorKind.PermissionsViolation)] + [InlineData("Permissions Violation for Publish with Reply of \"r\"", NatsServerErrorKind.PermissionsViolation)] + [InlineData("Authorization Violation", NatsServerErrorKind.AuthorizationViolation)] + [InlineData("User Authentication Expired", NatsServerErrorKind.AuthenticationExpired)] + [InlineData("User Authentication Revoked", NatsServerErrorKind.AuthenticationRevoked)] + [InlineData("Account Authentication Expired", NatsServerErrorKind.AccountAuthenticationExpired)] + [InlineData("Stale Connection", NatsServerErrorKind.StaleConnection)] + [InlineData("Maximum Connections Exceeded", NatsServerErrorKind.MaximumConnectionsExceeded)] + [InlineData("Maximum Account Active Connections Exceeded", NatsServerErrorKind.MaximumAccountConnectionsExceeded)] + [InlineData("Maximum Subscriptions Exceeded", NatsServerErrorKind.MaximumSubscriptionsExceeded)] + public void Recognised_kinds(string error, NatsServerErrorKind expected) + { + var args = new NatsServerErrorEventArgs(error); + Assert.Equal(expected, args.Kind); + Assert.Equal(error, args.Error); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData("Authentication Timeout")] + [InlineData("Maximum Payload Violation")] + [InlineData("Invalid Subject")] + [InlineData("Invalid Publish Subject")] + [InlineData("Invalid Client Protocol")] + [InlineData("permissions violation")] + [InlineData("Some Future Server Error")] + public void Unknown_kinds(string? error) + { + var args = new NatsServerErrorEventArgs(error!); + Assert.Equal(NatsServerErrorKind.Unknown, args.Kind); + Assert.Equal(error, args.Error); + } +} diff --git a/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs b/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs index 5627affb8..2f353dd71 100644 --- a/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs +++ b/tests/NATS.Client.JetStream.Tests/NatsJsContextFactoryTest.cs @@ -86,6 +86,8 @@ public class MockConnection : INatsConnection public event AsyncEventHandler? SlowConsumerDetected; public event AsyncEventHandler? LameDuckModeActivated; + + public event AsyncEventHandler? ServerError; #pragma warning restore CS0067 public INatsServerInfo? ServerInfo { get; } = null; diff --git a/tests/NATS.Client.TestUtilities/resources/configs/restricted-user.conf b/tests/NATS.Client.TestUtilities/resources/configs/restricted-user.conf index 4c8a6ab04..b640cbadf 100644 --- a/tests/NATS.Client.TestUtilities/resources/configs/restricted-user.conf +++ b/tests/NATS.Client.TestUtilities/resources/configs/restricted-user.conf @@ -1,4 +1,6 @@ authorization: { + # Generous timeout to avoid proxy-test flap on slow CI runners. + timeout: 30 users: [ { user:u, permissions: { publish:x } } ] diff --git a/tests/NATS.Net.DocsExamples/Advanced/ServerErrorsPage.cs b/tests/NATS.Net.DocsExamples/Advanced/ServerErrorsPage.cs new file mode 100644 index 000000000..9049359ef --- /dev/null +++ b/tests/NATS.Net.DocsExamples/Advanced/ServerErrorsPage.cs @@ -0,0 +1,59 @@ +// ReSharper disable SuggestVarOrType_SimpleTypes +// ReSharper disable SuggestVarOrType_Elsewhere +#pragma warning disable SA1123 +#pragma warning disable SA1124 +#pragma warning disable SA1509 +#pragma warning disable IDE0007 +#pragma warning disable IDE0008 + +using NATS.Client.Core; + +namespace NATS.Net.DocsExamples.Advanced; + +public class ServerErrorsPage +{ + public async Task Run() + { + Console.WriteLine("____________________________________________________________"); + Console.WriteLine("NATS.Net.DocsExamples.Advanced.ServerErrorsPage"); + + { + #region server-error + await using NatsConnection nc = new NatsConnection(); + + nc.ServerError += (sender, args) => + { + Console.WriteLine($"Server error: {args.Error}"); + return default; + }; + #endregion + } + + { + #region server-error-kind + await using NatsConnection nc = new NatsConnection(); + + nc.ServerError += (sender, args) => + { + switch (args.Kind) + { + case NatsServerErrorKind.PermissionsViolation: + Console.WriteLine($"Permission denied: {args.Error}"); + break; + case NatsServerErrorKind.AuthorizationViolation: + case NatsServerErrorKind.AuthenticationExpired: + case NatsServerErrorKind.AuthenticationRevoked: + case NatsServerErrorKind.AccountAuthenticationExpired: + Console.WriteLine($"Auth problem: {args.Error}"); + break; + default: + Console.WriteLine($"Server error ({args.Kind}): {args.Error}"); + break; + } + + return default; + }; + #endregion + } + } +} diff --git a/tools/site_src/documentation/advanced/intro.md b/tools/site_src/documentation/advanced/intro.md index bd6609350..5142eda87 100644 --- a/tools/site_src/documentation/advanced/intro.md +++ b/tools/site_src/documentation/advanced/intro.md @@ -93,6 +93,7 @@ essentially sent back to back after they're picked up from internal queues and b ## What's Next - [Slow Consumers](slow-consumers.md) explains how to detect and handle subscribers that can't keep up with the message rate, including the `MessageDropped` and `SlowConsumerDetected` events. +- [Server Errors](server-errors.md) covers the `ServerError` event for observing `-ERR` messages from the server (permission denials, auth issues, server-side limits). - [Serialization](serialization.md) is the process of converting an object into a format that can be stored or transmitted. - [Security](security.md) is an important aspect of any distributed system. NATS provides a number of security features to help you secure your applications. - [OpenTelemetry](opentelemetry.md) covers built-in distributed tracing with automatic trace context propagation, activity filtering, and enrichment. diff --git a/tools/site_src/documentation/advanced/server-errors.md b/tools/site_src/documentation/advanced/server-errors.md new file mode 100644 index 000000000..556c36f95 --- /dev/null +++ b/tools/site_src/documentation/advanced/server-errors.md @@ -0,0 +1,29 @@ +# Server Errors + +The NATS server sends `-ERR` protocol messages to a client when something goes wrong on the server side: a permission denial, an authentication problem, a client-level limit, and so on. The `ServerError` event on `NatsConnection` raises one of these messages every time the server emits one, so an application can observe them without scraping logs. + +## Subscribing to the Event + +Hook the event the same way as any other connection event. The raw error text is on `args.Error`: + +[!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/ServerErrorsPage.cs#server-error)] + +## Classifying the Error + +`args.Kind` is parsed from the raw text and exposes the common server errors as a [`NatsServerErrorKind`](xref:NATS.Client.Core.NatsServerErrorKind) enum value. Anything the parser does not recognise stays as `NatsServerErrorKind.Unknown`; `args.Error` always carries the original text. + +[!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/ServerErrorsPage.cs#server-error-kind)] + +The recognised kinds are: + +- `PermissionsViolation` - subscribe or publish denied for the connected user +- `AuthorizationViolation` - authorization failure during connect +- `AuthenticationExpired` - user authentication expired +- `AuthenticationRevoked` - user authentication revoked +- `AccountAuthenticationExpired` - account authentication expired +- `StaleConnection` - server timed out a stalled connection +- `MaximumConnectionsExceeded` - server-wide connection limit reached +- `MaximumAccountConnectionsExceeded` - per-account connection limit reached +- `MaximumSubscriptionsExceeded` - per-connection subscription limit reached + +The event fires for every `-ERR` line, including messages whose `Kind` is `Unknown`. Match on `args.Error` if the application needs to handle a server error not covered by the enum. diff --git a/tools/site_src/documentation/toc.yml b/tools/site_src/documentation/toc.yml index 8df2e68b7..44fc8566e 100644 --- a/tools/site_src/documentation/toc.yml +++ b/tools/site_src/documentation/toc.yml @@ -38,6 +38,8 @@ items: - name: Slow Consumers href: advanced/slow-consumers.md + - name: Server Errors + href: advanced/server-errors.md - name: Serialization href: advanced/serialization.md - name: Security From d5c9d044f87ea754c08c7d7ab53bc748201c7d78 Mon Sep 17 00:00:00 2001 From: monty Date: Wed, 13 May 2026 10:49:15 +0100 Subject: [PATCH 4/6] ci: test against v2.12, drop v2.10 (#1136) --- .github/workflows/perf.yml | 2 +- .github/workflows/test_linux.yml | 2 +- .github/workflows/test_linux_core.yml | 2 +- .github/workflows/test_windows.yml | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/perf.yml b/.github/workflows/perf.yml index e6ae413b6..fa0d8fc43 100644 --- a/.github/workflows/perf.yml +++ b/.github/workflows/perf.yml @@ -13,7 +13,7 @@ jobs: fail-fast: false matrix: config: - - branch: 'v2.10' + - branch: 'v2.12' - branch: 'latest' - branch: 'main' runs-on: ubuntu-latest diff --git a/.github/workflows/test_linux.yml b/.github/workflows/test_linux.yml index 20f31582a..b0e2c811d 100644 --- a/.github/workflows/test_linux.yml +++ b/.github/workflows/test_linux.yml @@ -13,8 +13,8 @@ jobs: fail-fast: false matrix: config: - - branch: 'v2.10' - branch: 'v2.11' + - branch: 'v2.12' - branch: 'latest' - branch: 'main' runs-on: ubuntu-latest diff --git a/.github/workflows/test_linux_core.yml b/.github/workflows/test_linux_core.yml index 5413a141a..8d90af12c 100644 --- a/.github/workflows/test_linux_core.yml +++ b/.github/workflows/test_linux_core.yml @@ -13,8 +13,8 @@ jobs: fail-fast: false matrix: config: - - branch: 'v2.10' - branch: 'v2.11' + - branch: 'v2.12' - branch: 'latest' - branch: 'main' runs-on: ubuntu-latest diff --git a/.github/workflows/test_windows.yml b/.github/workflows/test_windows.yml index 935f114bd..00a0a08b2 100644 --- a/.github/workflows/test_windows.yml +++ b/.github/workflows/test_windows.yml @@ -14,8 +14,8 @@ jobs: fail-fast: false matrix: config: - - branch: 'v2.10' - branch: 'v2.11' + - branch: 'v2.12' - branch: 'latest' - branch: 'main' runs-on: windows-latest From e18d8a552c047678a94cde4d91a22d62d9f8eb4f Mon Sep 17 00:00:00 2001 From: monty Date: Wed, 13 May 2026 11:12:28 +0100 Subject: [PATCH 5/6] Release 2.8.0 (#1135) --- version.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/version.txt b/version.txt index fa77cc3c9..834f26295 100644 --- a/version.txt +++ b/version.txt @@ -1 +1 @@ -2.8.0-preview.3 +2.8.0 From 8b7964b3ca1c616f5216082c558482774a9e0f4c Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Wed, 13 May 2026 11:30:41 +0100 Subject: [PATCH 6/6] Release 3.0.0-preview.6 --- version.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/version.txt b/version.txt index 95d8a112f..7965b72a5 100644 --- a/version.txt +++ b/version.txt @@ -1 +1 @@ -3.0.0-preview.5 +3.0.0-preview.6