diff --git a/CLAUDE.md b/CLAUDE.md index 85347dd..4e572b4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -145,6 +145,56 @@ Client PUB → parser → permission check → Sublist.Match() → └─ JetStream: if subject matches a stream, store + deliver to consumers ``` +## NuGet Package Management + +This solution uses **Central Package Management (CPM)** via `Directory.Packages.props` at the repo root. All package versions are defined centrally there. + +- In `.csproj` files, use `` **without** a `Version` attribute +- To add a new package: add a `` entry in `Directory.Packages.props`, then reference it without version in the project's csproj +- To update a version: change it only in `Directory.Packages.props` — all projects pick it up automatically +- Never specify `Version` on `` in individual csproj files + +## Logging + +Use **Microsoft.Extensions.Logging** (`ILogger`) for all logging throughout the server. Wire up **Serilog** as the logging provider in the host application. + +- Inject `ILogger` via constructor in all components (NatsServer, NatsClient, etc.) +- Use **Serilog.Context.LogContext** to push contextual properties (client ID, remote endpoint, subscription subject) so they appear on all log entries within that scope +- Use structured logging with message templates: `logger.LogInformation("Client {ClientId} subscribed to {Subject}", id, subject)` — never string interpolation +- Log levels: `Trace` for protocol bytes, `Debug` for per-message flow, `Information` for lifecycle events (connect/disconnect), `Warning` for protocol violations, `Error` for unexpected failures + +## Testing + +- **xUnit 3** for test framework +- **Shouldly** for assertions — use `value.ShouldBe(expected)`, `action.ShouldThrow()`, etc. Do NOT use `Assert.*` from xUnit +- **NSubstitute** for mocking/substitution when needed +- Do **NOT** use FluentAssertions or Moq — these are explicitly excluded +- Test project uses global `using Shouldly;` + +## Porting Guidelines + +- Use modern .NET 10 / C# 14 best practices (primary constructors, collection expressions, `field` keyword where stable, file-scoped namespaces, raw string literals, etc.) +- Prefer `readonly record struct` for small value types over mutable structs +- Use `required` properties and `init` setters for initialization-only state +- Use pattern matching and switch expressions where they improve clarity +- Prefer `System.Text.Json` source generators for JSON serialization +- Use `ValueTask` where appropriate for hot-path async methods + +## Agent Model Guidance + +- **Sonnet** (`model: "sonnet"`) — use for simpler implementation tasks: straightforward file modifications, adding packages, converting assertions, boilerplate code +- **Opus** (default) — use for complex tasks, architectural decisions, design work, tricky protocol logic, and code review +- **Parallel subagents** — use where tasks are independent and don't touch the same files (e.g., converting test files in parallel, adding packages while updating docs) + +## Documentation + +Follow the documentation rules in [`documentation_rules.md`](documentation_rules.md) for all project documentation. Key points: + +- Documentation lives in `Documentation/` with component subfolders (Protocol, Subscriptions, Server, Configuration, Operations) +- Use `PascalCase.md` file names, always specify language on code blocks, use real code snippets (not invented examples) +- Update documentation when code changes — see the trigger rules and component map in the rules file +- Technical and direct tone, explain "why" not just "what", present tense + ## Conventions - Reference the Go implementation file and line when porting a subsystem diff --git a/Directory.Packages.props b/Directory.Packages.props new file mode 100644 index 0000000..267b8c9 --- /dev/null +++ b/Directory.Packages.props @@ -0,0 +1,23 @@ + + + true + + + + + + + + + + + + + + + + + + + + diff --git a/Documentation/Configuration/Overview.md b/Documentation/Configuration/Overview.md new file mode 100644 index 0000000..b1d8735 --- /dev/null +++ b/Documentation/Configuration/Overview.md @@ -0,0 +1,185 @@ +# Configuration Overview + +`NatsOptions` is the single configuration object passed to `NatsServer` at construction time. The host application (`Program.cs`) also reads CLI arguments to populate it before server startup. + +Go reference: `golang/nats-server/server/opts.go` + +--- + +## NatsOptions Class + +`NatsOptions` is a plain mutable class with no validation logic — values are consumed directly by `NatsServer` and `NatsClient`. + +```csharp +namespace NATS.Server; + +public sealed class NatsOptions +{ + public string Host { get; set; } = "0.0.0.0"; + public int Port { get; set; } = 4222; + public string? ServerName { get; set; } + public int MaxPayload { get; set; } = 1024 * 1024; // 1MB + public int MaxControlLine { get; set; } = 4096; + public int MaxConnections { get; set; } = 65536; + public TimeSpan PingInterval { get; set; } = TimeSpan.FromMinutes(2); + public int MaxPingsOut { get; set; } = 2; +} +``` + +### Option reference + +| Option | Type | Default | Description | +|--------|------|---------|-------------| +| `Host` | `string` | `"0.0.0.0"` | Bind address for the TCP listener. Use `"127.0.0.1"` to restrict to loopback. | +| `Port` | `int` | `4222` | Client listen port. Standard NATS client port. | +| `ServerName` | `string?` | `null` (auto: `nats-dotnet-{MachineName}`) | Server name sent in the `INFO` message. When `null`, the server constructs a name from `Environment.MachineName`. | +| `MaxPayload` | `int` | `1048576` (1 MB) | Maximum allowed message payload in bytes. Clients that publish larger payloads are rejected. | +| `MaxControlLine` | `int` | `4096` | Maximum protocol control line size in bytes. Matches the Go server default. | +| `MaxConnections` | `int` | `65536` | Maximum concurrent client connections the server will accept. | +| `PingInterval` | `TimeSpan` | `2 minutes` | Interval between server-initiated `PING` messages to connected clients. | +| `MaxPingsOut` | `int` | `2` | Number of outstanding `PING`s without a `PONG` response before the server disconnects a client. | + +### How ServerName is resolved + +`NatsServer` constructs the `ServerInfo` sent to each client at connection time. If `ServerName` is `null`, it uses `nats-dotnet-{Environment.MachineName}`: + +```csharp +_serverInfo = new ServerInfo +{ + ServerId = Guid.NewGuid().ToString("N")[..20].ToUpperInvariant(), + ServerName = options.ServerName ?? $"nats-dotnet-{Environment.MachineName}", + Version = NatsProtocol.Version, + Host = options.Host, + Port = options.Port, + MaxPayload = options.MaxPayload, +}; +``` + +--- + +## CLI Arguments + +`Program.cs` parses command-line arguments before creating `NatsServer`. The three supported flags map directly to `NatsOptions` fields: + +| Flag | Alias | Field | Example | +|------|-------|-------|---------| +| `-p` | `--port` | `Port` | `-p 14222` | +| `-a` | `--addr` | `Host` | `-a 127.0.0.1` | +| `-n` | `--name` | `ServerName` | `-n my-server` | + +```csharp +for (int i = 0; i < args.Length; i++) +{ + switch (args[i]) + { + case "-p" or "--port" when i + 1 < args.Length: + options.Port = int.Parse(args[++i]); + break; + case "-a" or "--addr" when i + 1 < args.Length: + options.Host = args[++i]; + break; + case "-n" or "--name" when i + 1 < args.Length: + options.ServerName = args[++i]; + break; + } +} +``` + +Unrecognized flags are silently ignored. There is no `--help` output. + +--- + +## Protocol Constants + +`NatsProtocol` defines wire-level constants that mirror the Go server's defaults. These values are used by the parser and the `INFO` response: + +| Constant | Value | Description | +|----------|-------|-------------| +| `MaxControlLineSize` | `4096` | Maximum bytes in a protocol control line | +| `MaxPayloadSize` | `1048576` | Maximum message payload in bytes (1 MB) | +| `DefaultPort` | `4222` | Standard NATS client port | +| `Version` | `"0.1.0"` | Server version string sent in `INFO` | +| `ProtoVersion` | `1` | NATS protocol version number sent in `INFO` | + +```csharp +public static class NatsProtocol +{ + public const int MaxControlLineSize = 4096; + public const int MaxPayloadSize = 1024 * 1024; // 1MB + public const int DefaultPort = 4222; + public const string Version = "0.1.0"; + public const int ProtoVersion = 1; +} +``` + +`MaxControlLine` in `NatsOptions` and `MaxControlLineSize` in `NatsProtocol` carry the same value. `NatsOptions.MaxPayload` is used as the per-server runtime limit passed into `NatsParser`; `NatsProtocol.MaxPayloadSize` is the compile-time default. + +--- + +## Logging Configuration + +### Serilog setup + +Logging uses [Serilog](https://serilog.net/) with the console sink, configured in `Program.cs` before any other code runs: + +```csharp +Log.Logger = new LoggerConfiguration() + .MinimumLevel.Debug() + .Enrich.FromLogContext() + .WriteTo.Console(outputTemplate: "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj}{NewLine}{Exception}") + .CreateLogger(); +``` + +The output template produces lines like: + +``` +[14:32:01 INF] Listening on 0.0.0.0:4222 +[14:32:01 DBG] Client 1 connected from 127.0.0.1:54321 +``` + +### ILoggerFactory injection + +`Program.cs` wraps the Serilog logger in a `SerilogLoggerFactory` and passes it to `NatsServer`: + +```csharp +using var loggerFactory = new Serilog.Extensions.Logging.SerilogLoggerFactory(Log.Logger); +var server = new NatsServer(options, loggerFactory); +``` + +`NatsServer` stores the factory and uses it to create two kinds of loggers: + +- **`ILogger`** — obtained via `loggerFactory.CreateLogger()`, used for server-level events (listening, client connect/disconnect). +- **`ILogger` with named category** — obtained via `loggerFactory.CreateLogger($"NATS.Server.NatsClient[{clientId}]")`, created per connection. This gives each client a distinct category in log output so messages can be correlated by client ID without structured log properties. + +```csharp +// In NatsServer.StartAsync — one logger per accepted connection +var clientLogger = _loggerFactory.CreateLogger($"NATS.Server.NatsClient[{clientId}]"); +var client = new NatsClient(clientId, socket, _options, _serverInfo, clientLogger); +``` + +`NatsClient` takes `ILogger` (not `ILogger`) so it can accept the pre-named logger instance rather than deriving its category from its own type. + +### Log flushing + +`Log.CloseAndFlush()` runs in the `finally` block of `Program.cs` after the server stops, ensuring buffered log entries are written before the process exits: + +```csharp +try +{ + await server.StartAsync(cts.Token); +} +catch (OperationCanceledException) { } +finally +{ + Log.CloseAndFlush(); +} +``` + +--- + +## Related Documentation + +- [Operations Overview](../Operations/Overview.md) +- [Server Overview](../Server/Overview.md) + + diff --git a/Documentation/GettingStarted/Architecture.md b/Documentation/GettingStarted/Architecture.md new file mode 100644 index 0000000..6f63132 --- /dev/null +++ b/Documentation/GettingStarted/Architecture.md @@ -0,0 +1,197 @@ +# Architecture + +This document describes the overall architecture of the NATS .NET server — its layers, component responsibilities, message flow, and the mapping from the Go reference implementation. + +## Project Overview + +This project is a port of the [NATS server](https://github.com/nats-io/nats-server) (`golang/nats-server/`) to .NET 10 / C#. The Go source in `golang/nats-server/server/` is the authoritative reference. + +Current scope: base publish-subscribe server with wildcard subject matching and queue groups. Authentication, clustering (routes, gateways, leaf nodes), JetStream, and HTTP monitoring are not yet implemented. + +--- + +## Solution Layout + +``` +NatsDotNet.slnx +src/ + NATS.Server/ # Core server library — no executable entry point + NATS.Server.Host/ # Console application — wires logging, parses CLI args, starts server +tests/ + NATS.Server.Tests/ # xUnit test project — unit and integration tests +``` + +`NATS.Server` depends only on `Microsoft.Extensions.Logging.Abstractions`. All Serilog wiring is in `NATS.Server.Host`. This keeps the core library testable without a console host. + +--- + +## Layers + +The implementation is organized bottom-up: each layer depends only on the layers below it. + +``` +NatsServer (orchestrator: accept loop, client registry, message routing) + └── NatsClient (per-connection: I/O pipeline, command dispatch, subscription tracking) + └── NatsParser (protocol state machine: bytes → ParsedCommand) + └── SubList (trie: subject → matching subscriptions) + └── SubjectMatch (validation and wildcard matching primitives) +``` + +### SubjectMatch and SubList + +`SubjectMatch` (`Subscriptions/SubjectMatch.cs`) provides the primitive operations used throughout the subscription system: + +- `IsValidSubject(string)` — rejects empty tokens, whitespace, tokens after `>` +- `IsLiteral(string)` — returns `false` if the subject contains a bare `*` or `>` wildcard token +- `IsValidPublishSubject(string)` — combines both; publish subjects must be literal +- `MatchLiteral(string literal, string pattern)` — token-by-token matching for cache maintenance + +`SubList` (`Subscriptions/SubList.cs`) is a trie-based subscription store. Each trie level (`TrieLevel`) holds a `Dictionary` for literal tokens plus dedicated `Pwc` and `Fwc` pointers for `*` and `>` wildcards. Each `TrieNode` holds `PlainSubs` (a `HashSet`) and `QueueSubs` (a `Dictionary>` keyed by queue group name). + +`SubList.Match(string subject)` checks an in-memory cache first, then falls back to a recursive trie walk (`MatchLevel`) if there is a cache miss. The result is stored as a `SubListResult` — an immutable snapshot containing `PlainSubs` (array) and `QueueSubs` (jagged array, one sub-array per group). + +The cache holds up to 1,024 entries. When that limit is exceeded, 256 entries are swept. Wildcard subscriptions invalidate all matching cache keys on insert and removal; literal subscriptions invalidate only their own key. + +### NatsParser + +`NatsParser` (`Protocol/NatsParser.cs`) is a stateless-per-invocation parser that operates on `ReadOnlySequence` from `System.IO.Pipelines`. The public entry point is: + +```csharp +public bool TryParse(ref ReadOnlySequence buffer, out ParsedCommand command) +``` + +It reads one control line at a time (up to `NatsProtocol.MaxControlLineSize` = 4,096 bytes), identifies the command by the first two lowercased bytes, then dispatches to a per-command parse method. Commands with payloads (`PUB`, `HPUB`) set `_awaitingPayload = true` and return on the next call via `TryReadPayload` once enough bytes arrive. This handles TCP fragmentation without buffering the entire payload before parsing begins. + +`ParsedCommand` is a `readonly struct` — zero heap allocation for control-only commands (`PING`, `PONG`, `SUB`, `UNSUB`, `CONNECT`). Payload commands allocate a `byte[]` for the payload body. + +Command dispatch in `NatsClient.DispatchCommandAsync` covers: `Connect`, `Ping`/`Pong`, `Sub`, `Unsub`, `Pub`, `HPub`. + +### NatsClient + +`NatsClient` (`NatsClient.cs`) handles a single TCP connection. On `RunAsync`, it sends the initial `INFO` frame and then starts two concurrent tasks: + +```csharp +var fillTask = FillPipeAsync(pipe.Writer, ct); // socket → PipeWriter +var processTask = ProcessCommandsAsync(pipe.Reader, ct); // PipeReader → parser → dispatch +``` + +`FillPipeAsync` reads from the `NetworkStream` into a `PipeWriter` in 4,096-byte chunks. `ProcessCommandsAsync` reads from the `PipeReader`, calls `NatsParser.TryParse` in a loop, and dispatches each `ParsedCommand`. The tasks share a `Pipe` instance from `System.IO.Pipelines`. Either task completing (EOF, cancellation, or error) causes `RunAsync` to return, which triggers cleanup via `Router.RemoveClient(this)`. + +Write serialization uses a `SemaphoreSlim(1,1)` (`_writeLock`). All outbound writes (`SendMessageAsync`, `WriteAsync`) acquire this lock before touching the `NetworkStream`, preventing interleaved writes from concurrent message deliveries. + +Subscription state is a `Dictionary` keyed by SID. This dictionary is accessed only from the single processing task, so no locking is needed. `SUB` inserts into this dictionary and into `SubList`; `UNSUB` either sets `MaxMessages` for auto-unsubscribe or immediately removes from both. + +`NatsClient` exposes two interfaces to `NatsServer`: + +```csharp +public interface IMessageRouter +{ + void ProcessMessage(string subject, string? replyTo, + ReadOnlyMemory headers, ReadOnlyMemory payload, NatsClient sender); + void RemoveClient(NatsClient client); +} + +public interface ISubListAccess +{ + SubList SubList { get; } +} +``` + +`NatsServer` implements both. `NatsClient.Router` is set to the server instance immediately after construction. + +### NatsServer + +`NatsServer` (`NatsServer.cs`) owns the TCP listener, the shared `SubList`, and the client registry. Its `StartAsync` method runs the accept loop: + +```csharp +public async Task StartAsync(CancellationToken ct) +{ + _listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + _listener.Bind(new IPEndPoint( + _options.Host == "0.0.0.0" ? IPAddress.Any : IPAddress.Parse(_options.Host), + _options.Port)); + _listener.Listen(128); + // ... + while (!ct.IsCancellationRequested) + { + var socket = await _listener.AcceptAsync(ct); + // create NatsClient, fire-and-forget RunClientAsync + } +} +``` + +Each accepted connection gets a unique `clientId` (incremented via `Interlocked.Increment`), a scoped logger, and a `NatsClient` instance registered in `_clients` (`ConcurrentDictionary`). `RunClientAsync` is fired as a detached task — the accept loop does not await it. + +Message delivery happens in `ProcessMessage`: + +1. Call `_subList.Match(subject)` to get a `SubListResult`. +2. Iterate `result.PlainSubs` — deliver to each subscriber, skipping the sender if `echo` is false. +3. Iterate `result.QueueSubs` — for each group, use round-robin (modulo `Interlocked.Increment`) to pick one member to receive the message. + +Delivery via `SendMessageAsync` is fire-and-forget (not awaited) to avoid blocking the publishing client's processing task while waiting for slow subscribers. + +--- + +## Message Flow + +``` +Client sends: PUB orders.new 12\r\nhello world\r\n + +1. FillPipeAsync reads bytes from socket → PipeWriter +2. ProcessCommandsAsync reads from PipeReader +3. NatsParser.TryParse parses control line "PUB orders.new 12\r\n" + sets _awaitingPayload = true + on next call: reads payload + \r\n trailer + returns ParsedCommand { Type=Pub, Subject="orders.new", Payload=[...] } +4. DispatchCommandAsync calls ProcessPub(cmd) +5. ProcessPub calls Router.ProcessMessage("orders.new", null, default, payload, this) +6. NatsServer.ProcessMessage + → _subList.Match("orders.new") + returns SubListResult { PlainSubs=[sub1, sub2], QueueSubs=[[sub3, sub4]] } + → DeliverMessage(sub1, ...) → sub1.Client.SendMessageAsync(...) + → DeliverMessage(sub2, ...) → sub2.Client.SendMessageAsync(...) + → round-robin pick from [sub3, sub4], e.g. sub3 + → DeliverMessage(sub3, ...) → sub3.Client.SendMessageAsync(...) +7. SendMessageAsync acquires _writeLock, writes MSG frame to socket +``` + +--- + +## Go Reference Mapping + +| Go source | .NET source | +|-----------|-------------| +| `server/sublist.go` | `src/NATS.Server/Subscriptions/SubList.cs` | +| `server/parser.go` | `src/NATS.Server/Protocol/NatsParser.cs` | +| `server/client.go` | `src/NATS.Server/NatsClient.cs` | +| `server/server.go` | `src/NATS.Server/NatsServer.cs` | +| `server/opts.go` | `src/NATS.Server/NatsOptions.cs` | + +The Go `sublist.go` uses atomic generation counters to invalidate a result cache. The .NET `SubList` uses a different strategy: it maintains the cache under `ReaderWriterLockSlim` and does targeted invalidation at insert/remove time, avoiding the need for generation counters. + +The Go `client.go` uses goroutines for `readLoop` and `writeLoop`. The .NET equivalent uses `async Task` with `System.IO.Pipelines`: `FillPipeAsync` (writer side) and `ProcessCommandsAsync` (reader side). + +--- + +## Key .NET Design Choices + +| Concern | .NET choice | Reason | +|---------|-------------|--------| +| I/O buffering | `System.IO.Pipelines` (`Pipe`, `PipeReader`, `PipeWriter`) | Zero-copy buffer management; backpressure built in | +| SubList thread safety | `ReaderWriterLockSlim` | Multiple concurrent readers (match), exclusive writers (insert/remove) | +| Client registry | `ConcurrentDictionary` | Lock-free concurrent access from accept loop and cleanup tasks | +| Write serialization | `SemaphoreSlim(1,1)` per client | Prevents interleaved MSG frames from concurrent deliveries | +| Concurrency | `async/await` + `Task` | Maps Go goroutines to .NET task-based async; no dedicated threads per connection | +| Protocol constants | `NatsProtocol` static class | Pre-encoded byte arrays (`PongBytes`, `CrLf`, etc.) avoid per-call allocations | + +--- + +## Related Documentation + +- [Setup](./Setup.md) +- [Protocol Overview](../Protocol/Overview.md) +- [Subscriptions Overview](../Subscriptions/Overview.md) +- [Server Overview](../Server/Overview.md) +- [Configuration Overview](../Configuration/Overview.md) + + diff --git a/Documentation/GettingStarted/Setup.md b/Documentation/GettingStarted/Setup.md new file mode 100644 index 0000000..ed35122 --- /dev/null +++ b/Documentation/GettingStarted/Setup.md @@ -0,0 +1,183 @@ +# Setup + +This guide covers prerequisites, building, running, and testing the NATS .NET server. + +## Prerequisites + +### .NET 10 SDK + +The project targets `net10.0` with `LangVersion preview` (configured in `Directory.Build.props`). Install the .NET 10 preview SDK from https://dotnet.microsoft.com/download. + +Verify your installation: + +```bash +dotnet --version +# Expected: 10.0.x +``` + +### Go Toolchain (optional) + +The Go toolchain is only needed if you want to run the reference server from `golang/nats-server/` or execute Go test suites for cross-validation. It is not required for building or testing the .NET port. + +```bash +go version +# Expected: go1.22 or later +``` + +--- + +## Building + +The solution file uses the `.slnx` format (not `.sln`). Pass it explicitly if your tooling requires it. + +```bash +# Build all projects +dotnet build + +# Clean and rebuild +dotnet clean && dotnet build +``` + +The solution contains three projects: + +| Project | Path | +|---------|------| +| `NATS.Server` | `src/NATS.Server/` | +| `NATS.Server.Host` | `src/NATS.Server.Host/` | +| `NATS.Server.Tests` | `tests/NATS.Server.Tests/` | + +All projects share settings from `Directory.Build.props`: `net10.0` target framework, nullable reference types enabled, warnings treated as errors. + +--- + +## Running + +The server executable is `NATS.Server.Host`. With no arguments it binds to `0.0.0.0:4222`. + +```bash +dotnet run --project src/NATS.Server.Host +``` + +### CLI Arguments + +| Flag | Alias | Type | Default | Description | +|------|-------|------|---------|-------------| +| `-p` | `--port` | `int` | `4222` | TCP port to listen on | +| `-a` | `--addr` | `string` | `0.0.0.0` | Bind address | +| `-n` | `--name` | `string` | `nats-dotnet-` | Server name reported in INFO | + +```bash +# Custom port +dotnet run --project src/NATS.Server.Host -- -p 14222 + +# Custom address and name +dotnet run --project src/NATS.Server.Host -- -a 127.0.0.1 -p 4222 -n dev-server +``` + +The `--` separator is required to pass arguments through `dotnet run` to the application. + +Startup log output (Serilog console sink): + +``` +[12:00:00 INF] Listening on 0.0.0.0:4222 +``` + +--- + +## Testing + +```bash +# Run all tests +dotnet test + +# Run tests with verbose output +dotnet test -v normal + +# Run a single test project +dotnet test tests/NATS.Server.Tests + +# Run a specific test by name +dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~SubListTests" +``` + +### Test Stack + +| Package | Version | Purpose | +|---------|---------|---------| +| `xunit` | 2.9.3 | Test framework | +| `xunit.runner.visualstudio` | 3.1.4 | VS/Rider test runner integration | +| `Shouldly` | 4.3.0 | Assertion library | +| `NSubstitute` | 5.3.0 | Mocking | +| `NATS.Client.Core` | 2.7.2 | Official NATS .NET client for integration tests | +| `coverlet.collector` | 6.0.4 | Code coverage | + +Do not use FluentAssertions or Moq — the project uses Shouldly and NSubstitute exclusively. + +### Test Files + +| File | Covers | +|------|--------| +| `ParserTests.cs` | `NatsParser.TryParse` for each command type | +| `SubjectMatchTests.cs` | `SubjectMatch` validation and wildcard matching | +| `SubListTests.cs` | `SubList` trie insert, remove, match, and cache behaviour | +| `ClientTests.cs` | `NatsClient` command dispatch and subscription tracking | +| `ServerTests.cs` | `NatsServer` pub/sub, wildcards, queue groups | +| `IntegrationTests.cs` | End-to-end tests using `NATS.Client.Core` against a live server | + +--- + +## NuGet: Central Package Management + +Package versions are defined centrally in `Directory.Packages.props`. Individual `.csproj` files reference packages without a `Version` attribute: + +```xml + + + +``` + +```xml + + + +``` + +To add a new dependency: add a `` entry in `Directory.Packages.props`, then add a `` (without `Version`) in the relevant `.csproj`. + +--- + +## Logging + +The host configures Serilog via `Microsoft.Extensions.Logging` in `Program.cs`: + +```csharp +Log.Logger = new LoggerConfiguration() + .MinimumLevel.Debug() + .Enrich.FromLogContext() + .WriteTo.Console(outputTemplate: "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj}{NewLine}{Exception}") + .CreateLogger(); + +using var loggerFactory = new Serilog.Extensions.Logging.SerilogLoggerFactory(Log.Logger); +var server = new NatsServer(options, loggerFactory); +``` + +`NatsServer` and `NatsClient` receive `ILoggerFactory` and `ILogger` respectively via constructor injection. The core library (`NATS.Server`) depends only on `Microsoft.Extensions.Logging.Abstractions` — it has no direct Serilog dependency. The Serilog packages are wired in `NATS.Server.Host`. + +Per-client loggers are created with a scoped category name that includes the client ID: + +```csharp +var clientLogger = _loggerFactory.CreateLogger($"NATS.Server.NatsClient[{clientId}]"); +``` + +To adjust log levels at runtime, modify the `LoggerConfiguration` in `Program.cs`. + +--- + +## Related Documentation + +- [Architecture](./Architecture.md) +- [Configuration Overview](../Configuration/Overview.md) +- [Protocol Overview](../Protocol/Overview.md) +- [Server Overview](../Server/Overview.md) + + diff --git a/Documentation/Operations/Overview.md b/Documentation/Operations/Overview.md new file mode 100644 index 0000000..835d41d --- /dev/null +++ b/Documentation/Operations/Overview.md @@ -0,0 +1,193 @@ +# Operations Overview + +This document covers running the server, graceful shutdown, connecting clients, and the test suite. + +--- + +## Running the Server + +The host application is `src/NATS.Server.Host`. Run it with `dotnet run`: + +```bash +# Default (port 4222, bind 0.0.0.0) +dotnet run --project src/NATS.Server.Host + +# Custom port +dotnet run --project src/NATS.Server.Host -- -p 14222 + +# Custom bind address and name +dotnet run --project src/NATS.Server.Host -- -a 127.0.0.1 -n my-server +``` + +Arguments after `--` are passed to the program. The three supported flags are `-p`/`--port`, `-a`/`--addr`, and `-n`/`--name`. See [Configuration Overview](../Configuration/Overview.md) for the full CLI reference. + +On startup, the server logs the address it is listening on: + +``` +[14:32:01 INF] Listening on 0.0.0.0:4222 +``` + +### Full host setup + +`Program.cs` initializes Serilog, parses CLI arguments, starts the server, and handles graceful shutdown: + +```csharp +using NATS.Server; +using Serilog; + +Log.Logger = new LoggerConfiguration() + .MinimumLevel.Debug() + .Enrich.FromLogContext() + .WriteTo.Console(outputTemplate: "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj}{NewLine}{Exception}") + .CreateLogger(); + +var options = new NatsOptions(); + +for (int i = 0; i < args.Length; i++) +{ + switch (args[i]) + { + case "-p" or "--port" when i + 1 < args.Length: + options.Port = int.Parse(args[++i]); + break; + case "-a" or "--addr" when i + 1 < args.Length: + options.Host = args[++i]; + break; + case "-n" or "--name" when i + 1 < args.Length: + options.ServerName = args[++i]; + break; + } +} + +using var loggerFactory = new Serilog.Extensions.Logging.SerilogLoggerFactory(Log.Logger); +var server = new NatsServer(options, loggerFactory); + +var cts = new CancellationTokenSource(); +Console.CancelKeyPress += (_, e) => +{ + e.Cancel = true; + cts.Cancel(); +}; + +try +{ + await server.StartAsync(cts.Token); +} +catch (OperationCanceledException) { } +finally +{ + Log.CloseAndFlush(); +} +``` + +--- + +## Graceful Shutdown + +Pressing Ctrl+C triggers `Console.CancelKeyPress`. The handler sets `e.Cancel = true` — this prevents the process from terminating immediately — and calls `cts.Cancel()` to signal the `CancellationToken` passed to `server.StartAsync`. + +`NatsServer.StartAsync` exits its accept loop on cancellation. In-flight client connections are left to drain naturally. After `StartAsync` returns (via `OperationCanceledException` which is caught), the `finally` block runs `Log.CloseAndFlush()` to ensure all buffered log output is written before the process exits. + +--- + +## Testing + +Run the full test suite with: + +```bash +dotnet test +``` + +The test project is at `tests/NATS.Server.Tests/`. It uses xUnit with Shouldly for assertions. + +### Test summary + +69 tests across 6 test files: + +| File | Tests | Coverage | +|------|-------|----------| +| `SubjectMatchTests.cs` | 33 | Subject validation and wildcard matching | +| `SubListTests.cs` | 12 | Trie insert, remove, match, queue groups, cache | +| `ParserTests.cs` | 14 | All command types, split packets, case insensitivity | +| `ClientTests.cs` | 2 | Socket-level INFO on connect, PING/PONG | +| `ServerTests.cs` | 3 | End-to-end accept, pub/sub, wildcard delivery | +| `IntegrationTests.cs` | 5 | NATS.Client.Core protocol compatibility | + +### Test categories + +**SubjectMatchTests** — 33 `[Theory]` cases verifying `SubjectMatch.IsValidSubject` (16 cases), `SubjectMatch.IsValidPublishSubject` (6 cases), and `SubjectMatch.MatchLiteral` (11 cases). Covers empty strings, leading/trailing dots, embedded spaces, `>` in non-terminal position, and all wildcard combinations. + +**SubListTests** — 12 `[Fact]` tests exercising the `SubList` trie directly: literal insert and match, empty result, `*` wildcard at various token levels, `>` wildcard, root `>`, multiple overlapping subscriptions, remove, queue group grouping, `Count` tracking, and cache invalidation after a wildcard insert. + +**ParserTests** — 14 `async [Fact]` tests that write protocol bytes into a `Pipe` and assert on the resulting `ParsedCommand` list. Covers `PING`, `PONG`, `CONNECT`, `SUB` (with and without queue group), `UNSUB` (with and without `max-messages`), `PUB` (with payload, with reply-to, zero payload), `HPUB` (with header), `INFO`, multiple commands in a single buffer, and case-insensitive parsing. + +**ClientTests** — 2 `async [Fact]` tests using a real loopback socket pair. Verifies that `NatsClient` sends an `INFO` frame immediately on connection, and that it responds `PONG` to a `PING` after `CONNECT`. + +**ServerTests** — 3 `async [Fact]` tests that start `NatsServer` on a random port. Verifies `INFO` on connect, basic pub/sub delivery (`MSG` format), and wildcard subscription matching. + +**IntegrationTests** — 5 `async [Fact]` tests using the official `NATS.Client.Core` v2.7.2 NuGet package. Verifies end-to-end protocol compatibility with a real NATS client library: basic pub/sub, `*` wildcard delivery, `>` wildcard delivery, fan-out to two subscribers, and `PingAsync`. + +Integration tests use `NullLoggerFactory.Instance` for the server so test output is not cluttered with server logs. + +--- + +## Connecting with Standard NATS Clients + +The server speaks the standard NATS wire protocol. Any NATS client library can connect to it. + +```bash +# Using the nats CLI tool +nats sub "test.>" --server nats://127.0.0.1:4222 +nats pub test.hello "world" --server nats://127.0.0.1:4222 +``` + +From .NET, using `NATS.Client.Core`: + +```csharp +var opts = new NatsOpts { Url = "nats://127.0.0.1:4222" }; +await using var conn = new NatsConnection(opts); +await conn.ConnectAsync(); +await conn.PublishAsync("test.hello", "world"); +``` + +--- + +## Go Reference Server + +The Go reference implementation can be built from `golang/nats-server/` for comparison testing: + +```bash +# Build +cd golang/nats-server && go build + +# Run on default port +./nats-server + +# Run on custom port +./nats-server -p 14222 +``` + +The Go server is useful for verifying that the .NET port produces identical protocol output and behaves the same way under edge-case client interactions. + +--- + +## Current Limitations + +The following features present in the Go reference server are not yet ported: + +- Authentication — no username/password, token, NKey, or JWT support +- Clustering — no routes, gateways, or leaf nodes +- JetStream — no persistent streaming, streams, consumers, or RAFT +- Monitoring — no HTTP endpoints (`/varz`, `/connz`, `/healthz`, etc.) +- TLS — all connections are plaintext +- WebSocket — no WebSocket transport + +--- + +## Related Documentation + +- [Configuration Overview](../Configuration/Overview.md) +- [Server Overview](../Server/Overview.md) +- [Protocol Overview](../Protocol/Overview.md) + + diff --git a/Documentation/Protocol/Overview.md b/Documentation/Protocol/Overview.md new file mode 100644 index 0000000..b82ee3e --- /dev/null +++ b/Documentation/Protocol/Overview.md @@ -0,0 +1,151 @@ +# Protocol Overview + +NATS uses a line-oriented, text-based protocol over TCP. All commands are terminated by `\r\n`. This simplicity makes it easy to debug with raw TCP tools and keeps parsing overhead low. + +## Command Reference + +All commands flow either from server to client (S→C) or client to server (C→S). PING and PONG travel in both directions as part of the keepalive mechanism. + +| Command | Direction | Format | +|---------|-----------|--------| +| INFO | S→C | `INFO {json}\r\n` | +| CONNECT | C→S | `CONNECT {json}\r\n` | +| PUB | C→S | `PUB subject [reply] size\r\n[payload]\r\n` | +| HPUB | C→S | `HPUB subject [reply] hdr_size total_size\r\n[headers+payload]\r\n` | +| SUB | C→S | `SUB subject [queue] sid\r\n` | +| UNSUB | C→S | `UNSUB sid [max_msgs]\r\n` | +| MSG | S→C | `MSG subject sid [reply] size\r\n[payload]\r\n` | +| HMSG | S→C | `HMSG subject sid [reply] hdr_size total_size\r\n[headers+payload]\r\n` | +| PING/PONG | Both | `PING\r\n` / `PONG\r\n` | +| +OK/-ERR | S→C | `+OK\r\n` / `-ERR 'msg'\r\n` | + +Arguments in brackets are optional. `sid` is the subscription ID string assigned by the client. Commands with a payload body (PUB, HPUB, MSG, HMSG) use a two-line structure: a control line with sizes, then the raw payload bytes, then a terminating `\r\n`. + +## Connection Handshake + +The handshake is always server-initiated: + +1. Client opens TCP connection. +2. Server immediately sends `INFO {json}\r\n` describing itself. +3. Client sends `CONNECT {json}\r\n` with its options. +4. Normal operation begins (PUB, SUB, MSG, PING/PONG, etc.). + +If `verbose` is enabled in `ClientOptions`, the server sends `+OK` after each valid client command. If the server rejects the CONNECT (bad credentials, unsupported protocol version, etc.) it sends `-ERR 'message'\r\n` and closes the connection. + +## ServerInfo + +The `ServerInfo` JSON payload is sent in the initial INFO message. The `ClientId` and `ClientIp` fields are omitted from JSON when not set. + +```csharp +public sealed class ServerInfo +{ + [JsonPropertyName("server_id")] + public required string ServerId { get; set; } + + [JsonPropertyName("server_name")] + public required string ServerName { get; set; } + + [JsonPropertyName("version")] + public required string Version { get; set; } + + [JsonPropertyName("proto")] + public int Proto { get; set; } = NatsProtocol.ProtoVersion; + + [JsonPropertyName("host")] + public required string Host { get; set; } + + [JsonPropertyName("port")] + public int Port { get; set; } + + [JsonPropertyName("headers")] + public bool Headers { get; set; } = true; + + [JsonPropertyName("max_payload")] + public int MaxPayload { get; set; } = NatsProtocol.MaxPayloadSize; + + [JsonPropertyName("client_id")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingDefault)] + public ulong ClientId { get; set; } + + [JsonPropertyName("client_ip")] + [JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] + public string? ClientIp { get; set; } +} +``` + +`headers` signals that the server supports HPUB/HMSG. `max_payload` advertises the largest message body the server accepts (default 1 MB). `proto` is the protocol version integer; the current value is `1`. + +## ClientOptions + +The `ClientOptions` JSON payload is sent by the client in the CONNECT command. + +```csharp +public sealed class ClientOptions +{ + [JsonPropertyName("verbose")] + public bool Verbose { get; set; } + + [JsonPropertyName("pedantic")] + public bool Pedantic { get; set; } + + [JsonPropertyName("echo")] + public bool Echo { get; set; } = true; + + [JsonPropertyName("name")] + public string? Name { get; set; } + + [JsonPropertyName("lang")] + public string? Lang { get; set; } + + [JsonPropertyName("version")] + public string? Version { get; set; } + + [JsonPropertyName("protocol")] + public int Protocol { get; set; } + + [JsonPropertyName("headers")] + public bool Headers { get; set; } + + [JsonPropertyName("no_responders")] + public bool NoResponders { get; set; } +} +``` + +`echo` defaults to `true`, meaning a client receives its own published messages if it has a matching subscription. Setting `echo` to `false` suppresses that. `no_responders` causes the server to send a status message when a request has no subscribers, rather than letting the client time out. + +## Protocol Constants + +`NatsProtocol` centralises limits and pre-encoded byte arrays to avoid repeated allocations in the hot path. + +```csharp +public static class NatsProtocol +{ + public const int MaxControlLineSize = 4096; + public const int MaxPayloadSize = 1024 * 1024; // 1MB + public const int DefaultPort = 4222; + + // Pre-encoded protocol fragments + public static readonly byte[] CrLf = "\r\n"u8.ToArray(); + public static readonly byte[] PingBytes = "PING\r\n"u8.ToArray(); + public static readonly byte[] PongBytes = "PONG\r\n"u8.ToArray(); + public static readonly byte[] OkBytes = "+OK\r\n"u8.ToArray(); + public static readonly byte[] InfoPrefix = "INFO "u8.ToArray(); + public static readonly byte[] MsgPrefix = "MSG "u8.ToArray(); + public static readonly byte[] HmsgPrefix = "HMSG "u8.ToArray(); + public static readonly byte[] ErrPrefix = "-ERR "u8.ToArray(); +} +``` + +`MaxControlLineSize` (4096 bytes) is the maximum length of a command line before the payload. Any control line that exceeds this limit causes the parser to throw `ProtocolViolationException`. `MaxPayloadSize` (1 MB) is the default limit enforced by the parser; it is configurable per server instance. + +## Go Reference + +The Go implementation of protocol parsing is in `golang/nats-server/server/parser.go`. The .NET implementation follows the same command identification strategy and enforces the same control line and payload size limits. + +## Related Documentation + +- [Parser](Parser.md) +- [Server Overview](../Server/Overview.md) +- [Configuration Overview](../Configuration/Overview.md) + + diff --git a/Documentation/Protocol/Parser.md b/Documentation/Protocol/Parser.md new file mode 100644 index 0000000..d782b19 --- /dev/null +++ b/Documentation/Protocol/Parser.md @@ -0,0 +1,271 @@ +# Parser + +`NatsParser` is a stateful byte-level parser that processes NATS protocol commands from a `ReadOnlySequence` provided by `System.IO.Pipelines`. It is called repeatedly in a read loop until no more complete commands are available in the buffer. + +## Key Types + +### CommandType + +The `CommandType` enum identifies every command the parser can produce: + +```csharp +public enum CommandType +{ + Ping, + Pong, + Connect, + Info, + Pub, + HPub, + Sub, + Unsub, + Ok, + Err, +} +``` + +### ParsedCommand + +`ParsedCommand` is a `readonly struct` that carries the result of a successful parse. Using a struct avoids a heap allocation per command on the fast path. + +```csharp +public readonly struct ParsedCommand +{ + public CommandType Type { get; init; } + public string? Subject { get; init; } + public string? ReplyTo { get; init; } + public string? Queue { get; init; } + public string? Sid { get; init; } + public int MaxMessages { get; init; } + public int HeaderSize { get; init; } + public ReadOnlyMemory Payload { get; init; } + + public static ParsedCommand Simple(CommandType type) => new() { Type = type, MaxMessages = -1 }; +} +``` + +Fields that do not apply to a given command type are left at their default values (`null` for strings, `0` for integers). `MaxMessages` uses `-1` as a sentinel meaning "unset" (relevant for UNSUB with no max). `HeaderSize` is set for HPUB/HMSG; `-1` indicates no headers. `Payload` carries the raw body bytes for PUB/HPUB, and the raw JSON bytes for CONNECT/INFO. + +## TryParse + +`TryParse` is the main entry point. It is called by the read loop after each `PipeReader.ReadAsync` completes. + +```csharp +public bool TryParse(ref ReadOnlySequence buffer, out ParsedCommand command) +``` + +The method returns `true` and advances `buffer` past the consumed bytes when a complete command is available. It returns `false` — leaving `buffer` unchanged — when more data is needed. The caller must call `TryParse` in a loop until it returns `false`, then call `PipeReader.AdvanceTo` to signal how far the buffer was consumed. + +If the parser detects a malformed command it throws `ProtocolViolationException`, which the read loop catches to close the connection. + +## Command Identification + +After locating the `\r\n` control line terminator, the parser lowercase-normalises the first two bytes using a bitwise OR with `0x20` and dispatches on them. This single branch handles both upper- and lowercase input without a string comparison or allocation. + +```csharp +byte b0 = (byte)(lineSpan[0] | 0x20); // lowercase +byte b1 = (byte)(lineSpan[1] | 0x20); + +switch (b0) +{ + case (byte)'p': + if (b1 == (byte)'i') // PING + { + command = ParsedCommand.Simple(CommandType.Ping); + buffer = buffer.Slice(reader.Position); + return true; + } + + if (b1 == (byte)'o') // PONG + { + command = ParsedCommand.Simple(CommandType.Pong); + buffer = buffer.Slice(reader.Position); + return true; + } + + if (b1 == (byte)'u') // PUB + { + return ParsePub(lineSpan, ref buffer, reader.Position, out command); + } + + break; + + case (byte)'h': + if (b1 == (byte)'p') // HPUB + { + return ParseHPub(lineSpan, ref buffer, reader.Position, out command); + } + + break; + + case (byte)'s': + if (b1 == (byte)'u') // SUB + { + command = ParseSub(lineSpan); + buffer = buffer.Slice(reader.Position); + return true; + } + + break; + + case (byte)'u': + if (b1 == (byte)'n') // UNSUB + { + command = ParseUnsub(lineSpan); + buffer = buffer.Slice(reader.Position); + return true; + } + + break; + + case (byte)'c': + if (b1 == (byte)'o') // CONNECT + { + command = ParseConnect(lineSpan); + buffer = buffer.Slice(reader.Position); + return true; + } + + break; + + case (byte)'i': + if (b1 == (byte)'n') // INFO + { + command = ParseInfo(lineSpan); + buffer = buffer.Slice(reader.Position); + return true; + } + + break; + + case (byte)'+': // +OK + command = ParsedCommand.Simple(CommandType.Ok); + buffer = buffer.Slice(reader.Position); + return true; + + case (byte)'-': // -ERR + command = ParsedCommand.Simple(CommandType.Err); + buffer = buffer.Slice(reader.Position); + return true; +} + +throw new ProtocolViolationException("Unknown protocol operation"); +``` + +The two-character pairs are: `p+i` = PING, `p+o` = PONG, `p+u` = PUB, `h+p` = HPUB, `s+u` = SUB, `u+n` = UNSUB, `c+o` = CONNECT, `i+n` = INFO. `+` and `-` are matched on `b0` alone since their second characters are unambiguous. + +## Two-Phase Parsing for PUB and HPUB + +PUB and HPUB require a payload body that follows the control line. The parser handles split reads — where the TCP segment boundary falls inside the payload — through an `_awaitingPayload` state flag. + +**Phase 1 — control line:** The parser reads the control line up to `\r\n`, extracts the subject, optional reply-to, and payload size(s), then stores these in private fields (`_pendingSubject`, `_pendingReplyTo`, `_expectedPayloadSize`, `_pendingHeaderSize`, `_pendingType`) and sets `_awaitingPayload = true`. It then immediately calls `TryReadPayload` to attempt phase 2. + +**Phase 2 — payload read:** `TryReadPayload` checks whether `buffer.Length >= _expectedPayloadSize + 2` (the `+ 2` accounts for the trailing `\r\n`). If enough data is present, the payload bytes are copied to a new `byte[]`, the trailing `\r\n` is verified, the `ParsedCommand` is constructed, and `_awaitingPayload` is reset to `false`. If not enough data is present, `TryReadPayload` returns `false` and `_awaitingPayload` remains `true`. + +On the next call to `TryParse`, the check at the top of the method routes straight to `TryReadPayload` without re-parsing the control line: + +```csharp +if (_awaitingPayload) + return TryReadPayload(ref buffer, out command); +``` + +This means the parser correctly handles payloads that arrive across multiple `PipeReader.ReadAsync` completions without buffering the control line a second time. + +## Zero-Allocation Argument Splitting + +`SplitArgs` splits the argument portion of a control line into token ranges without allocating. The caller `stackalloc`s a `Span` sized to the maximum expected argument count for the command, then passes it to `SplitArgs`: + +```csharp +internal static int SplitArgs(Span data, Span ranges) +{ + int count = 0; + int start = -1; + + for (int i = 0; i < data.Length; i++) + { + byte b = data[i]; + if (b is (byte)' ' or (byte)'\t') + { + if (start >= 0) + { + if (count >= ranges.Length) + throw new ProtocolViolationException("Too many arguments"); + ranges[count++] = start..i; + start = -1; + } + } + else + { + if (start < 0) + start = i; + } + } + + if (start >= 0) + { + if (count >= ranges.Length) + throw new ProtocolViolationException("Too many arguments"); + ranges[count++] = start..data.Length; + } + + return count; +} +``` + +The returned `int` is the number of populated entries in `ranges`. Callers index into the original span using those ranges (e.g. `argsSpan[ranges[0]]`) to extract each token as a sub-span, then decode to `string` with `Encoding.ASCII.GetString`. Consecutive whitespace is collapsed: a new token only begins on a non-whitespace byte after one or more whitespace bytes. + +## Decimal Integer Parsing + +`ParseSize` converts an ASCII decimal integer in a byte span to an `int`. It is used for payload sizes and UNSUB max-message counts. + +```csharp +internal static int ParseSize(Span data) +{ + if (data.Length == 0 || data.Length > 9) + return -1; + int n = 0; + foreach (byte b in data) + { + if (b < (byte)'0' || b > (byte)'9') + return -1; + n = n * 10 + (b - '0'); + } + + return n; +} +``` + +The length cap of 9 digits prevents overflow without a checked-arithmetic check. A return value of `-1` signals a parse failure; callers treat this as a `ProtocolViolationException`. + +## Error Handling + +`ProtocolViolationException` is thrown for all malformed input: + +- Control line exceeds `MaxControlLineSize` (4096 bytes). +- Unknown command bytes. +- Wrong number of arguments for a command. +- Payload size is negative, exceeds `MaxPayloadSize`, or the trailing `\r\n` after the payload is absent. +- `SplitArgs` receives more tokens than the caller's `ranges` span can hold. + +The read loop is responsible for catching `ProtocolViolationException`, sending `-ERR` to the client, and closing the connection. + +## Limits + +| Limit | Value | Source | +|-------|-------|--------| +| Max control line | 4096 bytes | `NatsProtocol.MaxControlLineSize` | +| Max payload (default) | 1 048 576 bytes | `NatsProtocol.MaxPayloadSize` | +| Max size field digits | 9 | `ParseSize` length check | + +The max payload is configurable: `NatsParser` accepts a `maxPayload` constructor argument, which `NatsClient` sets from `NatsOptions`. + +## Go Reference + +The .NET parser is a direct port of the state machine in `golang/nats-server/server/parser.go`. The Go implementation uses the same two-byte command identification technique and the same two-phase control-line/payload split for PUB and HPUB. + +## Related Documentation + +- [Protocol Overview](Overview.md) +- [Server Overview](../Server/Overview.md) + + diff --git a/Documentation/Server/Client.md b/Documentation/Server/Client.md new file mode 100644 index 0000000..4b5ae6d --- /dev/null +++ b/Documentation/Server/Client.md @@ -0,0 +1,375 @@ +# Client Connection Handler + +`NatsClient` manages the full lifecycle of one TCP connection: sending the initial INFO handshake, reading and parsing the incoming byte stream, dispatching protocol commands, and writing outbound messages. One `NatsClient` instance exists per accepted socket. + +## Key Concepts + +### Fields and properties + +```csharp +public sealed class NatsClient : IDisposable +{ + private readonly Socket _socket; + private readonly NetworkStream _stream; + private readonly NatsOptions _options; + private readonly ServerInfo _serverInfo; + private readonly NatsParser _parser; + private readonly SemaphoreSlim _writeLock = new(1, 1); + private readonly Dictionary _subs = new(); + private readonly ILogger _logger; + + public ulong Id { get; } + public ClientOptions? ClientOpts { get; private set; } + public IMessageRouter? Router { get; set; } + public bool ConnectReceived { get; private set; } + + public long InMsgs; + public long OutMsgs; + public long InBytes; + public long OutBytes; + + public IReadOnlyDictionary Subscriptions => _subs; +} +``` + +`_writeLock` is a `SemaphoreSlim(1, 1)` — a binary semaphore that serializes all writes to `_stream`. Without it, concurrent `SendMessageAsync` calls from different publisher threads would interleave bytes on the wire. See [Write serialization](#write-serialization) below. + +`_subs` maps subscription IDs (SIDs) to `Subscription` objects. SIDs are client-assigned strings; `Dictionary` gives O(1) lookup for UNSUB processing. + +The four stat fields (`InMsgs`, `OutMsgs`, `InBytes`, `OutBytes`) are `long` fields accessed via `Interlocked` operations throughout the hot path. They are exposed as public fields rather than properties to allow `Interlocked.Increment` and `Interlocked.Add` directly by reference. + +`Router` is set by `NatsServer` after construction, before `RunAsync` is called. It is typed as `IMessageRouter?` rather than `NatsServer` so that tests can substitute a stub. + +### Constructor + +```csharp +public NatsClient(ulong id, Socket socket, NatsOptions options, ServerInfo serverInfo, ILogger logger) +{ + Id = id; + _socket = socket; + _stream = new NetworkStream(socket, ownsSocket: false); + _options = options; + _serverInfo = serverInfo; + _logger = logger; + _parser = new NatsParser(options.MaxPayload); +} +``` + +`NetworkStream` is created with `ownsSocket: false`. This keeps socket lifetime management in `NatsServer`, which disposes the socket explicitly in `Dispose`. If `ownsSocket` were `true`, disposing the `NetworkStream` would close the socket, potentially racing with the disposal path in `NatsServer`. + +`NatsParser` is constructed with `MaxPayload` from options. The parser enforces this limit: a payload larger than `MaxPayload` causes a `ProtocolViolationException` and terminates the connection. + +## Connection Lifecycle + +### RunAsync + +`RunAsync` is the single entry point for a connection. `NatsServer` calls it as a fire-and-forget task. + +```csharp +public async Task RunAsync(CancellationToken ct) +{ + var pipe = new Pipe(); + try + { + await SendInfoAsync(ct); + + var fillTask = FillPipeAsync(pipe.Writer, ct); + var processTask = ProcessCommandsAsync(pipe.Reader, ct); + + await Task.WhenAny(fillTask, processTask); + } + catch (OperationCanceledException) { } + catch (Exception ex) + { + _logger.LogDebug(ex, "Client {ClientId} connection error", Id); + } + finally + { + Router?.RemoveClient(this); + } +} +``` + +The method: + +1. Sends `INFO {json}\r\n` immediately on connect — required by the NATS protocol before the client sends CONNECT. +2. Creates a `System.IO.Pipelines.Pipe` and starts two concurrent tasks: `FillPipeAsync` reads bytes from the socket into the pipe's write end; `ProcessCommandsAsync` reads from the pipe's read end and dispatches commands. +3. Awaits `Task.WhenAny`. Either task completing signals the connection is done — either the socket closed (fill task returns) or a protocol error caused the process task to throw. +4. In `finally`, calls `Router?.RemoveClient(this)` to clean up subscriptions and remove the client from the server's client dictionary. + +`Router?.RemoveClient(this)` uses a null-conditional because `Router` could be null if the client is used in a test context without a server. + +### FillPipeAsync + +```csharp +private async Task FillPipeAsync(PipeWriter writer, CancellationToken ct) +{ + try + { + while (!ct.IsCancellationRequested) + { + var memory = writer.GetMemory(4096); + int bytesRead = await _stream.ReadAsync(memory, ct); + if (bytesRead == 0) + break; + + writer.Advance(bytesRead); + var result = await writer.FlushAsync(ct); + if (result.IsCompleted) + break; + } + } + finally + { + await writer.CompleteAsync(); + } +} +``` + +`writer.GetMemory(4096)` requests at least 4096 bytes of buffer space from the pipe. The pipe may provide more. `_stream.ReadAsync` fills as many bytes as the OS delivers in one call. `writer.Advance(bytesRead)` commits those bytes. `writer.FlushAsync` makes them available to the reader. + +When `bytesRead` is 0 the socket has closed. `writer.CompleteAsync()` in the `finally` block signals end-of-stream to the reader, which causes `ProcessCommandsAsync` to exit its loop on the next iteration. + +### ProcessCommandsAsync + +```csharp +private async Task ProcessCommandsAsync(PipeReader reader, CancellationToken ct) +{ + try + { + while (!ct.IsCancellationRequested) + { + var result = await reader.ReadAsync(ct); + var buffer = result.Buffer; + + while (_parser.TryParse(ref buffer, out var cmd)) + { + await DispatchCommandAsync(cmd, ct); + } + + reader.AdvanceTo(buffer.Start, buffer.End); + + if (result.IsCompleted) + break; + } + } + finally + { + await reader.CompleteAsync(); + } +} +``` + +`reader.ReadAsync` returns a `ReadResult` containing a `ReadOnlySequence`. The inner `while` loop calls `_parser.TryParse` repeatedly, which slices `buffer` forward past each complete command. When `TryParse` returns `false`, not enough data is available for a complete command. + +`reader.AdvanceTo(buffer.Start, buffer.End)` uses the two-argument form: `buffer.Start` (the consumed position — data before this is discarded) and `buffer.End` (the examined position — the pipe knows to wake this task when more data arrives beyond this point). This is the standard `System.IO.Pipelines` backpressure pattern. + +## Command Dispatch + +`DispatchCommandAsync` switches on the `CommandType` returned by the parser: + +```csharp +private async ValueTask DispatchCommandAsync(ParsedCommand cmd, CancellationToken ct) +{ + switch (cmd.Type) + { + case CommandType.Connect: + ProcessConnect(cmd); + break; + + case CommandType.Ping: + await WriteAsync(NatsProtocol.PongBytes, ct); + break; + + case CommandType.Pong: + // Update RTT tracking (placeholder) + break; + + case CommandType.Sub: + ProcessSub(cmd); + break; + + case CommandType.Unsub: + ProcessUnsub(cmd); + break; + + case CommandType.Pub: + case CommandType.HPub: + ProcessPub(cmd); + break; + } +} +``` + +### CONNECT + +`ProcessConnect` deserializes the JSON payload into a `ClientOptions` record and sets `ConnectReceived = true`. `ClientOptions` carries the `echo` flag (default `true`), the client name, language, and version strings. + +### PING / PONG + +PING is responded to immediately with the pre-allocated `NatsProtocol.PongBytes` (`"PONG\r\n"`). The response goes through `WriteAsync`, which acquires the write lock. PONG handling is currently a placeholder for future RTT tracking. + +### SUB + +```csharp +private void ProcessSub(ParsedCommand cmd) +{ + var sub = new Subscription + { + Subject = cmd.Subject!, + Queue = cmd.Queue, + Sid = cmd.Sid!, + }; + + _subs[cmd.Sid!] = sub; + sub.Client = this; + + if (Router is ISubListAccess sl) + sl.SubList.Insert(sub); +} +``` + +A `Subscription` is stored in `_subs` (keyed by SID) and inserted into the shared `SubList` trie. The `Client` back-reference on `Subscription` is set to `this` so that `NatsServer.ProcessMessage` can reach the client from the subscription without a separate lookup. + +`Router is ISubListAccess sl` checks the interface at runtime. In production, `Router` is `NatsServer`, which implements both interfaces. In tests using a stub `IMessageRouter` that does not implement `ISubListAccess`, the insert is silently skipped. + +### UNSUB + +```csharp +private void ProcessUnsub(ParsedCommand cmd) +{ + if (!_subs.TryGetValue(cmd.Sid!, out var sub)) + return; + + if (cmd.MaxMessages > 0) + { + sub.MaxMessages = cmd.MaxMessages; + return; + } + + _subs.Remove(cmd.Sid!); + + if (Router is ISubListAccess sl) + sl.SubList.Remove(sub); +} +``` + +UNSUB has two modes: + +- With `max_msgs > 0`: sets `sub.MaxMessages` to limit future deliveries. The subscription stays in the trie and the client's `_subs` dict. `DeliverMessage` in `NatsServer` checks `MessageCount` against `MaxMessages` on each delivery and silently drops messages beyond the limit. +- Without `max_msgs` (or `max_msgs == 0`): removes the subscription immediately from both `_subs` and the `SubList`. + +### PUB and HPUB + +```csharp +private void ProcessPub(ParsedCommand cmd) +{ + Interlocked.Increment(ref InMsgs); + Interlocked.Add(ref InBytes, cmd.Payload.Length); + + ReadOnlyMemory headers = default; + ReadOnlyMemory payload = cmd.Payload; + + if (cmd.Type == CommandType.HPub && cmd.HeaderSize > 0) + { + headers = cmd.Payload[..cmd.HeaderSize]; + payload = cmd.Payload[cmd.HeaderSize..]; + } + + Router?.ProcessMessage(cmd.Subject!, cmd.ReplyTo, headers, payload, this); +} +``` + +Stats are updated before routing. For HPUB, the combined payload from the parser is split into a header slice and a body slice using `cmd.HeaderSize`. Both slices are `ReadOnlyMemory` views over the same backing array — no copy. `Router.ProcessMessage` then delivers to all matching subscribers. + +## Write Serialization + +Multiple concurrent `SendMessageAsync` calls can arrive from different publisher connections at the same time. Without coordination, their writes would interleave on the socket and corrupt the message stream for the receiving client. `_writeLock` prevents this: + +```csharp +public async Task SendMessageAsync(string subject, string sid, string? replyTo, + ReadOnlyMemory headers, ReadOnlyMemory payload, CancellationToken ct) +{ + Interlocked.Increment(ref OutMsgs); + Interlocked.Add(ref OutBytes, payload.Length + headers.Length); + + byte[] line; + if (headers.Length > 0) + { + int totalSize = headers.Length + payload.Length; + line = Encoding.ASCII.GetBytes( + $"HMSG {subject} {sid} {(replyTo != null ? replyTo + " " : "")}{headers.Length} {totalSize}\r\n"); + } + else + { + line = Encoding.ASCII.GetBytes( + $"MSG {subject} {sid} {(replyTo != null ? replyTo + " " : "")}{payload.Length}\r\n"); + } + + await _writeLock.WaitAsync(ct); + try + { + await _stream.WriteAsync(line, ct); + if (headers.Length > 0) + await _stream.WriteAsync(headers, ct); + if (payload.Length > 0) + await _stream.WriteAsync(payload, ct); + await _stream.WriteAsync(NatsProtocol.CrLf, ct); + await _stream.FlushAsync(ct); + } + finally + { + _writeLock.Release(); + } +} +``` + +The control line is constructed before acquiring the lock so the string formatting work happens outside the critical section. Once the lock is held, all writes for one message — control line, optional headers, payload, and trailing `\r\n` — happen atomically from the perspective of other writers. + +Stats (`OutMsgs`, `OutBytes`) are updated before the lock because they are independent of the write ordering constraint. + +## Subscription Cleanup + +`RemoveAllSubscriptions` is called by `NatsServer.RemoveClient` when a connection ends: + +```csharp +public void RemoveAllSubscriptions(SubList subList) +{ + foreach (var sub in _subs.Values) + subList.Remove(sub); + _subs.Clear(); +} +``` + +This removes every subscription this client holds from the shared `SubList` trie, then clears the local dictionary. After this call, no future `ProcessMessage` call will deliver to this client's subscriptions. + +## Dispose + +```csharp +public void Dispose() +{ + _stream.Dispose(); + _socket.Dispose(); + _writeLock.Dispose(); +} +``` + +Disposing `_stream` closes the network stream. Disposing `_socket` closes the OS socket. Any in-flight `ReadAsync` or `WriteAsync` will fault with an `ObjectDisposedException` or `IOException`, which causes the read/write tasks to terminate. `_writeLock` is disposed last to release the `SemaphoreSlim`'s internal handle. + +## Go Reference + +The Go counterpart is `golang/nats-server/server/client.go`. Key differences in the .NET port: + +- Go uses separate goroutines for `readLoop` and `writeLoop`; the .NET port uses `FillPipeAsync` and `ProcessCommandsAsync` as concurrent `Task`s sharing a `Pipe`. +- Go uses dynamic buffer sizing (512 to 65536 bytes) in `readLoop`; the .NET port requests 4096-byte chunks from the `PipeWriter`. +- Go uses a mutex for write serialization (`c.mu`); the .NET port uses `SemaphoreSlim(1,1)` to allow `await`-based waiting without blocking a thread. +- The `System.IO.Pipelines` `Pipe` replaces Go's direct `net.Conn` reads. This separates the I/O pump from command parsing and avoids partial-read handling in the parser itself. + +## Related Documentation + +- [Server Overview](Overview.md) +- [Protocol Parser](../Protocol/Parser.md) +- [SubList Trie](../Subscriptions/SubList.md) +- [Subscriptions Overview](../Subscriptions/Overview.md) + + diff --git a/Documentation/Server/Overview.md b/Documentation/Server/Overview.md new file mode 100644 index 0000000..dd4e37c --- /dev/null +++ b/Documentation/Server/Overview.md @@ -0,0 +1,223 @@ +# Server Overview + +`NatsServer` is the top-level orchestrator: it binds the TCP listener, accepts incoming connections, and routes published messages to matching subscribers. Each connected client is managed by a `NatsClient` instance; `NatsServer` coordinates them through two interfaces that decouple message routing from connection management. + +## Key Concepts + +### Interfaces + +`NatsServer` exposes two interfaces that `NatsClient` depends on: + +```csharp +public interface IMessageRouter +{ + void ProcessMessage(string subject, string? replyTo, ReadOnlyMemory headers, + ReadOnlyMemory payload, NatsClient sender); + void RemoveClient(NatsClient client); +} + +public interface ISubListAccess +{ + SubList SubList { get; } +} +``` + +`IMessageRouter` is the surface `NatsClient` calls when a PUB command arrives. `ISubListAccess` gives `NatsClient` access to the shared `SubList` so it can insert and remove subscriptions directly — without needing a concrete reference to `NatsServer`. Both interfaces are implemented by `NatsServer`, and both are injected into `NatsClient` through the `Router` property after construction. + +Defining them separately makes unit testing straightforward: a test can supply a stub `IMessageRouter` without standing up a real server. + +### Fields and State + +```csharp +public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable +{ + private readonly NatsOptions _options; + private readonly ConcurrentDictionary _clients = new(); + private readonly SubList _subList = new(); + private readonly ServerInfo _serverInfo; + private readonly ILogger _logger; + private readonly ILoggerFactory _loggerFactory; + private Socket? _listener; + private ulong _nextClientId; + + public SubList SubList => _subList; +} +``` + +`_clients` tracks every live connection. `_nextClientId` is incremented with `Interlocked.Increment` for each accepted socket, producing monotonically increasing client IDs without a lock. `_loggerFactory` is retained so per-client loggers can be created at accept time, each tagged with the client ID. + +### Constructor + +The constructor takes `NatsOptions` and `ILoggerFactory`. It builds a `ServerInfo` struct that is sent to every connecting client in the initial INFO message: + +```csharp +public NatsServer(NatsOptions options, ILoggerFactory loggerFactory) +{ + _options = options; + _loggerFactory = loggerFactory; + _logger = loggerFactory.CreateLogger(); + _serverInfo = new ServerInfo + { + ServerId = Guid.NewGuid().ToString("N")[..20].ToUpperInvariant(), + ServerName = options.ServerName ?? $"nats-dotnet-{Environment.MachineName}", + Version = NatsProtocol.Version, + Host = options.Host, + Port = options.Port, + MaxPayload = options.MaxPayload, + }; +} +``` + +The `ServerId` is derived from a GUID — taking the first 20 characters of its `"N"` format (32 hex digits, no hyphens) and uppercasing them. This matches the fixed-length alphanumeric server ID format used by the Go server. + +## Accept Loop + +`StartAsync` binds the socket, enables `SO_REUSEADDR` so the port can be reused immediately after a restart, and enters an async accept loop: + +```csharp +public async Task StartAsync(CancellationToken ct) +{ + _listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + _listener.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); + _listener.Bind(new IPEndPoint( + _options.Host == "0.0.0.0" ? IPAddress.Any : IPAddress.Parse(_options.Host), + _options.Port)); + _listener.Listen(128); + + while (!ct.IsCancellationRequested) + { + var socket = await _listener.AcceptAsync(ct); + var clientId = Interlocked.Increment(ref _nextClientId); + + var clientLogger = _loggerFactory.CreateLogger($"NATS.Server.NatsClient[{clientId}]"); + var client = new NatsClient(clientId, socket, _options, _serverInfo, clientLogger); + client.Router = this; + _clients[clientId] = client; + + _ = RunClientAsync(client, ct); + } +} +``` + +`RunClientAsync` is fire-and-forget (`_ = ...`). The accept loop does not await it, so accepting new connections is not blocked by any single client's I/O. Each client runs concurrently on the thread pool. + +The backlog of 128 passed to `Listen` controls the OS-level queue of unaccepted connections — matching the Go server default. + +## Message Routing + +`ProcessMessage` is called by `NatsClient` for every PUB or HPUB command. It is the hot path: called once per published message. + +```csharp +public void ProcessMessage(string subject, string? replyTo, ReadOnlyMemory headers, + ReadOnlyMemory payload, NatsClient sender) +{ + var result = _subList.Match(subject); + + // Deliver to plain subscribers + foreach (var sub in result.PlainSubs) + { + if (sub.Client == null || sub.Client == sender && !(sender.ClientOpts?.Echo ?? true)) + continue; + + DeliverMessage(sub, subject, replyTo, headers, payload); + } + + // Deliver to one member of each queue group (round-robin) + foreach (var queueGroup in result.QueueSubs) + { + if (queueGroup.Length == 0) continue; + + var idx = Math.Abs((int)Interlocked.Increment(ref sender.OutMsgs)) % queueGroup.Length; + Interlocked.Decrement(ref sender.OutMsgs); + + for (int attempt = 0; attempt < queueGroup.Length; attempt++) + { + var sub = queueGroup[(idx + attempt) % queueGroup.Length]; + if (sub.Client != null && (sub.Client != sender || (sender.ClientOpts?.Echo ?? true))) + { + DeliverMessage(sub, subject, replyTo, headers, payload); + break; + } + } + } +} +``` + +### Plain subscriber delivery + +Each subscription in `result.PlainSubs` receives the message unless: + +- `sub.Client` is null (the subscription was removed concurrently), or +- the subscriber is the sender and the sender has `echo: false` in its CONNECT options. + +The `echo` flag defaults to `true` in `ClientOptions`, so publishers receive their own messages unless they explicitly opt out. + +### Queue group delivery + +Queue groups provide load-balanced fan-out: exactly one member of each group receives each message. The selection uses a round-robin counter derived from `sender.OutMsgs`. An `Interlocked.Increment` picks the starting index; the `Interlocked.Decrement` immediately after undoes the side effect on the stat, since `OutMsgs` will be incremented correctly inside `SendMessageAsync` when the message is actually dispatched. + +The loop walks from the selected index, wrapping around, until it finds an eligible member (non-null client, echo check). This handles stale subscriptions where the client has disconnected but the subscription object has not yet been cleaned up. + +### DeliverMessage and auto-unsub + +```csharp +private static void DeliverMessage(Subscription sub, string subject, string? replyTo, + ReadOnlyMemory headers, ReadOnlyMemory payload) +{ + var client = sub.Client; + if (client == null) return; + + var count = Interlocked.Increment(ref sub.MessageCount); + if (sub.MaxMessages > 0 && count > sub.MaxMessages) + return; + + _ = client.SendMessageAsync(subject, sub.Sid, replyTo, headers, payload, CancellationToken.None); +} +``` + +`MessageCount` is incremented atomically before the send. If it exceeds `MaxMessages` (set by an UNSUB with a message count argument), the message is silently dropped. The subscription itself is not removed here — removal happens when the client processes the count limit through `ProcessUnsub`, or when the client disconnects and `RemoveAllSubscriptions` is called. + +`SendMessageAsync` is again fire-and-forget. Multiple deliveries to different clients happen concurrently. + +## Client Removal + +```csharp +public void RemoveClient(NatsClient client) +{ + _clients.TryRemove(client.Id, out _); + client.RemoveAllSubscriptions(_subList); +} +``` + +`RemoveClient` is called either from `RunClientAsync`'s `finally` block (after a client disconnects or errors) or from `NatsClient.RunAsync`'s own `finally` block. Both paths may call it; `TryRemove` is idempotent, so double-calls are safe. After removal from `_clients`, all subscriptions belonging to that client are purged from the `SubList` trie and its internal cache. + +## Shutdown and Dispose + +```csharp +public void Dispose() +{ + _listener?.Dispose(); + foreach (var client in _clients.Values) + client.Dispose(); + _subList.Dispose(); +} +``` + +Disposing the listener socket causes `AcceptAsync` to throw, which unwinds `StartAsync`. Client sockets are disposed, which closes their `NetworkStream` and causes their read loops to terminate. `SubList.Dispose` releases its `ReaderWriterLockSlim`. + +## Go Reference + +The Go counterpart is `golang/nats-server/server/server.go`. Key differences in the .NET port: + +- Go uses goroutines for the accept loop and per-client read/write loops; the .NET port uses `async`/`await` with `Task`. +- Go uses `sync/atomic` for client ID generation; the .NET port uses `Interlocked.Increment`. +- Go passes the server to clients via the `srv` field on the client struct; the .NET port uses the `IMessageRouter` interface through the `Router` property. + +## Related Documentation + +- [Client Connection Handler](Client.md) +- [SubList Trie](../Subscriptions/SubList.md) +- [Protocol Overview](../Protocol/Overview.md) +- [Configuration](../Configuration/Overview.md) + + diff --git a/Documentation/Subscriptions/Overview.md b/Documentation/Subscriptions/Overview.md new file mode 100644 index 0000000..c69497a --- /dev/null +++ b/Documentation/Subscriptions/Overview.md @@ -0,0 +1,169 @@ +# Subscriptions Overview + +The subscription system maps published subjects to interested subscribers. It consists of four classes: `Subscription` (the subscription model), `SubListResult` (a match result container), `SubjectMatch` (subject validation and wildcard matching), and `SubList` (the trie that routes messages to subscribers). + +Go reference: `golang/nats-server/server/sublist.go` + +--- + +## Subjects + +A NATS subject is a dot-separated string of tokens: `foo.bar.baz`. Each token is a non-empty sequence of non-whitespace characters. The token separator is `.`. + +### Wildcards + +Two wildcard tokens are supported: + +| Token | Name | Matches | +|-------|------|---------| +| `*` | Partial wildcard (Pwc) | Exactly one token at that position | +| `>` | Full wildcard (Fwc) | One or more remaining tokens | + +Rules: +- `foo.*` matches `foo.bar` but not `foo.bar.baz` (only one token after `foo`). +- `foo.>` matches `foo.bar` and `foo.bar.baz` (one or more tokens after `foo`). +- `>` must be the last token in a subject. `foo.>.bar` is invalid. +- Publishers cannot use wildcards — only subscribers can. + +### Queue Groups + +Subscribers with the same queue name form a queue group. When a message matches a queue group, exactly one member of that group receives the message. The server selects a member per published message, which distributes load across the group. + +Queue subscriptions are independent of plain subscriptions. If a subject has both plain subscribers and queue group subscribers, all plain subscribers receive the message and one member from each queue group receives it. + +--- + +## The `Subscription` Class + +A `Subscription` represents one subscriber's interest in a subject. + +```csharp +public sealed class Subscription +{ + public required string Subject { get; init; } + public string? Queue { get; init; } + public required string Sid { get; init; } + public long MessageCount; // Interlocked + public long MaxMessages; // 0 = unlimited + public NatsClient? Client { get; set; } +} +``` + +- `Subject` — the subject pattern, which may contain wildcards. +- `Queue` — the queue group name. `null` means this is a plain subscription. +- `Sid` — the subscription ID string assigned by the client. Used to identify the subscription in `UNSUB` commands. +- `MessageCount` — updated with `Interlocked.Increment` on each delivered message. Compared against `MaxMessages` to implement `UNSUB maxMessages`. +- `MaxMessages` — the auto-unsubscribe limit. `0` means unlimited. +- `Client` — back-reference to the `NatsClient` that owns this subscription. Used during message delivery to write `MSG` frames to the connection. + +--- + +## The `SubListResult` Class + +`SubListResult` is the return type of `SubList.Match()`. It separates plain subscribers from queue group subscribers. + +```csharp +public sealed class SubListResult +{ + public static readonly SubListResult Empty = new([], []); + + public Subscription[] PlainSubs { get; } + public Subscription[][] QueueSubs { get; } // outer = groups, inner = members + + public SubListResult(Subscription[] plainSubs, Subscription[][] queueSubs) + { + PlainSubs = plainSubs; + QueueSubs = queueSubs; + } +} +``` + +- `PlainSubs` — all plain subscribers whose subject pattern matches the published subject. +- `QueueSubs` — one entry per queue group. Each inner array contains all members of that group. The message router picks one member per group. +- `Empty` — a static singleton returned when no subscriptions match. Avoids allocation on the common case of no interest. + +--- + +## The `SubjectMatch` Class + +`SubjectMatch` is a static utility class for subject validation and wildcard pattern matching. It defines the three fundamental constants used throughout the subscription system: + +```csharp +public static class SubjectMatch +{ + public const char Pwc = '*'; // partial wildcard + public const char Fwc = '>'; // full wildcard + public const char Sep = '.'; // token separator + ... +} +``` + +### Validation methods + +**`IsValidSubject(string subject)`** — returns `true` if the subject is a well-formed NATS subject. Rejects: +- Empty strings. +- Empty tokens (consecutive separators, or leading/trailing separators). +- Tokens containing whitespace characters. +- Any token after `>` (full wildcard must be last). + +**`IsLiteral(string subject)`** — returns `true` if the subject contains no wildcards. A character is treated as a wildcard only when it appears as a complete token (preceded and followed by `.` or the string boundary). + +**`IsValidPublishSubject(string subject)`** — returns `true` if the subject is both valid and literal. Publishers must use literal subjects; this is the combined check applied before inserting a message into the routing path. + +### Pattern matching + +**`MatchLiteral(string literal, string pattern)`** — tests whether a literal subject matches a pattern that may contain wildcards. Used by the cache invalidation logic to determine which cached results are affected when a wildcard subscription is added or removed. + +```csharp +public static bool MatchLiteral(string literal, string pattern) +{ + int li = 0, pi = 0; + + while (pi < pattern.Length) + { + // Get next pattern token + int pTokenStart = pi; + while (pi < pattern.Length && pattern[pi] != Sep) + pi++; + int pTokenLen = pi - pTokenStart; + if (pi < pattern.Length) + pi++; // skip separator + + // Full wildcard -- matches everything remaining + if (pTokenLen == 1 && pattern[pTokenStart] == Fwc) + return li < literal.Length; // must have at least one token left + + // Get next literal token + if (li >= literal.Length) + return false; + int lTokenStart = li; + while (li < literal.Length && literal[li] != Sep) + li++; + int lTokenLen = li - lTokenStart; + if (li < literal.Length) + li++; // skip separator + + // Partial wildcard -- matches any single token + if (pTokenLen == 1 && pattern[pTokenStart] == Pwc) + continue; + + // Literal comparison + if (pTokenLen != lTokenLen) + return false; + if (string.Compare(literal, lTokenStart, pattern, pTokenStart, pTokenLen, StringComparison.Ordinal) != 0) + return false; + } + + return li >= literal.Length; // both exhausted +} +``` + +The algorithm walks both strings token by token without allocating. `>` short-circuits immediately, returning `true` as long as at least one literal token remains. `*` skips the corresponding literal token without comparison. Literal tokens are compared with `StringComparison.Ordinal`. + +--- + +## Related Documentation + +- [SubList Trie](../Subscriptions/SubList.md) + + diff --git a/Documentation/Subscriptions/SubList.md b/Documentation/Subscriptions/SubList.md new file mode 100644 index 0000000..f035c30 --- /dev/null +++ b/Documentation/Subscriptions/SubList.md @@ -0,0 +1,242 @@ +# SubList + +`SubList` is the subscription routing trie. Every published message triggers a `Match()` call to find all interested subscribers. `SubList` stores subscriptions indexed by their subject tokens and returns a `SubListResult` containing both plain subscribers and queue groups. + +Go reference: `golang/nats-server/server/sublist.go` + +--- + +## Thread Safety + +`SubList` uses a `ReaderWriterLockSlim` (`_lock`) with the following locking discipline: + +| Operation | Lock | +|-----------|------| +| `Count` read | Read lock | +| `Match()` — cache hit | Read lock only | +| `Match()` — cache miss | Write lock (to update cache) | +| `Insert()` | Write lock | +| `Remove()` | Write lock | + +Cache misses in `Match()` require a write lock because the cache must be updated after the trie traversal. To avoid a race between the read-lock check and the write-lock update, `Match()` uses double-checked locking: after acquiring the write lock, it checks the cache again before doing trie work. + +--- + +## Trie Structure + +The trie is built from two private classes, `TrieLevel` and `TrieNode`, nested inside `SubList`. + +### `TrieLevel` and `TrieNode` + +```csharp +private sealed class TrieLevel +{ + public readonly Dictionary Nodes = new(StringComparer.Ordinal); + public TrieNode? Pwc; // partial wildcard (*) + public TrieNode? Fwc; // full wildcard (>) +} + +private sealed class TrieNode +{ + public TrieLevel? Next; + public readonly HashSet PlainSubs = []; + public readonly Dictionary> QueueSubs = new(StringComparer.Ordinal); + + public bool IsEmpty => PlainSubs.Count == 0 && QueueSubs.Count == 0 && + (Next == null || (Next.Nodes.Count == 0 && Next.Pwc == null && Next.Fwc == null)); +} +``` + +Each level in the trie represents one token position in a subject. A `TrieLevel` holds: + +- `Nodes` — a dictionary keyed by literal token string, mapping to the `TrieNode` for that token. Uses `StringComparer.Ordinal` for performance. +- `Pwc` — the node for the `*` wildcard at this level, or `null` if no `*` subscriptions exist at this depth. +- `Fwc` — the node for the `>` wildcard at this level, or `null` if no `>` subscriptions exist at this depth. + +A `TrieNode` sits at the boundary between two levels. It holds the subscriptions registered for subjects whose last token leads to this node: + +- `PlainSubs` — a `HashSet` of plain (non-queue) subscribers. +- `QueueSubs` — a dictionary from queue name to the set of members in that queue group. Uses `StringComparer.Ordinal`. +- `Next` — the next `TrieLevel` for deeper token positions. `null` for leaf nodes. +- `IsEmpty` — `true` when the node and all its descendants have no subscriptions. Used during `Remove()` to prune dead branches. + +The trie root is a `TrieLevel` (`_root`) with no parent node. + +### `TokenEnumerator` + +`TokenEnumerator` is a `ref struct` that splits a subject string by `.` without allocating. It operates on a `ReadOnlySpan` derived from the original string. + +```csharp +private ref struct TokenEnumerator +{ + private ReadOnlySpan _remaining; + + public TokenEnumerator(string subject) + { + _remaining = subject.AsSpan(); + Current = default; + } + + public ReadOnlySpan Current { get; private set; } + + public TokenEnumerator GetEnumerator() => this; + + public bool MoveNext() + { + if (_remaining.IsEmpty) + return false; + + int sep = _remaining.IndexOf(SubjectMatch.Sep); + if (sep < 0) + { + Current = _remaining; + _remaining = default; + } + else + { + Current = _remaining[..sep]; + _remaining = _remaining[(sep + 1)..]; + } + return true; + } +} +``` + +`TokenEnumerator` implements the `foreach` pattern directly (via `GetEnumerator()` returning `this`), so it can be used in `foreach` loops without boxing. `Insert()` uses it during trie traversal to avoid string allocations per token. + +--- + +## Insert + +`Insert(Subscription sub)` adds a subscription to the trie under a write lock. + +The method walks the trie one token at a time using `TokenEnumerator`. For each token: +- If the token is `*`, it creates or follows `level.Pwc`. +- If the token is `>`, it creates or follows `level.Fwc` and sets `sawFwc = true` to reject further tokens. +- Otherwise it creates or follows `level.Nodes[token]`. + +At each step, `node.Next` is created if absent, and `level` advances to `node.Next`. + +After all tokens are consumed, the subscription is added to the terminal node: +- Plain subscription: `node.PlainSubs.Add(sub)`. +- Queue subscription: `node.QueueSubs[sub.Queue].Add(sub)`, creating the inner `HashSet` if this is the first member of that group. + +`_count` is incremented and `AddToCache` is called to update any cached results that would now include this subscription. + +--- + +## Remove + +`Remove(Subscription sub)` removes a subscription from the trie under a write lock. + +The method walks the trie along the subscription's subject, recording the path as a `List<(TrieLevel, TrieNode, string token, bool isPwc, bool isFwc)>`. If any node along the path is missing, the method returns without error (the subscription was never inserted). + +After locating the terminal node, the subscription is removed from `PlainSubs` or from the appropriate `QueueSubs` group. If the queue group becomes empty, its entry is removed from the dictionary. + +If removal succeeds: +- `_count` is decremented. +- `RemoveFromCache` is called to invalidate affected cache entries. +- The path list is walked backwards. At each step, if `node.IsEmpty` is `true`, the node is removed from its parent level (`Pwc = null`, `Fwc = null`, or `Nodes.Remove(token)`). This prunes dead branches so the trie does not accumulate empty nodes over time. + +--- + +## Match + +`Match(string subject)` is called for every published message. It returns a `SubListResult` containing all matching plain and queue subscriptions. + +### Cache check and fallback + +```csharp +public SubListResult Match(string subject) +{ + // Check cache under read lock first. + _lock.EnterReadLock(); + try + { + if (_cache != null && _cache.TryGetValue(subject, out var cached)) + return cached; + } + finally + { + _lock.ExitReadLock(); + } + + // Cache miss -- tokenize and match under write lock (needed for cache update). + var tokens = Tokenize(subject); + if (tokens == null) + return SubListResult.Empty; + + _lock.EnterWriteLock(); + try + { + // Re-check cache after acquiring write lock. + if (_cache != null && _cache.TryGetValue(subject, out var cached)) + return cached; + + var plainSubs = new List(); + var queueSubs = new List>(); + + MatchLevel(_root, tokens, 0, plainSubs, queueSubs); + ... + if (_cache != null) + { + _cache[subject] = result; + if (_cache.Count > CacheMax) { /* sweep */ } + } + return result; + } + finally { _lock.ExitWriteLock(); } +} +``` + +On a read-lock cache hit, `Match()` returns immediately with no trie traversal. On a miss, `Tokenize()` splits the subject before acquiring the write lock (subjects with empty tokens return `SubListResult.Empty` immediately). The write lock is then taken and the cache is checked again before invoking `MatchLevel`. + +### `MatchLevel` traversal + +`MatchLevel` is a recursive method that descends the trie matching tokens against the subject array. At each level, for each remaining token position: + +1. If `level.Fwc` is set, all subscriptions from that node are added to the result. The `>` wildcard matches all remaining tokens, so no further recursion is needed for this branch. +2. If `level.Pwc` is set, `MatchLevel` recurses with the next token index and `pwc.Next` as the new level. This handles `*` matching the current token. +3. A literal dictionary lookup on `level.Nodes[tokens[i]]` advances the level pointer for the next iteration. + +After all tokens are consumed, subscriptions from the final literal node and the final `*` position (if present at the last level) are added to the result. The `*` case at the last token requires explicit handling because the loop exits before the recursive call for `*` can execute. + +`AddNodeToResults` flattens a node's `PlainSubs` into the accumulator list and merges its `QueueSubs` groups into the queue accumulator, combining groups by name across multiple matching nodes. + +--- + +## Cache Strategy + +The cache is a `Dictionary` keyed by the literal published subject. All operations use `StringComparer.Ordinal`. + +**Size limits:** The cache holds at most `CacheMax` (1024) entries. When `_cache.Count` exceeds this, a sweep removes entries until the count reaches `CacheSweep` (256). The sweep takes the first `count - 256` keys from the dictionary — no LRU ordering is maintained. + +**`AddToCache`** is called from `Insert()` to keep cached results consistent after adding a subscription: +- For a literal subscription subject, `AddToCache` does a direct lookup. If the exact key is in the cache, it creates a new `SubListResult` with the subscription appended and replaces the cached entry. +- For a wildcard subscription subject, `AddToCache` scans all cached keys and updates any entry whose key is matched by `SubjectMatch.MatchLiteral(key, subject)`. + +**`RemoveFromCache`** is called from `Remove()` to invalidate cached results after removing a subscription: +- For a literal subscription subject, `RemoveFromCache` removes the exact cache key. +- For a wildcard subscription subject, `RemoveFromCache` removes all cached keys matched by the pattern. Because it is difficult to reconstruct the correct result without a full trie traversal, invalidation is preferred over update. + +The asymmetry between `AddToCache` (updates in place) and `RemoveFromCache` (invalidates) avoids a second trie traversal on removal at the cost of a cache miss on the next `Match()` for those keys. + +--- + +## Disposal + +`SubList` implements `IDisposable`. `Dispose()` releases the `ReaderWriterLockSlim`: + +```csharp +public void Dispose() => _lock.Dispose(); +``` + +`SubList` instances are owned by `NatsServer` and disposed during server shutdown. + +--- + +## Related Documentation + +- [Subscriptions Overview](../Subscriptions/Overview.md) + + diff --git a/documentation_rules.md b/documentation_rules.md new file mode 100644 index 0000000..4c03e7a --- /dev/null +++ b/documentation_rules.md @@ -0,0 +1,318 @@ +# Documentation Rules + +This document defines the documentation system for the NATS .NET server project. It provides guidelines for generating, updating, and maintaining project documentation. + +The documentation is intended for internal team reference — explaining what the system is, how it works, how to extend it, and how to debug it. + +## Folder Structure + +``` +Documentation/ +├── Instructions/ # Guidelines for LLMs (meta-documentation) +│ └── (this file serves as the single instructions reference) +│ +├── GettingStarted/ # Onboarding, prerequisites, first run +├── Protocol/ # Wire protocol, parser, command types +├── Subscriptions/ # SubList trie, subject matching, wildcards +├── Server/ # NatsServer orchestrator, NatsClient handler +├── Configuration/ # NatsOptions, appsettings, CLI arguments +├── Operations/ # Deployment, monitoring, health checks, troubleshooting +└── Plans/ # Design documents and implementation plans +``` + +Future module folders (add as modules are ported): + +``` +├── Authentication/ # Auth mechanisms, NKeys, JWT, accounts +├── Clustering/ # Routes, gateways, leaf nodes +├── JetStream/ # Streams, consumers, storage, RAFT +├── Monitoring/ # HTTP endpoints (/varz, /connz, etc.) +├── WebSocket/ # WebSocket transport +└── TLS/ # TLS configuration and setup +``` + +--- + +## Style Guide + +### Tone and Voice + +- **Technical and direct** — no marketing language. Avoid "powerful", "robust", "seamless", "blazing fast". +- **Assume the reader is a .NET developer** — don't explain dependency injection, async/await, or LINQ basics. +- **Explain "why" not just "what"** — document reasoning behind patterns and decisions. +- **Use present tense** — "The parser reads..." not "The parser will read..." + +### Formatting Rules + +| Aspect | Convention | +|--------|------------| +| File names | `PascalCase.md` | +| H1 (`#`) | Document title only, Title Case | +| H2 (`##`) | Major sections, Title Case | +| H3+ (`###`) | Subsections, Sentence case | +| Code blocks | Always specify language (`csharp`, `json`, `bash`, `xml`) | +| Code snippets | 5-25 lines typical; include class/method context | +| Cross-references | Relative paths: `[See SubList](../Subscriptions/SubList.md)` | +| Inline code | Backticks for code refs: `NatsServer`, `SubList.Match()`, `NatsOptions` | +| Lists | Bullets for unordered, numbers for sequential steps | +| Tables | For structured reference (config options, command formats) | + +### Naming Conventions + +- Match code terminology exactly: `SubList` not "Subject List", `NatsClient` not "NATS Client Handler" +- Use backticks for all code references: `NatsParser`, `appsettings.json`, `dotnet test` +- Spell out acronyms on first use: "NATS Adaptive Edge Messaging (NATS)" — common acronyms that don't need expansion: API, JSON, TCP, HTTP, TLS, JWT + +### Code Snippet Guidelines + +**Do:** +- Copy snippets from actual source files +- Include enough context (class name, method signature) +- Specify the language in code blocks +- Show 5-25 line examples + +```csharp +// Good — shows class context +public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable +{ + public async Task StartAsync(CancellationToken ct) + { + _listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + _listener.Bind(new IPEndPoint(IPAddress.Parse(_options.Host), _options.Port)); + _listener.Listen(128); + } +} +``` + +**Don't:** +- Invent example code that doesn't exist in the codebase +- Include 100+ line dumps without explanation +- Use pseudocode when real code is available +- Omit the language specifier on code blocks + +### Structure Conventions + +Every documentation file must include: + +1. **Title and purpose** — H1 heading with 1-2 sentence description +2. **Key concepts** — if the topic requires background understanding +3. **Code examples** — embedded snippets from actual codebase +4. **Configuration** — if the component has configurable options +5. **Related documentation** — links to related topics + +Organize content from general to specific: +1. Overview/introduction +2. Key concepts +3. Basic usage +4. Advanced usage / internals +5. Configuration +6. Troubleshooting +7. Related documentation + +End each document with: +```markdown +## Related Documentation + +- [Related Topic](../Component/Topic.md) +``` + +### What to Avoid + +- Don't document the obvious (e.g., "The constructor creates a new instance") +- Don't duplicate source code comments — reference the file instead +- Don't include temporary information (dates, version numbers, "coming soon") +- Don't over-explain .NET basics + +--- + +## Generating Documentation + +### Document Types + +Each component folder should contain these standard files: + +| File | Purpose | +|------|---------| +| `Overview.md` | What the component does, key concepts, architecture | +| `Development.md` | How to add/modify features, patterns to follow | +| `Configuration.md` | All configurable options with defaults and examples | +| `Troubleshooting.md` | Common issues, error messages, debugging steps | + +Create additional topic-specific files as needed (e.g., `Protocol/Parser.md`, `Subscriptions/SubList.md`). + +### Generation Process + +1. **Identify scope** — which component folder does this belong to? (See Component Map below) +2. **Read source code** — understand the current implementation, identify key classes/methods/patterns, note configuration options +3. **Check existing documentation** — avoid duplication, cross-reference rather than repeat +4. **Write documentation** — follow the style guide, use real code snippets +5. **Verify accuracy** — confirm snippets match source, verify file paths and class names, test commands + +### Creating New Component Folders + +1. Create the folder under `Documentation/` +2. Add at minimum `Overview.md` +3. Add other standard files as content warrants +4. Update the Component Map section below +5. Add cross-references from related documentation + +--- + +## Updating Documentation + +### Update Triggers + +| Code Change | Update These Docs | +|-------------|-------------------| +| New protocol command | `Protocol/` relevant file | +| Parser modified | `Protocol/Parser.md` | +| Subject matching changed | `Subscriptions/SubjectMatch.md` | +| SubList trie modified | `Subscriptions/SubList.md` | +| New subscription type | `Subscriptions/Overview.md` | +| NatsServer changed | `Server/Overview.md` | +| NatsClient changed | `Server/Client.md` | +| Config option added/removed | Component's `Configuration.md` | +| NatsOptions changed | `Configuration/Overview.md` | +| Host startup changed | `Operations/Deployment.md` + `Configuration/` | +| New test patterns | Corresponding component docs | +| Auth mechanism added | `Authentication/` (create if needed) | +| Clustering added | `Clustering/` (create if needed) | + +### Update Process + +1. **Identify affected documentation** — use the Component Map to determine which docs need updating +2. **Read current documentation** — understand existing structure before making changes +3. **Make targeted updates** — only modify sections affected by the code change; don't rewrite unaffected sections +4. **Update code snippets** — if the code change affects documented examples, update them to match +5. **Update cross-references** — add links to newly related docs, remove links to deleted content +6. **Add verification comment** — at the bottom: `` + +### Deletion Handling + +- When code is removed, remove corresponding doc sections +- When code is renamed, update all references (docs, snippets, cross-reference links) +- If an entire feature is removed, delete the doc file and update any index/overview docs +- Search all docs for links to removed content + +### What Not to Update + +- Don't reformat documentation that wasn't affected by the code change +- Don't update examples that still work correctly +- Don't add new content unrelated to the code change +- Don't change writing style in unaffected sections + +--- + +## Component Map + +### Source to Documentation Mapping + +| Source Path | Documentation Folder | +|-------------|---------------------| +| `src/NATS.Server/Protocol/NatsParser.cs` | `Protocol/` | +| `src/NATS.Server/Protocol/NatsProtocol.cs` | `Protocol/` | +| `src/NATS.Server/Subscriptions/SubList.cs` | `Subscriptions/` | +| `src/NATS.Server/Subscriptions/SubjectMatch.cs` | `Subscriptions/` | +| `src/NATS.Server/Subscriptions/Subscription.cs` | `Subscriptions/` | +| `src/NATS.Server/Subscriptions/SubListResult.cs` | `Subscriptions/` | +| `src/NATS.Server/NatsServer.cs` | `Server/` | +| `src/NATS.Server/NatsClient.cs` | `Server/` | +| `src/NATS.Server/NatsOptions.cs` | `Configuration/` | +| `src/NATS.Server.Host/Program.cs` | `Operations/` and `Configuration/` | +| `tests/NATS.Server.Tests/` | Document in corresponding component | +| `golang/nats-server/server/` | Reference material (not documented separately) | + +### Component Details + +#### Protocol/ + +Documents the wire protocol and parser. + +**Source paths:** +- `src/NATS.Server/Protocol/NatsParser.cs` — state machine parser +- `src/NATS.Server/Protocol/NatsProtocol.cs` — constants, ServerInfo, ClientOptions + +**Typical files:** +- `Overview.md` — NATS protocol format, command types, wire format +- `Parser.md` — Parser implementation, `TryParse` flow, state machine +- `Commands.md` — Individual command formats (PUB, SUB, UNSUB, MSG, etc.) + +#### Subscriptions/ + +Documents subject matching and the subscription trie. + +**Source paths:** +- `src/NATS.Server/Subscriptions/SubList.cs` — trie + cache +- `src/NATS.Server/Subscriptions/SubjectMatch.cs` — validation and wildcard matching +- `src/NATS.Server/Subscriptions/Subscription.cs` — subscription model +- `src/NATS.Server/Subscriptions/SubListResult.cs` — match result container + +**Typical files:** +- `Overview.md` — Subject namespace, wildcard rules, queue groups +- `SubList.md` — Trie internals, cache invalidation, thread safety +- `SubjectMatch.md` — Validation rules, wildcard matching algorithm + +#### Server/ + +Documents the server orchestrator and client connection handler. + +**Source paths:** +- `src/NATS.Server/NatsServer.cs` — accept loop, message routing +- `src/NATS.Server/NatsClient.cs` — per-connection read/write, subscription tracking + +**Typical files:** +- `Overview.md` — Server architecture, connection lifecycle, message flow +- `Client.md` — Client connection handling, command dispatch, write serialization +- `MessageRouting.md` — How messages flow from PUB to subscribers + +#### Configuration/ + +Documents server configuration options. + +**Source paths:** +- `src/NATS.Server/NatsOptions.cs` — configuration model +- `src/NATS.Server.Host/Program.cs` — CLI argument parsing, Serilog setup + +**Typical files:** +- `Overview.md` — All options with defaults and descriptions +- `Logging.md` — Serilog configuration, log levels, LogContext usage + +#### Operations/ + +Documents deployment and operational concerns. + +**Source paths:** +- `src/NATS.Server.Host/` — host application + +**Typical files:** +- `Overview.md` — Running the server, CLI arguments +- `Deployment.md` — Deployment procedures +- `Troubleshooting.md` — Common issues and debugging + +#### GettingStarted/ + +Documents onboarding and project overview. + +**Typical files:** +- `Setup.md` — Prerequisites, building, running +- `Architecture.md` — System overview, Go reference mapping +- `Development.md` — Development workflow, testing, contributing + +### Ambiguous Cases + +| Code Type | Document In | +|-----------|-------------| +| Logging setup | `Configuration/Logging.md` | +| Integration tests | `Operations/Testing.md` or corresponding component | +| Shared interfaces (`IMessageRouter`, `ISubListAccess`) | `Server/Overview.md` | +| Go reference code | Don't document separately; reference in `.NET` component docs | + +### Adding New Components + +When a new module is ported (Authentication, Clustering, JetStream, etc.): + +1. Create a new folder under `Documentation/` +2. Add at minimum `Overview.md` +3. Add this mapping table entry +4. Update CLAUDE.md documentation index if it has one +5. Cross-reference from related component docs diff --git a/src/NATS.Server.Host/NATS.Server.Host.csproj b/src/NATS.Server.Host/NATS.Server.Host.csproj index 2424711..0d9510d 100644 --- a/src/NATS.Server.Host/NATS.Server.Host.csproj +++ b/src/NATS.Server.Host/NATS.Server.Host.csproj @@ -8,4 +8,9 @@ + + + + + diff --git a/src/NATS.Server.Host/Program.cs b/src/NATS.Server.Host/Program.cs index ada3a79..b8ffacd 100644 --- a/src/NATS.Server.Host/Program.cs +++ b/src/NATS.Server.Host/Program.cs @@ -1,5 +1,11 @@ -// src/NATS.Server.Host/Program.cs using NATS.Server; +using Serilog; + +Log.Logger = new LoggerConfiguration() + .MinimumLevel.Debug() + .Enrich.FromLogContext() + .WriteTo.Console(outputTemplate: "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj}{NewLine}{Exception}") + .CreateLogger(); var options = new NatsOptions(); @@ -20,7 +26,8 @@ for (int i = 0; i < args.Length; i++) } } -var server = new NatsServer(options); +using var loggerFactory = new Serilog.Extensions.Logging.SerilogLoggerFactory(Log.Logger); +var server = new NatsServer(options, loggerFactory); var cts = new CancellationTokenSource(); Console.CancelKeyPress += (_, e) => @@ -29,12 +36,12 @@ Console.CancelKeyPress += (_, e) => cts.Cancel(); }; -Console.WriteLine($"[NATS] Listening on {options.Host}:{options.Port}"); - try { await server.StartAsync(cts.Token); } catch (OperationCanceledException) { } - -Console.WriteLine("[NATS] Server stopped."); +finally +{ + Log.CloseAndFlush(); +} diff --git a/src/NATS.Server/NATS.Server.csproj b/src/NATS.Server/NATS.Server.csproj index 61718a1..55d61f5 100644 --- a/src/NATS.Server/NATS.Server.csproj +++ b/src/NATS.Server/NATS.Server.csproj @@ -1,3 +1,5 @@ - + + + diff --git a/src/NATS.Server/NatsClient.cs b/src/NATS.Server/NatsClient.cs index 9dfb3db..ebc2e2f 100644 --- a/src/NATS.Server/NatsClient.cs +++ b/src/NATS.Server/NatsClient.cs @@ -3,6 +3,7 @@ using System.IO.Pipelines; using System.Net.Sockets; using System.Text; using System.Text.Json; +using Microsoft.Extensions.Logging; using NATS.Server.Protocol; using NATS.Server.Subscriptions; @@ -29,6 +30,7 @@ public sealed class NatsClient : IDisposable private readonly NatsParser _parser; private readonly SemaphoreSlim _writeLock = new(1, 1); private readonly Dictionary _subs = new(); + private readonly ILogger _logger; public ulong Id { get; } public ClientOptions? ClientOpts { get; private set; } @@ -43,13 +45,14 @@ public sealed class NatsClient : IDisposable public IReadOnlyDictionary Subscriptions => _subs; - public NatsClient(ulong id, Socket socket, NatsOptions options, ServerInfo serverInfo) + public NatsClient(ulong id, Socket socket, NatsOptions options, ServerInfo serverInfo, ILogger logger) { Id = id; _socket = socket; _stream = new NetworkStream(socket, ownsSocket: false); _options = options; _serverInfo = serverInfo; + _logger = logger; _parser = new NatsParser(options.MaxPayload); } @@ -68,7 +71,10 @@ public sealed class NatsClient : IDisposable await Task.WhenAny(fillTask, processTask); } catch (OperationCanceledException) { } - catch (Exception) { /* connection error -- clean up */ } + catch (Exception ex) + { + _logger.LogDebug(ex, "Client {ClientId} connection error", Id); + } finally { Router?.RemoveClient(this); @@ -160,6 +166,7 @@ public sealed class NatsClient : IDisposable ClientOpts = JsonSerializer.Deserialize(cmd.Payload.Span) ?? new ClientOptions(); ConnectReceived = true; + _logger.LogDebug("CONNECT received from client {ClientId}, name={ClientName}", Id, ClientOpts?.Name); } private void ProcessSub(ParsedCommand cmd) @@ -174,12 +181,16 @@ public sealed class NatsClient : IDisposable _subs[cmd.Sid!] = sub; sub.Client = this; + _logger.LogDebug("SUB {Subject} {Sid} from client {ClientId}", cmd.Subject, cmd.Sid, Id); + if (Router is ISubListAccess sl) sl.SubList.Insert(sub); } private void ProcessUnsub(ParsedCommand cmd) { + _logger.LogDebug("UNSUB {Sid} from client {ClientId}", cmd.Sid, Id); + if (!_subs.TryGetValue(cmd.Sid!, out var sub)) return; diff --git a/src/NATS.Server/NatsServer.cs b/src/NATS.Server/NatsServer.cs index e2fd343..fafd641 100644 --- a/src/NATS.Server/NatsServer.cs +++ b/src/NATS.Server/NatsServer.cs @@ -1,6 +1,7 @@ using System.Collections.Concurrent; using System.Net; using System.Net.Sockets; +using Microsoft.Extensions.Logging; using NATS.Server.Protocol; using NATS.Server.Subscriptions; @@ -12,14 +13,18 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable private readonly ConcurrentDictionary _clients = new(); private readonly SubList _subList = new(); private readonly ServerInfo _serverInfo; + private readonly ILogger _logger; + private readonly ILoggerFactory _loggerFactory; private Socket? _listener; private ulong _nextClientId; public SubList SubList => _subList; - public NatsServer(NatsOptions options) + public NatsServer(NatsOptions options, ILoggerFactory loggerFactory) { _options = options; + _loggerFactory = loggerFactory; + _logger = loggerFactory.CreateLogger(); _serverInfo = new ServerInfo { ServerId = Guid.NewGuid().ToString("N")[..20].ToUpperInvariant(), @@ -40,6 +45,8 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable _options.Port)); _listener.Listen(128); + _logger.LogInformation("Listening on {Host}:{Port}", _options.Host, _options.Port); + try { while (!ct.IsCancellationRequested) @@ -47,7 +54,10 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable var socket = await _listener.AcceptAsync(ct); var clientId = Interlocked.Increment(ref _nextClientId); - var client = new NatsClient(clientId, socket, _options, _serverInfo); + _logger.LogDebug("Client {ClientId} connected from {RemoteEndpoint}", clientId, socket.RemoteEndPoint); + + var clientLogger = _loggerFactory.CreateLogger($"NATS.Server.NatsClient[{clientId}]"); + var client = new NatsClient(clientId, socket, _options, _serverInfo, clientLogger); client.Router = this; _clients[clientId] = client; @@ -69,6 +79,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable } finally { + _logger.LogDebug("Client {ClientId} disconnected", client.Id); RemoveClient(client); } } @@ -127,6 +138,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable public void RemoveClient(NatsClient client) { _clients.TryRemove(client.Id, out _); + _logger.LogDebug("Removed client {ClientId}", client.Id); client.RemoveAllSubscriptions(_subList); } diff --git a/tests/NATS.Server.Tests/ClientTests.cs b/tests/NATS.Server.Tests/ClientTests.cs index 67a1a08..9c5540b 100644 --- a/tests/NATS.Server.Tests/ClientTests.cs +++ b/tests/NATS.Server.Tests/ClientTests.cs @@ -3,6 +3,8 @@ using System.Net; using System.Net.Sockets; using System.Text; using System.Text.Json; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using NATS.Server; using NATS.Server.Protocol; @@ -37,7 +39,7 @@ public class ClientTests : IAsyncDisposable Port = 4222, }; - _natsClient = new NatsClient(1, _serverSocket, new NatsOptions(), serverInfo); + _natsClient = new NatsClient(1, _serverSocket, new NatsOptions(), serverInfo, NullLogger.Instance); } public async ValueTask DisposeAsync() @@ -57,9 +59,9 @@ public class ClientTests : IAsyncDisposable var n = await _clientSocket.ReceiveAsync(buf, SocketFlags.None); var response = Encoding.ASCII.GetString(buf, 0, n); - Assert.StartsWith("INFO ", response); - Assert.Contains("server_id", response); - Assert.Contains("\r\n", response); + response.ShouldStartWith("INFO "); + response.ShouldContain("server_id"); + response.ShouldContain("\r\n"); await _cts.CancelAsync(); } @@ -80,7 +82,7 @@ public class ClientTests : IAsyncDisposable var n = await _clientSocket.ReceiveAsync(buf, SocketFlags.None); var response = Encoding.ASCII.GetString(buf, 0, n); - Assert.Contains("PONG\r\n", response); + response.ShouldContain("PONG\r\n"); await _cts.CancelAsync(); } diff --git a/tests/NATS.Server.Tests/IntegrationTests.cs b/tests/NATS.Server.Tests/IntegrationTests.cs index 0c0acfb..debb8fd 100644 --- a/tests/NATS.Server.Tests/IntegrationTests.cs +++ b/tests/NATS.Server.Tests/IntegrationTests.cs @@ -1,6 +1,7 @@ using System.Net; using System.Net.Sockets; using System.Text; +using Microsoft.Extensions.Logging.Abstractions; using NATS.Client.Core; using NATS.Server; @@ -16,7 +17,7 @@ public class IntegrationTests : IAsyncDisposable public IntegrationTests() { _port = GetFreePort(); - _server = new NatsServer(new NatsOptions { Port = _port }); + _server = new NatsServer(new NatsOptions { Port = _port }, NullLoggerFactory.Instance); _serverTask = _server.StartAsync(_cts.Token); Thread.Sleep(200); // Let server start } @@ -65,7 +66,7 @@ public class IntegrationTests : IAsyncDisposable await pub.PublishAsync("test.subject", "Hello NATS!"); var result = await received.Task.WaitAsync(TimeSpan.FromSeconds(5)); - Assert.Equal("Hello NATS!", result); + result.ShouldBe("Hello NATS!"); } [Fact] @@ -92,7 +93,7 @@ public class IntegrationTests : IAsyncDisposable await pub.PublishAsync("test.hello", "data"); var result = await received.Task.WaitAsync(TimeSpan.FromSeconds(5)); - Assert.Equal("test.hello", result); + result.ShouldBe("test.hello"); } [Fact] @@ -119,7 +120,7 @@ public class IntegrationTests : IAsyncDisposable await pub.PublishAsync("test.foo.bar.baz", "data"); var result = await received.Task.WaitAsync(TimeSpan.FromSeconds(5)); - Assert.Equal("test.foo.bar.baz", result); + result.ShouldBe("test.foo.bar.baz"); } [Fact] @@ -163,8 +164,8 @@ public class IntegrationTests : IAsyncDisposable await pub.PublishAsync("fanout", "hello"); await done.Task.WaitAsync(TimeSpan.FromSeconds(5)); - Assert.Equal(1, count1); - Assert.Equal(1, count2); + count1.ShouldBe(1); + count2.ShouldBe(1); } [Fact] diff --git a/tests/NATS.Server.Tests/NATS.Server.Tests.csproj b/tests/NATS.Server.Tests/NATS.Server.Tests.csproj index dba85ed..feb18ff 100644 --- a/tests/NATS.Server.Tests/NATS.Server.Tests.csproj +++ b/tests/NATS.Server.Tests/NATS.Server.Tests.csproj @@ -5,15 +5,18 @@ - - - - - + + + + + + + + diff --git a/tests/NATS.Server.Tests/ParserTests.cs b/tests/NATS.Server.Tests/ParserTests.cs index a108f23..4817cca 100644 --- a/tests/NATS.Server.Tests/ParserTests.cs +++ b/tests/NATS.Server.Tests/ParserTests.cs @@ -40,117 +40,117 @@ public class ParserTests public async Task Parse_PING() { var cmds = await ParseAsync("PING\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Ping, cmds[0].Type); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Ping); } [Fact] public async Task Parse_PONG() { var cmds = await ParseAsync("PONG\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Pong, cmds[0].Type); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Pong); } [Fact] public async Task Parse_CONNECT() { var cmds = await ParseAsync("CONNECT {\"verbose\":false,\"echo\":true}\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Connect, cmds[0].Type); - Assert.Contains("verbose", Encoding.ASCII.GetString(cmds[0].Payload.ToArray())); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Connect); + Encoding.ASCII.GetString(cmds[0].Payload.ToArray()).ShouldContain("verbose"); } [Fact] public async Task Parse_SUB_without_queue() { var cmds = await ParseAsync("SUB foo 1\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Sub, cmds[0].Type); - Assert.Equal("foo", cmds[0].Subject); - Assert.Null(cmds[0].Queue); - Assert.Equal("1", cmds[0].Sid); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Sub); + cmds[0].Subject.ShouldBe("foo"); + cmds[0].Queue.ShouldBeNull(); + cmds[0].Sid.ShouldBe("1"); } [Fact] public async Task Parse_SUB_with_queue() { var cmds = await ParseAsync("SUB foo workers 1\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Sub, cmds[0].Type); - Assert.Equal("foo", cmds[0].Subject); - Assert.Equal("workers", cmds[0].Queue); - Assert.Equal("1", cmds[0].Sid); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Sub); + cmds[0].Subject.ShouldBe("foo"); + cmds[0].Queue.ShouldBe("workers"); + cmds[0].Sid.ShouldBe("1"); } [Fact] public async Task Parse_UNSUB() { var cmds = await ParseAsync("UNSUB 1\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Unsub, cmds[0].Type); - Assert.Equal("1", cmds[0].Sid); - Assert.Equal(-1, cmds[0].MaxMessages); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Unsub); + cmds[0].Sid.ShouldBe("1"); + cmds[0].MaxMessages.ShouldBe(-1); } [Fact] public async Task Parse_UNSUB_with_max() { var cmds = await ParseAsync("UNSUB 1 5\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Unsub, cmds[0].Type); - Assert.Equal("1", cmds[0].Sid); - Assert.Equal(5, cmds[0].MaxMessages); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Unsub); + cmds[0].Sid.ShouldBe("1"); + cmds[0].MaxMessages.ShouldBe(5); } [Fact] public async Task Parse_PUB_with_payload() { var cmds = await ParseAsync("PUB foo 5\r\nHello\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Pub, cmds[0].Type); - Assert.Equal("foo", cmds[0].Subject); - Assert.Null(cmds[0].ReplyTo); - Assert.Equal("Hello", Encoding.ASCII.GetString(cmds[0].Payload.ToArray())); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Pub); + cmds[0].Subject.ShouldBe("foo"); + cmds[0].ReplyTo.ShouldBeNull(); + Encoding.ASCII.GetString(cmds[0].Payload.ToArray()).ShouldBe("Hello"); } [Fact] public async Task Parse_PUB_with_reply() { var cmds = await ParseAsync("PUB foo reply 5\r\nHello\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Pub, cmds[0].Type); - Assert.Equal("foo", cmds[0].Subject); - Assert.Equal("reply", cmds[0].ReplyTo); - Assert.Equal("Hello", Encoding.ASCII.GetString(cmds[0].Payload.ToArray())); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Pub); + cmds[0].Subject.ShouldBe("foo"); + cmds[0].ReplyTo.ShouldBe("reply"); + Encoding.ASCII.GetString(cmds[0].Payload.ToArray()).ShouldBe("Hello"); } [Fact] public async Task Parse_multiple_commands() { var cmds = await ParseAsync("PING\r\nPONG\r\nSUB foo 1\r\n"); - Assert.Equal(3, cmds.Count); - Assert.Equal(CommandType.Ping, cmds[0].Type); - Assert.Equal(CommandType.Pong, cmds[1].Type); - Assert.Equal(CommandType.Sub, cmds[2].Type); + cmds.Count.ShouldBe(3); + cmds[0].Type.ShouldBe(CommandType.Ping); + cmds[1].Type.ShouldBe(CommandType.Pong); + cmds[2].Type.ShouldBe(CommandType.Sub); } [Fact] public async Task Parse_PUB_zero_payload() { var cmds = await ParseAsync("PUB foo 0\r\n\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Pub, cmds[0].Type); - Assert.Empty(cmds[0].Payload.ToArray()); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Pub); + cmds[0].Payload.ToArray().ShouldBeEmpty(); } [Fact] public async Task Parse_case_insensitive() { var cmds = await ParseAsync("ping\r\npub FOO 3\r\nabc\r\n"); - Assert.Equal(2, cmds.Count); - Assert.Equal(CommandType.Ping, cmds[0].Type); - Assert.Equal(CommandType.Pub, cmds[1].Type); + cmds.Count.ShouldBe(2); + cmds[0].Type.ShouldBe(CommandType.Ping); + cmds[1].Type.ShouldBe(CommandType.Pub); } [Fact] @@ -161,17 +161,17 @@ public class ParserTests var payload = "Hello"; var total = header.Length + payload.Length; var cmds = await ParseAsync($"HPUB foo {header.Length} {total}\r\n{header}{payload}\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.HPub, cmds[0].Type); - Assert.Equal("foo", cmds[0].Subject); - Assert.Equal(header.Length, cmds[0].HeaderSize); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.HPub); + cmds[0].Subject.ShouldBe("foo"); + cmds[0].HeaderSize.ShouldBe(header.Length); } [Fact] public async Task Parse_INFO() { var cmds = await ParseAsync("INFO {\"server_id\":\"test\"}\r\n"); - Assert.Single(cmds); - Assert.Equal(CommandType.Info, cmds[0].Type); + cmds.ShouldHaveSingleItem(); + cmds[0].Type.ShouldBe(CommandType.Info); } } diff --git a/tests/NATS.Server.Tests/ServerTests.cs b/tests/NATS.Server.Tests/ServerTests.cs index c1f7036..7dd6560 100644 --- a/tests/NATS.Server.Tests/ServerTests.cs +++ b/tests/NATS.Server.Tests/ServerTests.cs @@ -1,6 +1,7 @@ using System.Net; using System.Net.Sockets; using System.Text; +using Microsoft.Extensions.Logging.Abstractions; using NATS.Server; namespace NATS.Server.Tests; @@ -15,7 +16,7 @@ public class ServerTests : IAsyncDisposable { // Use random port _port = GetFreePort(); - _server = new NatsServer(new NatsOptions { Port = _port }); + _server = new NatsServer(new NatsOptions { Port = _port }, NullLoggerFactory.Instance); } public async ValueTask DisposeAsync() @@ -54,7 +55,7 @@ public class ServerTests : IAsyncDisposable using var client = await ConnectClientAsync(); var response = await ReadLineAsync(client); - Assert.StartsWith("INFO ", response); + response.ShouldStartWith("INFO "); await _cts.CancelAsync(); } @@ -84,7 +85,7 @@ public class ServerTests : IAsyncDisposable var n = await sub.ReceiveAsync(buf, SocketFlags.None); var msg = Encoding.ASCII.GetString(buf, 0, n); - Assert.Contains("MSG foo 1 5\r\nHello\r\n", msg); + msg.ShouldContain("MSG foo 1 5\r\nHello\r\n"); await _cts.CancelAsync(); } @@ -110,7 +111,7 @@ public class ServerTests : IAsyncDisposable var n = await sub.ReceiveAsync(buf, SocketFlags.None); var msg = Encoding.ASCII.GetString(buf, 0, n); - Assert.Contains("MSG foo.bar 1 5\r\n", msg); + msg.ShouldContain("MSG foo.bar 1 5\r\n"); await _cts.CancelAsync(); } } diff --git a/tests/NATS.Server.Tests/SubListTests.cs b/tests/NATS.Server.Tests/SubListTests.cs index e7e7ba7..51d1544 100644 --- a/tests/NATS.Server.Tests/SubListTests.cs +++ b/tests/NATS.Server.Tests/SubListTests.cs @@ -15,9 +15,9 @@ public class SubListTests sl.Insert(sub); var r = sl.Match("foo.bar"); - Assert.Single(r.PlainSubs); - Assert.Same(sub, r.PlainSubs[0]); - Assert.Empty(r.QueueSubs); + r.PlainSubs.ShouldHaveSingleItem(); + r.PlainSubs[0].ShouldBeSameAs(sub); + r.QueueSubs.ShouldBeEmpty(); } [Fact] @@ -27,7 +27,7 @@ public class SubListTests sl.Insert(MakeSub("foo.bar")); var r = sl.Match("foo.baz"); - Assert.Empty(r.PlainSubs); + r.PlainSubs.ShouldBeEmpty(); } [Fact] @@ -37,9 +37,9 @@ public class SubListTests var sub = MakeSub("foo.*"); sl.Insert(sub); - Assert.Single(sl.Match("foo.bar").PlainSubs); - Assert.Single(sl.Match("foo.baz").PlainSubs); - Assert.Empty(sl.Match("foo.bar.baz").PlainSubs); + sl.Match("foo.bar").PlainSubs.ShouldHaveSingleItem(); + sl.Match("foo.baz").PlainSubs.ShouldHaveSingleItem(); + sl.Match("foo.bar.baz").PlainSubs.ShouldBeEmpty(); } [Fact] @@ -49,9 +49,9 @@ public class SubListTests var sub = MakeSub("foo.>"); sl.Insert(sub); - Assert.Single(sl.Match("foo.bar").PlainSubs); - Assert.Single(sl.Match("foo.bar.baz").PlainSubs); - Assert.Empty(sl.Match("foo").PlainSubs); + sl.Match("foo.bar").PlainSubs.ShouldHaveSingleItem(); + sl.Match("foo.bar.baz").PlainSubs.ShouldHaveSingleItem(); + sl.Match("foo").PlainSubs.ShouldBeEmpty(); } [Fact] @@ -60,9 +60,9 @@ public class SubListTests var sl = new SubList(); sl.Insert(MakeSub(">")); - Assert.Single(sl.Match("foo").PlainSubs); - Assert.Single(sl.Match("foo.bar").PlainSubs); - Assert.Single(sl.Match("foo.bar.baz").PlainSubs); + sl.Match("foo").PlainSubs.ShouldHaveSingleItem(); + sl.Match("foo.bar").PlainSubs.ShouldHaveSingleItem(); + sl.Match("foo.bar.baz").PlainSubs.ShouldHaveSingleItem(); } [Fact] @@ -75,7 +75,7 @@ public class SubListTests sl.Insert(MakeSub(">", sid: "4")); var r = sl.Match("foo.bar"); - Assert.Equal(4, r.PlainSubs.Length); + r.PlainSubs.Length.ShouldBe(4); } [Fact] @@ -84,10 +84,10 @@ public class SubListTests var sl = new SubList(); var sub = MakeSub("foo.bar"); sl.Insert(sub); - Assert.Single(sl.Match("foo.bar").PlainSubs); + sl.Match("foo.bar").PlainSubs.ShouldHaveSingleItem(); sl.Remove(sub); - Assert.Empty(sl.Match("foo.bar").PlainSubs); + sl.Match("foo.bar").PlainSubs.ShouldBeEmpty(); } [Fact] @@ -99,19 +99,19 @@ public class SubListTests sl.Insert(MakeSub("foo.bar", queue: "loggers", sid: "3")); var r = sl.Match("foo.bar"); - Assert.Empty(r.PlainSubs); - Assert.Equal(2, r.QueueSubs.Length); // 2 queue groups + r.PlainSubs.ShouldBeEmpty(); + r.QueueSubs.Length.ShouldBe(2); // 2 queue groups } [Fact] public void Count_tracks_subscriptions() { var sl = new SubList(); - Assert.Equal(0u, sl.Count); + sl.Count.ShouldBe(0u); sl.Insert(MakeSub("foo", sid: "1")); sl.Insert(MakeSub("bar", sid: "2")); - Assert.Equal(2u, sl.Count); + sl.Count.ShouldBe(2u); sl.Remove(MakeSub("foo", sid: "1")); // Remove by reference won't work — we need the same instance @@ -123,9 +123,9 @@ public class SubListTests var sl = new SubList(); var sub = MakeSub("foo"); sl.Insert(sub); - Assert.Equal(1u, sl.Count); + sl.Count.ShouldBe(1u); sl.Remove(sub); - Assert.Equal(0u, sl.Count); + sl.Count.ShouldBe(0u); } [Fact] @@ -136,13 +136,13 @@ public class SubListTests // Prime the cache var r1 = sl.Match("foo.bar"); - Assert.Single(r1.PlainSubs); + r1.PlainSubs.ShouldHaveSingleItem(); // Insert a wildcard that matches — cache should be invalidated sl.Insert(MakeSub("foo.*", sid: "2")); var r2 = sl.Match("foo.bar"); - Assert.Equal(2, r2.PlainSubs.Length); + r2.PlainSubs.Length.ShouldBe(2); } [Fact] @@ -154,6 +154,6 @@ public class SubListTests sl.Insert(MakeSub("foo.bar.*", sid: "3")); var r = sl.Match("foo.bar.baz"); - Assert.Equal(3, r.PlainSubs.Length); + r.PlainSubs.Length.ShouldBe(3); } } diff --git a/tests/NATS.Server.Tests/SubjectMatchTests.cs b/tests/NATS.Server.Tests/SubjectMatchTests.cs index 8bc10c2..a48994f 100644 --- a/tests/NATS.Server.Tests/SubjectMatchTests.cs +++ b/tests/NATS.Server.Tests/SubjectMatchTests.cs @@ -23,7 +23,7 @@ public class SubjectMatchTests [InlineData("foo\tbar", false)] // no tabs public void IsValidSubject(string subject, bool expected) { - Assert.Equal(expected, SubjectMatch.IsValidSubject(subject)); + SubjectMatch.IsValidSubject(subject).ShouldBe(expected); } [Theory] @@ -35,7 +35,7 @@ public class SubjectMatchTests [InlineData("*", false)] public void IsValidPublishSubject(string subject, bool expected) { - Assert.Equal(expected, SubjectMatch.IsValidPublishSubject(subject)); + SubjectMatch.IsValidPublishSubject(subject).ShouldBe(expected); } [Theory] @@ -52,6 +52,6 @@ public class SubjectMatchTests [InlineData("foo.bar", "foo.bar.>", false)] public void MatchLiteral(string literal, string pattern, bool expected) { - Assert.Equal(expected, SubjectMatch.MatchLiteral(literal, pattern)); + SubjectMatch.MatchLiteral(literal, pattern).ShouldBe(expected); } }