diff --git a/tests/NATS.Net.DocsExamples/Advanced/SlowConsumerPage.cs b/tests/NATS.Net.DocsExamples/Advanced/SlowConsumerPage.cs new file mode 100644 index 000000000..8022b9f45 --- /dev/null +++ b/tests/NATS.Net.DocsExamples/Advanced/SlowConsumerPage.cs @@ -0,0 +1,82 @@ +// 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 System.Threading.Channels; +using NATS.Client.Core; + +namespace NATS.Net.DocsExamples.Advanced; + +public class SlowConsumerPage +{ + public async Task Run() + { + Console.WriteLine("____________________________________________________________"); + Console.WriteLine("NATS.Net.DocsExamples.Advanced.SlowConsumerPage"); + + { + #region message-dropped + await using NatsConnection nc = new NatsConnection(); + + nc.MessageDropped += (sender, args) => + { + Console.WriteLine($"Dropped message on '{args.Subject}' with {args.Pending} pending messages"); + return default; + }; + #endregion + } + + { + #region slow-consumer-detected + await using NatsConnection nc = new NatsConnection(); + + nc.SlowConsumerDetected += (sender, args) => + { + Console.WriteLine($"Slow consumer detected on '{args.Subscription.Subject}'"); + return default; + }; + #endregion + } + + { + #region tuning + // Set global default capacity + NatsOpts opts = new NatsOpts { SubPendingChannelCapacity = 4096 }; + await using NatsConnection nc = new NatsConnection(opts); + + // Override capacity for a specific subscription + NatsSubOpts subOpts = new NatsSubOpts + { + ChannelOpts = new NatsSubChannelOpts { Capacity = 8192 }, + }; + + await foreach (NatsMsg msg in nc.SubscribeAsync("events.>", opts: subOpts)) + { + Console.WriteLine($"Received: {msg.Data}"); + } + #endregion + } + + { + #region suppress-warnings + NatsOpts opts = new NatsOpts + { + SuppressSlowConsumerWarnings = true, + }; + + await using NatsConnection nc = new NatsConnection(opts); + + // Events still fire even when log warnings are suppressed + nc.SlowConsumerDetected += (sender, args) => + { + Console.WriteLine($"Slow consumer on '{args.Subscription.Subject}'"); + return default; + }; + #endregion + } + } +} diff --git a/tools/site_src/documentation/advanced/intro.md b/tools/site_src/documentation/advanced/intro.md index ce111115b..09c96ecd1 100644 --- a/tools/site_src/documentation/advanced/intro.md +++ b/tools/site_src/documentation/advanced/intro.md @@ -92,6 +92,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. - [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. - [AOT Deployment](aot.md) is a way to deploy your applications as native platform executables, which produces faster startup times and better performance in most cases. diff --git a/tools/site_src/documentation/advanced/slow-consumers.md b/tools/site_src/documentation/advanced/slow-consumers.md new file mode 100644 index 000000000..97e173c8b --- /dev/null +++ b/tools/site_src/documentation/advanced/slow-consumers.md @@ -0,0 +1,42 @@ +# Slow Consumers + +When a subscriber can't keep up with the rate of incoming messages, its internal bounded channel fills up. +By default, new messages are dropped when this happens. This is aligned with NATS at-most-once delivery. + +## Detecting Dropped Messages + +The `MessageDropped` event fires every time a message is dropped from a subscription's internal channel: + +[!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/SlowConsumerPage.cs#message-dropped)] + +## Detecting Slow Consumers + +The `SlowConsumerDetected` event (added in v2.7.2) fires once per slow consumer episode. +It resets after the subscription catches up (channel drains), so it will fire again if the subscription falls behind a second time: + +[!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/SlowConsumerPage.cs#slow-consumer-detected)] + +## Tuning Channel Capacity + +Each subscription has an internal bounded channel with a default capacity of 1024 messages. +You can tune this globally or per subscription: + +[!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/SlowConsumerPage.cs#tuning)] + +## Channel Full Mode + +The `SubPendingChannelFullMode` option controls what happens when the channel is full: + +- `BoundedChannelFullMode.DropNewest` (default for `NatsConnection`) - newest messages are dropped +- `BoundedChannelFullMode.Wait` (default for `NatsClient`) - backpressure is applied, but the server may disconnect you as a slow consumer + +> [!WARNING] +> Using `BoundedChannelFullMode.Wait` prevents message loss at the client level, but if the subscriber +> can't keep up, the NATS server itself may disconnect the client as a slow consumer. + +## Suppressing Warnings + +By default, a warning is logged once per slow consumer episode. You can suppress these logs while still +receiving the events: + +[!code-csharp[](../../../../tests/NATS.Net.DocsExamples/Advanced/SlowConsumerPage.cs#suppress-warnings)] diff --git a/tools/site_src/documentation/toc.yml b/tools/site_src/documentation/toc.yml index 481d7f7b0..c470f90ae 100644 --- a/tools/site_src/documentation/toc.yml +++ b/tools/site_src/documentation/toc.yml @@ -36,6 +36,8 @@ - name: Advanced Options href: advanced/intro.md items: + - name: Slow Consumers + href: advanced/slow-consumers.md - name: Serialization href: advanced/serialization.md - name: Security