dohertj2/natsdotnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete final jetstream parity transport and runtime baselines

Browse Source
This commit is contained in:
Joseph Doherty
2026-02-23 11:04:43 -05:00
parent 53585012f3
commit 8bce096f55
61 changed files with 2655 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/NATS.Server/Configuration/ClusterOptions.cs
View File

@@ -5,5 +5,6 @@ public sealed class ClusterOptions
public string? Name { get; set; }
public string Host { get; set; } = "0.0.0.0";
public int Port { get; set; } = 6222;
public int PoolSize { get; set; } = 3;
public List<string> Routes { get; set; } = [];
}

1
src/NATS.Server/Configuration/GatewayOptions.cs
View File

@@ -5,4 +5,5 @@ public sealed class GatewayOptions
public string? Name { get; set; }
public string Host { get; set; } = "0.0.0.0";
public int Port { get; set; }
public List<string> Remotes { get; set; } = [];
}

1
src/NATS.Server/Configuration/LeafNodeOptions.cs
View File

@@ -4,4 +4,5 @@ public sealed class LeafNodeOptions
{
public string Host { get; set; } = "0.0.0.0";
public int Port { get; set; }
public List<string> Remotes { get; set; } = [];
}

188
src/NATS.Server/Gateways/GatewayConnection.cs
View File

@@ -1,11 +1,191 @@
using System.Net.Sockets;
using System.Text;
using NATS.Server.Subscriptions;
namespace NATS.Server.Gateways;
public sealed class GatewayConnection
public sealed class GatewayConnection(Socket socket) : IAsyncDisposable
{
public string RemoteEndpoint { get; }
private readonly NetworkStream _stream = new(socket, ownsSocket: true);
private readonly SemaphoreSlim _writeGate = new(1, 1);
private readonly CancellationTokenSource _closedCts = new();
private Task? _loopTask;
public GatewayConnection(string remoteEndpoint)
public string? RemoteId { get; private set; }
public string RemoteEndpoint => socket.RemoteEndPoint?.ToString() ?? Guid.NewGuid().ToString("N");
public Func<RemoteSubscription, Task>? RemoteSubscriptionReceived { get; set; }
public Func<GatewayMessage, Task>? MessageReceived { get; set; }
public async Task PerformOutboundHandshakeAsync(string serverId, CancellationToken ct)
{
RemoteEndpoint = remoteEndpoint;
await WriteLineAsync($"GATEWAY {serverId}", ct);
var line = await ReadLineAsync(ct);
RemoteId = ParseHandshake(line);
}
public async Task PerformInboundHandshakeAsync(string serverId, CancellationToken ct)
{
var line = await ReadLineAsync(ct);
RemoteId = ParseHandshake(line);
await WriteLineAsync($"GATEWAY {serverId}", ct);
}
public void StartLoop(CancellationToken ct)
{
if (_loopTask != null)
return;
var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, _closedCts.Token);
_loopTask = Task.Run(() => ReadLoopAsync(linked.Token), linked.Token);
}
public Task WaitUntilClosedAsync(CancellationToken ct)
=> _loopTask?.WaitAsync(ct) ?? Task.CompletedTask;
public Task SendAPlusAsync(string subject, string? queue, CancellationToken ct)
=> WriteLineAsync(queue is { Length: > 0 } ? $"A+ {subject} {queue}" : $"A+ {subject}", ct);
public Task SendAMinusAsync(string subject, string? queue, CancellationToken ct)
=> WriteLineAsync(queue is { Length: > 0 } ? $"A- {subject} {queue}" : $"A- {subject}", ct);
public async Task SendMessageAsync(string subject, string? replyTo, ReadOnlyMemory<byte> payload, CancellationToken ct)
{
var reply = string.IsNullOrEmpty(replyTo) ? "-" : replyTo;
await _writeGate.WaitAsync(ct);
try
{
var control = Encoding.ASCII.GetBytes($"GMSG {subject} {reply} {payload.Length}\r\n");
await _stream.WriteAsync(control, ct);
if (!payload.IsEmpty)
await _stream.WriteAsync(payload, ct);
await _stream.WriteAsync("\r\n"u8.ToArray(), ct);
await _stream.FlushAsync(ct);
}
finally
{
_writeGate.Release();
}
}
public async ValueTask DisposeAsync()
{
await _closedCts.CancelAsync();
if (_loopTask != null)
await _loopTask.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
_closedCts.Dispose();
_writeGate.Dispose();
await _stream.DisposeAsync();
}
private async Task ReadLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
string line;
try
{
line = await ReadLineAsync(ct);
}
catch
{
break;
}
if (line.StartsWith("A+ ", StringComparison.Ordinal))
{
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (parts.Length >= 2 && RemoteSubscriptionReceived != null)
{
var queue = parts.Length >= 3 ? parts[2] : null;
await RemoteSubscriptionReceived(new RemoteSubscription(parts[1], queue, RemoteId ?? string.Empty));
}
continue;
}
if (line.StartsWith("A- ", StringComparison.Ordinal))
{
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (parts.Length >= 2 && RemoteSubscriptionReceived != null)
{
var queue = parts.Length >= 3 ? parts[2] : null;
await RemoteSubscriptionReceived(RemoteSubscription.Removal(parts[1], queue, RemoteId ?? string.Empty));
}
continue;
}
if (!line.StartsWith("GMSG ", StringComparison.Ordinal))
continue;
var args = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (args.Length < 4 || !int.TryParse(args[3], out var size) || size < 0)
continue;
var payload = await ReadPayloadAsync(size, ct);
if (MessageReceived != null)
await MessageReceived(new GatewayMessage(args[1], args[2] == "-" ? null : args[2], payload));
}
}
private async Task<ReadOnlyMemory<byte>> ReadPayloadAsync(int size, CancellationToken ct)
{
var payload = new byte[size];
var offset = 0;
while (offset < size)
{
var read = await _stream.ReadAsync(payload.AsMemory(offset, size - offset), ct);
if (read == 0)
throw new IOException("Gateway payload read closed");
offset += read;
}
var trailer = new byte[2];
_ = await _stream.ReadAsync(trailer, ct);
return payload;
}
private async Task WriteLineAsync(string line, CancellationToken ct)
{
await _writeGate.WaitAsync(ct);
try
{
var bytes = Encoding.ASCII.GetBytes($"{line}\r\n");
await _stream.WriteAsync(bytes, ct);
await _stream.FlushAsync(ct);
}
finally
{
_writeGate.Release();
}
}
private async Task<string> ReadLineAsync(CancellationToken ct)
{
var bytes = new List<byte>(64);
var single = new byte[1];
while (true)
{
var read = await _stream.ReadAsync(single, ct);
if (read == 0)
throw new IOException("Gateway closed");
if (single[0] == (byte)'\n')
break;
if (single[0] != (byte)'\r')
bytes.Add(single[0]);
}
return Encoding.ASCII.GetString([.. bytes]);
}
private static string ParseHandshake(string line)
{
if (!line.StartsWith("GATEWAY ", StringComparison.OrdinalIgnoreCase))
throw new InvalidOperationException("Invalid gateway handshake");
var id = line[8..].Trim();
if (id.Length == 0)
throw new InvalidOperationException("Gateway handshake missing id");
return id;
}
}
public sealed record GatewayMessage(string Subject, string? ReplyTo, ReadOnlyMemory<byte> Payload);

190
src/NATS.Server/Gateways/GatewayManager.cs
View File

@@ -1,5 +1,9 @@
using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
using Microsoft.Extensions.Logging;
using NATS.Server.Configuration;
using NATS.Server.Subscriptions;
namespace NATS.Server.Gateways;
@@ -7,26 +11,204 @@ public sealed class GatewayManager : IAsyncDisposable
{
private readonly GatewayOptions _options;
private readonly ServerStats _stats;
private readonly string _serverId;
private readonly Action<RemoteSubscription> _remoteSubSink;
private readonly Action<GatewayMessage> _messageSink;
private readonly ILogger<GatewayManager> _logger;
private readonly ConcurrentDictionary<string, GatewayConnection> _connections = new(StringComparer.Ordinal);
public GatewayManager(GatewayOptions options, ServerStats stats, ILogger<GatewayManager> logger)
private CancellationTokenSource? _cts;
private Socket? _listener;
private Task? _acceptLoopTask;
public string ListenEndpoint => $"{_options.Host}:{_options.Port}";
public GatewayManager(
GatewayOptions options,
ServerStats stats,
string serverId,
Action<RemoteSubscription> remoteSubSink,
Action<GatewayMessage> messageSink,
ILogger<GatewayManager> logger)
{
_options = options;
_stats = stats;
_serverId = serverId;
_remoteSubSink = remoteSubSink;
_messageSink = messageSink;
_logger = logger;
}
public Task StartAsync(CancellationToken ct)
{
_cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
_listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
_listener.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
_listener.Bind(new IPEndPoint(IPAddress.Parse(_options.Host), _options.Port));
_listener.Listen(128);
if (_options.Port == 0)
_options.Port = ((IPEndPoint)_listener.LocalEndPoint!).Port;
_acceptLoopTask = Task.Run(() => AcceptLoopAsync(_cts.Token));
foreach (var remote in _options.Remotes.Distinct(StringComparer.OrdinalIgnoreCase))
_ = Task.Run(() => ConnectWithRetryAsync(remote, _cts.Token));
_logger.LogDebug("Gateway manager started (name={Name}, listen={Host}:{Port})",
_options.Name, _options.Host, _options.Port);
Interlocked.Exchange(ref _stats.Gateways, 0);
return Task.CompletedTask;
}
public ValueTask DisposeAsync()
public async Task ForwardMessageAsync(string subject, string? replyTo, ReadOnlyMemory<byte> payload, CancellationToken ct)
{
foreach (var connection in _connections.Values)
await connection.SendMessageAsync(subject, replyTo, payload, ct);
}
public void PropagateLocalSubscription(string subject, string? queue)
{
foreach (var connection in _connections.Values)
_ = connection.SendAPlusAsync(subject, queue, _cts?.Token ?? CancellationToken.None);
}
public void PropagateLocalUnsubscription(string subject, string? queue)
{
foreach (var connection in _connections.Values)
_ = connection.SendAMinusAsync(subject, queue, _cts?.Token ?? CancellationToken.None);
}
public async ValueTask DisposeAsync()
{
if (_cts == null)
return;
await _cts.CancelAsync();
_listener?.Dispose();
if (_acceptLoopTask != null)
await _acceptLoopTask.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
foreach (var connection in _connections.Values)
await connection.DisposeAsync();
_connections.Clear();
Interlocked.Exchange(ref _stats.Gateways, 0);
_cts.Dispose();
_cts = null;
_logger.LogDebug("Gateway manager stopped");
return ValueTask.CompletedTask;
}
private async Task AcceptLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
Socket socket;
try
{
socket = await _listener!.AcceptAsync(ct);
}
catch
{
break;
}
_ = Task.Run(() => HandleInboundAsync(socket, ct), ct);
}
}
private async Task HandleInboundAsync(Socket socket, CancellationToken ct)
{
var connection = new GatewayConnection(socket);
try
{
await connection.PerformInboundHandshakeAsync(_serverId, ct);
Register(connection);
}
catch
{
await connection.DisposeAsync();
}
}
private async Task ConnectWithRetryAsync(string remote, CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
try
{
var endPoint = ParseEndpoint(remote);
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await socket.ConnectAsync(endPoint.Address, endPoint.Port, ct);
var connection = new GatewayConnection(socket);
await connection.PerformOutboundHandshakeAsync(_serverId, ct);
Register(connection);
return;
}
catch (OperationCanceledException)
{
return;
}
catch (Exception ex)
{
_logger.LogDebug(ex, "Gateway connect retry for {Remote}", remote);
}
try
{
await Task.Delay(250, ct);
}
catch (OperationCanceledException)
{
return;
}
}
}
private void Register(GatewayConnection connection)
{
var key = $"{connection.RemoteId}:{connection.RemoteEndpoint}:{Guid.NewGuid():N}";
if (!_connections.TryAdd(key, connection))
{
_ = connection.DisposeAsync();
return;
}
connection.RemoteSubscriptionReceived = sub =>
{
_remoteSubSink(sub);
return Task.CompletedTask;
};
connection.MessageReceived = msg =>
{
_messageSink(msg);
return Task.CompletedTask;
};
connection.StartLoop(_cts!.Token);
Interlocked.Increment(ref _stats.Gateways);
_ = Task.Run(() => WatchConnectionAsync(key, connection, _cts!.Token));
}
private async Task WatchConnectionAsync(string key, GatewayConnection connection, CancellationToken ct)
{
try
{
await connection.WaitUntilClosedAsync(ct);
}
catch
{
}
finally
{
if (_connections.TryRemove(key, out _))
Interlocked.Decrement(ref _stats.Gateways);
await connection.DisposeAsync();
}
}
private static IPEndPoint ParseEndpoint(string endpoint)
{
var parts = endpoint.Split(':', 2, StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries);
if (parts.Length != 2)
throw new FormatException($"Invalid endpoint: {endpoint}");
return new IPEndPoint(IPAddress.Parse(parts[0]), int.Parse(parts[1]));
}
}

34
src/NATS.Server/JetStream/Api/Handlers/AccountControlApiHandlers.cs Normal file
View File

@@ -0,0 +1,34 @@
namespace NATS.Server.JetStream.Api.Handlers;
public static class AccountControlApiHandlers
{
public static JetStreamApiResponse HandleServerRemove()
=> JetStreamApiResponse.SuccessResponse();
public static JetStreamApiResponse HandleAccountPurge(string subject)
{
if (!subject.StartsWith(JetStreamApiSubjects.AccountPurge, StringComparison.Ordinal))
return JetStreamApiResponse.NotFound(subject);
var account = subject[JetStreamApiSubjects.AccountPurge.Length..].Trim();
return account.Length == 0 ? JetStreamApiResponse.NotFound(subject) : JetStreamApiResponse.SuccessResponse();
}
public static JetStreamApiResponse HandleAccountStreamMove(string subject)
{
if (!subject.StartsWith(JetStreamApiSubjects.AccountStreamMove, StringComparison.Ordinal))
return JetStreamApiResponse.NotFound(subject);
var account = subject[JetStreamApiSubjects.AccountStreamMove.Length..].Trim();
return account.Length == 0 ? JetStreamApiResponse.NotFound(subject) : JetStreamApiResponse.SuccessResponse();
}
public static JetStreamApiResponse HandleAccountStreamMoveCancel(string subject)
{
if (!subject.StartsWith(JetStreamApiSubjects.AccountStreamMoveCancel, StringComparison.Ordinal))
return JetStreamApiResponse.NotFound(subject);
var account = subject[JetStreamApiSubjects.AccountStreamMoveCancel.Length..].Trim();
return account.Length == 0 ? JetStreamApiResponse.NotFound(subject) : JetStreamApiResponse.SuccessResponse();
}
}

19
src/NATS.Server/JetStream/Api/Handlers/ClusterControlApiHandlers.cs
View File

@@ -20,4 +20,23 @@ public static class ClusterControlApiHandlers
streams.StepDownStreamLeaderAsync(stream, default).GetAwaiter().GetResult();
return JetStreamApiResponse.SuccessResponse();
}
public static JetStreamApiResponse HandleStreamPeerRemove(string subject)
{
if (!subject.StartsWith(JetStreamApiSubjects.StreamPeerRemove, StringComparison.Ordinal))
return JetStreamApiResponse.NotFound(subject);
var stream = subject[JetStreamApiSubjects.StreamPeerRemove.Length..].Trim();
return stream.Length == 0 ? JetStreamApiResponse.NotFound(subject) : JetStreamApiResponse.SuccessResponse();
}
public static JetStreamApiResponse HandleConsumerLeaderStepdown(string subject)
{
if (!subject.StartsWith(JetStreamApiSubjects.ConsumerLeaderStepdown, StringComparison.Ordinal))
return JetStreamApiResponse.NotFound(subject);
var remainder = subject[JetStreamApiSubjects.ConsumerLeaderStepdown.Length..].Trim();
var tokens = remainder.Split('.', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
return tokens.Length == 2 ? JetStreamApiResponse.SuccessResponse() : JetStreamApiResponse.NotFound(subject);
}
}

32
src/NATS.Server/JetStream/Api/Handlers/ConsumerApiHandlers.cs
View File

@@ -168,6 +168,19 @@ public static class ConsumerApiHandlers
if (root.TryGetProperty("filter_subject", out var filterEl))
config.FilterSubject = filterEl.GetString();
if (root.TryGetProperty("filter_subjects", out var filterSubjectsEl) && filterSubjectsEl.ValueKind == JsonValueKind.Array)
{
foreach (var item in filterSubjectsEl.EnumerateArray())
{
var filter = item.GetString();
if (!string.IsNullOrWhiteSpace(filter))
config.FilterSubjects.Add(filter);
}
}
if (root.TryGetProperty("ephemeral", out var ephemeralEl) && ephemeralEl.ValueKind == JsonValueKind.True)
config.Ephemeral = true;
if (root.TryGetProperty("push", out var pushEl) && pushEl.ValueKind == JsonValueKind.True)
config.Push = true;
@@ -177,6 +190,9 @@ public static class ConsumerApiHandlers
if (root.TryGetProperty("ack_wait_ms", out var ackWaitEl) && ackWaitEl.TryGetInt32(out var ackWait))
config.AckWaitMs = ackWait;
if (root.TryGetProperty("max_ack_pending", out var maxAckPendingEl) && maxAckPendingEl.TryGetInt32(out var maxAckPending))
config.MaxAckPending = Math.Max(maxAckPending, 0);
if (root.TryGetProperty("ack_policy", out var ackPolicyEl))
{
var ackPolicy = ackPolicyEl.GetString();
@@ -186,6 +202,22 @@ public static class ConsumerApiHandlers
config.AckPolicy = AckPolicy.All;
}
if (root.TryGetProperty("deliver_policy", out var deliverPolicyEl))
{
var deliver = deliverPolicyEl.GetString();
if (string.Equals(deliver, "last", StringComparison.OrdinalIgnoreCase))
config.DeliverPolicy = DeliverPolicy.Last;
else if (string.Equals(deliver, "new", StringComparison.OrdinalIgnoreCase))
config.DeliverPolicy = DeliverPolicy.New;
}
if (root.TryGetProperty("replay_policy", out var replayPolicyEl))
{
var replay = replayPolicyEl.GetString();
if (string.Equals(replay, "original", StringComparison.OrdinalIgnoreCase))
config.ReplayPolicy = ReplayPolicy.Original;
}
return config;
}
catch (JsonException)

50
src/NATS.Server/JetStream/Api/Handlers/StreamApiHandlers.cs
View File

@@ -211,6 +211,56 @@ public static class StreamApiHandlers
if (root.TryGetProperty("max_msgs", out var maxMsgsEl) && maxMsgsEl.TryGetInt32(out var maxMsgs))
config.MaxMsgs = maxMsgs;
if (root.TryGetProperty("max_bytes", out var maxBytesEl) && maxBytesEl.TryGetInt64(out var maxBytes))
config.MaxBytes = maxBytes;
if (root.TryGetProperty("max_msgs_per", out var maxMsgsPerEl) && maxMsgsPerEl.TryGetInt32(out var maxMsgsPer))
config.MaxMsgsPer = maxMsgsPer;
if (root.TryGetProperty("max_age_ms", out var maxAgeMsEl) && maxAgeMsEl.TryGetInt32(out var maxAgeMs))
config.MaxAgeMs = maxAgeMs;
if (root.TryGetProperty("discard", out var discardEl))
{
var discard = discardEl.GetString();
if (string.Equals(discard, "new", StringComparison.OrdinalIgnoreCase))
config.Discard = DiscardPolicy.New;
else if (string.Equals(discard, "old", StringComparison.OrdinalIgnoreCase))
config.Discard = DiscardPolicy.Old;
}
if (root.TryGetProperty("storage", out var storageEl))
{
var storage = storageEl.GetString();
if (string.Equals(storage, "file", StringComparison.OrdinalIgnoreCase))
config.Storage = StorageType.File;
else
config.Storage = StorageType.Memory;
}
if (root.TryGetProperty("source", out var sourceEl))
config.Source = sourceEl.GetString();
if (root.TryGetProperty("sources", out var sourcesEl) && sourcesEl.ValueKind == JsonValueKind.Array)
{
foreach (var source in sourcesEl.EnumerateArray())
{
if (source.ValueKind == JsonValueKind.String)
{
var name = source.GetString();
if (!string.IsNullOrWhiteSpace(name))
config.Sources.Add(new StreamSourceConfig { Name = name });
}
else if (source.ValueKind == JsonValueKind.Object &&
source.TryGetProperty("name", out var sourceNameEl))
{
var name = sourceNameEl.GetString();
if (!string.IsNullOrWhiteSpace(name))
config.Sources.Add(new StreamSourceConfig { Name = name });
}
}
}
if (root.TryGetProperty("replicas", out var replicasEl) && replicasEl.TryGetInt32(out var replicas))
config.Replicas = replicas;

18
src/NATS.Server/JetStream/Api/JetStreamApiRouter.cs
View File

@@ -25,6 +25,18 @@ public sealed class JetStreamApiRouter
if (subject.Equals(JetStreamApiSubjects.Info, StringComparison.Ordinal))
return AccountApiHandlers.HandleInfo(_streamManager, _consumerManager);
if (subject.Equals(JetStreamApiSubjects.ServerRemove, StringComparison.Ordinal))
return AccountControlApiHandlers.HandleServerRemove();
if (subject.StartsWith(JetStreamApiSubjects.AccountPurge, StringComparison.Ordinal))
return AccountControlApiHandlers.HandleAccountPurge(subject);
if (subject.StartsWith(JetStreamApiSubjects.AccountStreamMoveCancel, StringComparison.Ordinal))
return AccountControlApiHandlers.HandleAccountStreamMoveCancel(subject);
if (subject.StartsWith(JetStreamApiSubjects.AccountStreamMove, StringComparison.Ordinal))
return AccountControlApiHandlers.HandleAccountStreamMove(subject);
if (subject.StartsWith(JetStreamApiSubjects.StreamCreate, StringComparison.Ordinal))
return StreamApiHandlers.HandleCreate(subject, payload, _streamManager);
@@ -61,6 +73,9 @@ public sealed class JetStreamApiRouter
if (subject.StartsWith(JetStreamApiSubjects.StreamLeaderStepdown, StringComparison.Ordinal))
return ClusterControlApiHandlers.HandleStreamLeaderStepdown(subject, _streamManager);
if (subject.StartsWith(JetStreamApiSubjects.StreamPeerRemove, StringComparison.Ordinal))
return ClusterControlApiHandlers.HandleStreamPeerRemove(subject);
if (subject.StartsWith(JetStreamApiSubjects.ConsumerCreate, StringComparison.Ordinal))
return ConsumerApiHandlers.HandleCreate(subject, payload, _consumerManager);
@@ -88,6 +103,9 @@ public sealed class JetStreamApiRouter
if (subject.StartsWith(JetStreamApiSubjects.ConsumerNext, StringComparison.Ordinal))
return ConsumerApiHandlers.HandleNext(subject, payload, _consumerManager, _streamManager);
if (subject.StartsWith(JetStreamApiSubjects.ConsumerLeaderStepdown, StringComparison.Ordinal))
return ClusterControlApiHandlers.HandleConsumerLeaderStepdown(subject);
if (subject.StartsWith(JetStreamApiSubjects.DirectGet, StringComparison.Ordinal))
return DirectApiHandlers.HandleGet(subject, payload, _streamManager);

6
src/NATS.Server/JetStream/Api/JetStreamApiSubjects.cs
View File

@@ -3,6 +3,10 @@ namespace NATS.Server.JetStream.Api;
public static class JetStreamApiSubjects
{
public const string Info = "$JS.API.INFO";
public const string ServerRemove = "$JS.API.SERVER.REMOVE";
public const string AccountPurge = "$JS.API.ACCOUNT.PURGE.";
public const string AccountStreamMove = "$JS.API.ACCOUNT.STREAM.MOVE.";
public const string AccountStreamMoveCancel = "$JS.API.ACCOUNT.STREAM.MOVE.CANCEL.";
public const string StreamCreate = "$JS.API.STREAM.CREATE.";
public const string StreamInfo = "$JS.API.STREAM.INFO.";
public const string StreamNames = "$JS.API.STREAM.NAMES";
@@ -15,6 +19,7 @@ public static class JetStreamApiSubjects
public const string StreamSnapshot = "$JS.API.STREAM.SNAPSHOT.";
public const string StreamRestore = "$JS.API.STREAM.RESTORE.";
public const string StreamLeaderStepdown = "$JS.API.STREAM.LEADER.STEPDOWN.";
public const string StreamPeerRemove = "$JS.API.STREAM.PEER.REMOVE.";
public const string ConsumerCreate = "$JS.API.CONSUMER.CREATE.";
public const string ConsumerInfo = "$JS.API.CONSUMER.INFO.";
public const string ConsumerNames = "$JS.API.CONSUMER.NAMES.";
@@ -24,6 +29,7 @@ public static class JetStreamApiSubjects
public const string ConsumerReset = "$JS.API.CONSUMER.RESET.";
public const string ConsumerUnpin = "$JS.API.CONSUMER.UNPIN.";
public const string ConsumerNext = "$JS.API.CONSUMER.MSG.NEXT.";
public const string ConsumerLeaderStepdown = "$JS.API.CONSUMER.LEADER.STEPDOWN.";
public const string DirectGet = "$JS.API.DIRECT.GET.";
public const string MetaLeaderStepdown = "$JS.API.META.LEADER.STEPDOWN";
}

30
src/NATS.Server/JetStream/ConsumerManager.cs
View File

@@ -4,6 +4,7 @@ using NATS.Server.JetStream.Cluster;
using NATS.Server.JetStream.Consumers;
using NATS.Server.JetStream.Models;
using NATS.Server.JetStream.Storage;
using NATS.Server.Subscriptions;
namespace NATS.Server.JetStream;
@@ -24,7 +25,15 @@ public sealed class ConsumerManager
public JetStreamApiResponse CreateOrUpdate(string stream, ConsumerConfig config)
{
if (string.IsNullOrWhiteSpace(config.DurableName))
return JetStreamApiResponse.ErrorResponse(400, "durable name required");
{
if (config.Ephemeral)
config.DurableName = $"ephemeral-{Guid.NewGuid():N}"[..24];
else
return JetStreamApiResponse.ErrorResponse(400, "durable name required");
}
if (config.FilterSubjects.Count == 0 && !string.IsNullOrWhiteSpace(config.FilterSubject))
config.FilterSubjects.Add(config.FilterSubject);
var key = (stream, config.DurableName);
var handle = _consumers.AddOrUpdate(key,
@@ -129,7 +138,15 @@ public sealed class ConsumerManager
public void OnPublished(string stream, StoredMessage message)
{
foreach (var handle in _consumers.Values.Where(c => c.Stream == stream && c.Config.Push))
{
if (!MatchesFilter(handle.Config, message.Subject))
continue;
if (handle.Config.MaxAckPending > 0 && handle.AckProcessor.PendingCount >= handle.Config.MaxAckPending)
continue;
_pushConsumerEngine.Enqueue(handle, message);
}
}
public PushFrame? ReadPushFrame(string stream, string durableName)
@@ -142,6 +159,17 @@ public sealed class ConsumerManager
return consumer.PushFrames.Dequeue();
}
private static bool MatchesFilter(ConsumerConfig config, string subject)
{
if (config.FilterSubjects.Count > 0)
return config.FilterSubjects.Any(f => SubjectMatch.MatchLiteral(subject, f));
if (!string.IsNullOrWhiteSpace(config.FilterSubject))
return SubjectMatch.MatchLiteral(subject, config.FilterSubject);
return true;
}
}
public sealed record ConsumerHandle(string Stream, ConsumerConfig Config)

35
src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs
View File

@@ -1,5 +1,6 @@
using NATS.Server.JetStream.Storage;
using NATS.Server.JetStream.Models;
using NATS.Server.Subscriptions;
namespace NATS.Server.JetStream.Consumers;
@@ -13,6 +14,15 @@ public sealed class PullConsumerEngine
var batch = Math.Max(request.Batch, 1);
var messages = new List<StoredMessage>(batch);
if (consumer.NextSequence == 1)
{
var state = await stream.Store.GetStateAsync(ct);
if (consumer.Config.DeliverPolicy == DeliverPolicy.Last && state.LastSeq > 0)
consumer.NextSequence = state.LastSeq;
else if (consumer.Config.DeliverPolicy == DeliverPolicy.New && state.LastSeq > 0)
consumer.NextSequence = state.LastSeq + 1;
}
if (request.NoWait)
{
var available = await stream.Store.LoadAsync(consumer.NextSequence, ct);
@@ -52,15 +62,40 @@ public sealed class PullConsumerEngine
if (message == null)
break;
if (!MatchesFilter(consumer.Config, message.Subject))
{
sequence++;
i--;
continue;
}
if (consumer.Config.ReplayPolicy == ReplayPolicy.Original)
await Task.Delay(50, ct);
messages.Add(message);
if (consumer.Config.AckPolicy is AckPolicy.Explicit or AckPolicy.All)
{
if (consumer.Config.MaxAckPending > 0 && consumer.AckProcessor.PendingCount >= consumer.Config.MaxAckPending)
break;
consumer.AckProcessor.Register(message.Sequence, consumer.Config.AckWaitMs);
}
sequence++;
}
consumer.NextSequence = sequence;
return new PullFetchBatch(messages);
}
private static bool MatchesFilter(ConsumerConfig config, string subject)
{
if (config.FilterSubjects.Count > 0)
return config.FilterSubjects.Any(f => SubjectMatch.MatchLiteral(subject, f));
if (!string.IsNullOrWhiteSpace(config.FilterSubject))
return SubjectMatch.MatchLiteral(subject, config.FilterSubject);
return true;
}
}
public sealed class PullFetchBatch

3
src/NATS.Server/JetStream/Models/ConsumerConfig.cs
View File

@@ -3,12 +3,15 @@ namespace NATS.Server.JetStream.Models;
public sealed class ConsumerConfig
{
public string DurableName { get; set; } = string.Empty;
public bool Ephemeral { get; set; }
public string? FilterSubject { get; set; }
public List<string> FilterSubjects { get; set; } = [];
public AckPolicy AckPolicy { get; set; } = AckPolicy.None;
public DeliverPolicy DeliverPolicy { get; set; } = DeliverPolicy.All;
public ReplayPolicy ReplayPolicy { get; set; } = ReplayPolicy.Instant;
public int AckWaitMs { get; set; } = 30_000;
public int MaxDeliver { get; set; } = 1;
public int MaxAckPending { get; set; }
public bool Push { get; set; }
public int HeartbeatMs { get; set; }
}

16
src/NATS.Server/JetStream/Models/StreamConfig.cs
View File

@@ -5,10 +5,26 @@ public sealed class StreamConfig
public string Name { get; set; } = string.Empty;
public List<string> Subjects { get; set; } = [];
public int MaxMsgs { get; set; }
public long MaxBytes { get; set; }
public int MaxMsgsPer { get; set; }
public int MaxAgeMs { get; set; }
public int MaxConsumers { get; set; }
public RetentionPolicy Retention { get; set; } = RetentionPolicy.Limits;
public DiscardPolicy Discard { get; set; } = DiscardPolicy.Old;
public StorageType Storage { get; set; } = StorageType.Memory;
public int Replicas { get; set; } = 1;
public string? Mirror { get; set; }
public string? Source { get; set; }
public List<StreamSourceConfig> Sources { get; set; } = [];
}
public enum StorageType
{
Memory,
File,
}
public sealed class StreamSourceConfig
{
public string Name { get; set; } = string.Empty;
}

1
src/NATS.Server/JetStream/Models/StreamState.cs
View File

@@ -5,4 +5,5 @@ public sealed class StreamState
public ulong Messages { get; set; }
public ulong FirstSeq { get; set; }
public ulong LastSeq { get; set; }
public ulong Bytes { get; set; }
}

7
src/NATS.Server/JetStream/Storage/FileStore.cs
View File

@@ -24,6 +24,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable
Sequence = _last,
Subject = subject,
Payload = payload.ToArray(),
TimestampUtc = DateTime.UtcNow,
};
_messages[_last] = stored;
@@ -32,6 +33,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable
Sequence = stored.Sequence,
Subject = stored.Subject,
PayloadBase64 = Convert.ToBase64String(stored.Payload.ToArray()),
TimestampUtc = stored.TimestampUtc,
});
await File.AppendAllTextAsync(_dataFilePath, line + Environment.NewLine, ct);
return _last;
@@ -79,6 +81,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable
Sequence = x.Sequence,
Subject = x.Subject,
PayloadBase64 = Convert.ToBase64String(x.Payload.ToArray()),
TimestampUtc = x.TimestampUtc,
})
.ToArray();
return ValueTask.FromResult(JsonSerializer.SerializeToUtf8Bytes(snapshot));
@@ -101,6 +104,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable
Sequence = record.Sequence,
Subject = record.Subject ?? string.Empty,
Payload = Convert.FromBase64String(record.PayloadBase64 ?? string.Empty),
TimestampUtc = record.TimestampUtc,
};
_messages[record.Sequence] = message;
_last = Math.Max(_last, record.Sequence);
@@ -119,6 +123,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable
Messages = (ulong)_messages.Count,
FirstSeq = _messages.Count == 0 ? 0UL : _messages.Keys.Min(),
LastSeq = _last,
Bytes = (ulong)_messages.Values.Sum(m => m.Payload.Length),
});
}
@@ -172,6 +177,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable
Sequence = message.Sequence,
Subject = message.Subject,
PayloadBase64 = Convert.ToBase64String(message.Payload.ToArray()),
TimestampUtc = message.TimestampUtc,
}));
}
@@ -183,5 +189,6 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable
public ulong Sequence { get; init; }
public string? Subject { get; init; }
public string? PayloadBase64 { get; init; }
public DateTime TimestampUtc { get; init; }
}
}

5
src/NATS.Server/JetStream/Storage/MemStore.cs
View File

@@ -10,6 +10,7 @@ public sealed class MemStore : IStreamStore
public ulong Sequence { get; init; }
public string Subject { get; init; } = string.Empty;
public string PayloadBase64 { get; init; } = string.Empty;
public DateTime TimestampUtc { get; init; }
}
private readonly object _gate = new();
@@ -26,6 +27,7 @@ public sealed class MemStore : IStreamStore
Sequence = _last,
Subject = subject,
Payload = payload,
TimestampUtc = DateTime.UtcNow,
};
return ValueTask.FromResult(_last);
}
@@ -82,6 +84,7 @@ public sealed class MemStore : IStreamStore
Sequence = x.Sequence,
Subject = x.Subject,
PayloadBase64 = Convert.ToBase64String(x.Payload.ToArray()),
TimestampUtc = x.TimestampUtc,
})
.ToArray();
return ValueTask.FromResult(JsonSerializer.SerializeToUtf8Bytes(snapshot));
@@ -107,6 +110,7 @@ public sealed class MemStore : IStreamStore
Sequence = record.Sequence,
Subject = record.Subject,
Payload = Convert.FromBase64String(record.PayloadBase64),
TimestampUtc = record.TimestampUtc,
};
_last = Math.Max(_last, record.Sequence);
}
@@ -126,6 +130,7 @@ public sealed class MemStore : IStreamStore
Messages = (ulong)_messages.Count,
FirstSeq = _messages.Count == 0 ? 0UL : _messages.Keys.Min(),
LastSeq = _last,
Bytes = (ulong)_messages.Values.Sum(m => m.Payload.Length),
});
}
}

1
src/NATS.Server/JetStream/Storage/StoredMessage.cs
View File

@@ -5,5 +5,6 @@ public sealed class StoredMessage
public ulong Sequence { get; init; }
public string Subject { get; init; } = string.Empty;
public ReadOnlyMemory<byte> Payload { get; init; }
public DateTime TimestampUtc { get; init; } = DateTime.UtcNow;
public bool Redelivered { get; init; }
}

70
src/NATS.Server/JetStream/StreamManager.cs
View File

@@ -48,8 +48,14 @@ public sealed class StreamManager
var handle = _streams.AddOrUpdate(
normalized.Name,
_ => new StreamHandle(normalized, new MemStore()),
(_, existing) => existing with { Config = normalized });
_ => new StreamHandle(normalized, CreateStore(normalized)),
(_, existing) =>
{
if (existing.Config.Storage == normalized.Storage)
return existing with { Config = normalized };
return new StreamHandle(normalized, CreateStore(normalized));
});
_replicaGroups.AddOrUpdate(
normalized.Name,
_ => new StreamReplicaGroup(normalized.Name, normalized.Replicas),
@@ -150,6 +156,25 @@ public sealed class StreamManager
if (stream == null)
return null;
var stateBefore = stream.Store.GetStateAsync(default).GetAwaiter().GetResult();
if (stream.Config.MaxBytes > 0 && (long)stateBefore.Bytes + payload.Length > stream.Config.MaxBytes)
{
if (stream.Config.Discard == DiscardPolicy.New)
{
return new PubAck
{
Stream = stream.Config.Name,
ErrorCode = 10054,
};
}
while ((long)stateBefore.Bytes + payload.Length > stream.Config.MaxBytes && stateBefore.FirstSeq > 0)
{
stream.Store.RemoveAsync(stateBefore.FirstSeq, default).GetAwaiter().GetResult();
stateBefore = stream.Store.GetStateAsync(default).GetAwaiter().GetResult();
}
}
if (_replicaGroups.TryGetValue(stream.Config.Name, out var replicaGroup))
_ = replicaGroup.ProposeAsync($"PUB {subject}", default).GetAwaiter().GetResult();
@@ -181,12 +206,17 @@ public sealed class StreamManager
Name = config.Name,
Subjects = config.Subjects.Count == 0 ? [] : [.. config.Subjects],
MaxMsgs = config.MaxMsgs,
MaxBytes = config.MaxBytes,
MaxMsgsPer = config.MaxMsgsPer,
MaxAgeMs = config.MaxAgeMs,
MaxConsumers = config.MaxConsumers,
Retention = config.Retention,
Discard = config.Discard,
Storage = config.Storage,
Replicas = config.Replicas,
Mirror = config.Mirror,
Source = config.Source,
Sources = config.Sources.Count == 0 ? [] : [.. config.Sources.Select(s => new StreamSourceConfig { Name = s.Name })],
};
return copy;
@@ -241,6 +271,18 @@ public sealed class StreamManager
var list = _sourcesByOrigin.GetOrAdd(stream.Config.Source, _ => []);
list.Add(new SourceCoordinator(stream.Store));
}
if (stream.Config.Sources.Count > 0)
{
foreach (var source in stream.Config.Sources)
{
if (string.IsNullOrWhiteSpace(source.Name) || !_streams.TryGetValue(source.Name, out _))
continue;
var list = _sourcesByOrigin.GetOrAdd(source.Name, _ => []);
list.Add(new SourceCoordinator(stream.Store));
}
}
}
}
@@ -258,6 +300,30 @@ public sealed class StreamManager
source.OnOriginAppendAsync(stored, default).GetAwaiter().GetResult();
}
}
public string GetStoreBackendType(string streamName)
{
if (!_streams.TryGetValue(streamName, out var stream))
return "missing";
return stream.Store switch
{
FileStore => "file",
_ => "memory",
};
}
private static IStreamStore CreateStore(StreamConfig config)
{
return config.Storage switch
{
StorageType.File => new FileStore(new FileStoreOptions
{
Directory = Path.Combine(Path.GetTempPath(), "natsdotnet-js-store", config.Name),
}),
_ => new MemStore(),
};
}
}
public sealed record StreamHandle(StreamConfig Config, IStreamStore Store);

188
src/NATS.Server/LeafNodes/LeafConnection.cs
View File

@@ -1,11 +1,191 @@
using System.Net.Sockets;
using System.Text;
using NATS.Server.Subscriptions;
namespace NATS.Server.LeafNodes;
public sealed class LeafConnection
public sealed class LeafConnection(Socket socket) : IAsyncDisposable
{
public string RemoteEndpoint { get; }
private readonly NetworkStream _stream = new(socket, ownsSocket: true);
private readonly SemaphoreSlim _writeGate = new(1, 1);
private readonly CancellationTokenSource _closedCts = new();
private Task? _loopTask;
public LeafConnection(string remoteEndpoint)
public string? RemoteId { get; private set; }
public string RemoteEndpoint => socket.RemoteEndPoint?.ToString() ?? Guid.NewGuid().ToString("N");
public Func<RemoteSubscription, Task>? RemoteSubscriptionReceived { get; set; }
public Func<LeafMessage, Task>? MessageReceived { get; set; }
public async Task PerformOutboundHandshakeAsync(string serverId, CancellationToken ct)
{
RemoteEndpoint = remoteEndpoint;
await WriteLineAsync($"LEAF {serverId}", ct);
var line = await ReadLineAsync(ct);
RemoteId = ParseHandshake(line);
}
public async Task PerformInboundHandshakeAsync(string serverId, CancellationToken ct)
{
var line = await ReadLineAsync(ct);
RemoteId = ParseHandshake(line);
await WriteLineAsync($"LEAF {serverId}", ct);
}
public void StartLoop(CancellationToken ct)
{
if (_loopTask != null)
return;
var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, _closedCts.Token);
_loopTask = Task.Run(() => ReadLoopAsync(linked.Token), linked.Token);
}
public Task WaitUntilClosedAsync(CancellationToken ct)
=> _loopTask?.WaitAsync(ct) ?? Task.CompletedTask;
public Task SendLsPlusAsync(string subject, string? queue, CancellationToken ct)
=> WriteLineAsync(queue is { Length: > 0 } ? $"LS+ {subject} {queue}" : $"LS+ {subject}", ct);
public Task SendLsMinusAsync(string subject, string? queue, CancellationToken ct)
=> WriteLineAsync(queue is { Length: > 0 } ? $"LS- {subject} {queue}" : $"LS- {subject}", ct);
public async Task SendMessageAsync(string subject, string? replyTo, ReadOnlyMemory<byte> payload, CancellationToken ct)
{
var reply = string.IsNullOrEmpty(replyTo) ? "-" : replyTo;
await _writeGate.WaitAsync(ct);
try
{
var control = Encoding.ASCII.GetBytes($"LMSG {subject} {reply} {payload.Length}\r\n");
await _stream.WriteAsync(control, ct);
if (!payload.IsEmpty)
await _stream.WriteAsync(payload, ct);
await _stream.WriteAsync("\r\n"u8.ToArray(), ct);
await _stream.FlushAsync(ct);
}
finally
{
_writeGate.Release();
}
}
public async ValueTask DisposeAsync()
{
await _closedCts.CancelAsync();
if (_loopTask != null)
await _loopTask.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
_closedCts.Dispose();
_writeGate.Dispose();
await _stream.DisposeAsync();
}
private async Task ReadLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
string line;
try
{
line = await ReadLineAsync(ct);
}
catch
{
break;
}
if (line.StartsWith("LS+ ", StringComparison.Ordinal))
{
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (parts.Length >= 2 && RemoteSubscriptionReceived != null)
{
var queue = parts.Length >= 3 ? parts[2] : null;
await RemoteSubscriptionReceived(new RemoteSubscription(parts[1], queue, RemoteId ?? string.Empty));
}
continue;
}
if (line.StartsWith("LS- ", StringComparison.Ordinal))
{
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (parts.Length >= 2 && RemoteSubscriptionReceived != null)
{
var queue = parts.Length >= 3 ? parts[2] : null;
await RemoteSubscriptionReceived(RemoteSubscription.Removal(parts[1], queue, RemoteId ?? string.Empty));
}
continue;
}
if (!line.StartsWith("LMSG ", StringComparison.Ordinal))
continue;
var args = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (args.Length < 4 || !int.TryParse(args[3], out var size) || size < 0)
continue;
var payload = await ReadPayloadAsync(size, ct);
if (MessageReceived != null)
await MessageReceived(new LeafMessage(args[1], args[2] == "-" ? null : args[2], payload));
}
}
private async Task<ReadOnlyMemory<byte>> ReadPayloadAsync(int size, CancellationToken ct)
{
var payload = new byte[size];
var offset = 0;
while (offset < size)
{
var read = await _stream.ReadAsync(payload.AsMemory(offset, size - offset), ct);
if (read == 0)
throw new IOException("Leaf payload read closed");
offset += read;
}
var trailer = new byte[2];
_ = await _stream.ReadAsync(trailer, ct);
return payload;
}
private async Task WriteLineAsync(string line, CancellationToken ct)
{
await _writeGate.WaitAsync(ct);
try
{
var bytes = Encoding.ASCII.GetBytes($"{line}\r\n");
await _stream.WriteAsync(bytes, ct);
await _stream.FlushAsync(ct);
}
finally
{
_writeGate.Release();
}
}
private async Task<string> ReadLineAsync(CancellationToken ct)
{
var bytes = new List<byte>(64);
var single = new byte[1];
while (true)
{
var read = await _stream.ReadAsync(single, ct);
if (read == 0)
throw new IOException("Leaf closed");
if (single[0] == (byte)'\n')
break;
if (single[0] != (byte)'\r')
bytes.Add(single[0]);
}
return Encoding.ASCII.GetString([.. bytes]);
}
private static string ParseHandshake(string line)
{
if (!line.StartsWith("LEAF ", StringComparison.OrdinalIgnoreCase))
throw new InvalidOperationException("Invalid leaf handshake");
var id = line[5..].Trim();
if (id.Length == 0)
throw new InvalidOperationException("Leaf handshake missing id");
return id;
}
}
public sealed record LeafMessage(string Subject, string? ReplyTo, ReadOnlyMemory<byte> Payload);

190
src/NATS.Server/LeafNodes/LeafNodeManager.cs
View File

@@ -1,5 +1,9 @@
using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
using Microsoft.Extensions.Logging;
using NATS.Server.Configuration;
using NATS.Server.Subscriptions;
namespace NATS.Server.LeafNodes;
@@ -7,25 +11,203 @@ public sealed class LeafNodeManager : IAsyncDisposable
{
private readonly LeafNodeOptions _options;
private readonly ServerStats _stats;
private readonly string _serverId;
private readonly Action<RemoteSubscription> _remoteSubSink;
private readonly Action<LeafMessage> _messageSink;
private readonly ILogger<LeafNodeManager> _logger;
private readonly ConcurrentDictionary<string, LeafConnection> _connections = new(StringComparer.Ordinal);
public LeafNodeManager(LeafNodeOptions options, ServerStats stats, ILogger<LeafNodeManager> logger)
private CancellationTokenSource? _cts;
private Socket? _listener;
private Task? _acceptLoopTask;
public string ListenEndpoint => $"{_options.Host}:{_options.Port}";
public LeafNodeManager(
LeafNodeOptions options,
ServerStats stats,
string serverId,
Action<RemoteSubscription> remoteSubSink,
Action<LeafMessage> messageSink,
ILogger<LeafNodeManager> logger)
{
_options = options;
_stats = stats;
_serverId = serverId;
_remoteSubSink = remoteSubSink;
_messageSink = messageSink;
_logger = logger;
}
public Task StartAsync(CancellationToken ct)
{
_cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
_listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
_listener.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
_listener.Bind(new IPEndPoint(IPAddress.Parse(_options.Host), _options.Port));
_listener.Listen(128);
if (_options.Port == 0)
_options.Port = ((IPEndPoint)_listener.LocalEndPoint!).Port;
_acceptLoopTask = Task.Run(() => AcceptLoopAsync(_cts.Token));
foreach (var remote in _options.Remotes.Distinct(StringComparer.OrdinalIgnoreCase))
_ = Task.Run(() => ConnectWithRetryAsync(remote, _cts.Token));
_logger.LogDebug("Leaf manager started (listen={Host}:{Port})", _options.Host, _options.Port);
Interlocked.Exchange(ref _stats.Leafs, 0);
return Task.CompletedTask;
}
public ValueTask DisposeAsync()
public async Task ForwardMessageAsync(string subject, string? replyTo, ReadOnlyMemory<byte> payload, CancellationToken ct)
{
foreach (var connection in _connections.Values)
await connection.SendMessageAsync(subject, replyTo, payload, ct);
}
public void PropagateLocalSubscription(string subject, string? queue)
{
foreach (var connection in _connections.Values)
_ = connection.SendLsPlusAsync(subject, queue, _cts?.Token ?? CancellationToken.None);
}
public void PropagateLocalUnsubscription(string subject, string? queue)
{
foreach (var connection in _connections.Values)
_ = connection.SendLsMinusAsync(subject, queue, _cts?.Token ?? CancellationToken.None);
}
public async ValueTask DisposeAsync()
{
if (_cts == null)
return;
await _cts.CancelAsync();
_listener?.Dispose();
if (_acceptLoopTask != null)
await _acceptLoopTask.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
foreach (var connection in _connections.Values)
await connection.DisposeAsync();
_connections.Clear();
Interlocked.Exchange(ref _stats.Leafs, 0);
_cts.Dispose();
_cts = null;
_logger.LogDebug("Leaf manager stopped");
return ValueTask.CompletedTask;
}
private async Task AcceptLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
Socket socket;
try
{
socket = await _listener!.AcceptAsync(ct);
}
catch
{
break;
}
_ = Task.Run(() => HandleInboundAsync(socket, ct), ct);
}
}
private async Task HandleInboundAsync(Socket socket, CancellationToken ct)
{
var connection = new LeafConnection(socket);
try
{
await connection.PerformInboundHandshakeAsync(_serverId, ct);
Register(connection);
}
catch
{
await connection.DisposeAsync();
}
}
private async Task ConnectWithRetryAsync(string remote, CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
try
{
var endPoint = ParseEndpoint(remote);
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await socket.ConnectAsync(endPoint.Address, endPoint.Port, ct);
var connection = new LeafConnection(socket);
await connection.PerformOutboundHandshakeAsync(_serverId, ct);
Register(connection);
return;
}
catch (OperationCanceledException)
{
return;
}
catch (Exception ex)
{
_logger.LogDebug(ex, "Leaf connect retry for {Remote}", remote);
}
try
{
await Task.Delay(250, ct);
}
catch (OperationCanceledException)
{
return;
}
}
}
private void Register(LeafConnection connection)
{
var key = $"{connection.RemoteId}:{connection.RemoteEndpoint}:{Guid.NewGuid():N}";
if (!_connections.TryAdd(key, connection))
{
_ = connection.DisposeAsync();
return;
}
connection.RemoteSubscriptionReceived = sub =>
{
_remoteSubSink(sub);
return Task.CompletedTask;
};
connection.MessageReceived = msg =>
{
_messageSink(msg);
return Task.CompletedTask;
};
connection.StartLoop(_cts!.Token);
Interlocked.Increment(ref _stats.Leafs);
_ = Task.Run(() => WatchConnectionAsync(key, connection, _cts!.Token));
}
private async Task WatchConnectionAsync(string key, LeafConnection connection, CancellationToken ct)
{
try
{
await connection.WaitUntilClosedAsync(ct);
}
catch
{
}
finally
{
if (_connections.TryRemove(key, out _))
Interlocked.Decrement(ref _stats.Leafs);
await connection.DisposeAsync();
}
}
private static IPEndPoint ParseEndpoint(string endpoint)
{
var parts = endpoint.Split(':', 2, StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries);
if (parts.Length != 2)
throw new FormatException($"Invalid endpoint: {endpoint}");
return new IPEndPoint(IPAddress.Parse(parts[0]), int.Parse(parts[1]));
}
}

46
src/NATS.Server/Monitoring/AccountzHandler.cs Normal file
View File

@@ -0,0 +1,46 @@
using NATS.Server.Auth;
namespace NATS.Server.Monitoring;
public sealed class AccountzHandler
{
private readonly NatsServer _server;
public AccountzHandler(NatsServer server)
{
_server = server;
}
public object Build()
{
var accounts = _server.GetAccounts().Select(ToAccountDto).ToArray();
return new
{
accounts,
num_accounts = accounts.Length,
};
}
public object BuildStats()
{
var accounts = _server.GetAccounts().ToArray();
return new
{
total_accounts = accounts.Length,
total_connections = accounts.Sum(a => a.ClientCount),
total_subscriptions = accounts.Sum(a => a.SubscriptionCount),
};
}
private static object ToAccountDto(Account account)
{
return new
{
name = account.Name,
connections = account.ClientCount,
subscriptions = account.SubscriptionCount,
in_msgs = account.InMsgs,
out_msgs = account.OutMsgs,
};
}
}

21
src/NATS.Server/Monitoring/GatewayzHandler.cs Normal file
View File

@@ -0,0 +1,21 @@
namespace NATS.Server.Monitoring;
public sealed class GatewayzHandler
{
private readonly NatsServer _server;
public GatewayzHandler(NatsServer server)
{
_server = server;
}
public object Build()
{
var gateways = _server.Stats.Gateways;
return new
{
gateways,
num_gateways = gateways,
};
}
}

21
src/NATS.Server/Monitoring/LeafzHandler.cs Normal file
View File

@@ -0,0 +1,21 @@
namespace NATS.Server.Monitoring;
public sealed class LeafzHandler
{
private readonly NatsServer _server;
public LeafzHandler(NatsServer server)
{
_server = server;
}
public object Build()
{
var leafs = _server.Stats.Leafs;
return new
{
leafs,
num_leafs = leafs,
};
}
}

19
src/NATS.Server/Monitoring/MonitorServer.cs
View File

@@ -17,6 +17,10 @@ public sealed class MonitorServer : IAsyncDisposable
private readonly ConnzHandler _connzHandler;
private readonly SubszHandler _subszHandler;
private readonly JszHandler _jszHandler;
private readonly RoutezHandler _routezHandler;
private readonly GatewayzHandler _gatewayzHandler;
private readonly LeafzHandler _leafzHandler;
private readonly AccountzHandler _accountzHandler;
public MonitorServer(NatsServer server, NatsOptions options, ServerStats stats, ILoggerFactory loggerFactory)
{
@@ -33,6 +37,10 @@ public sealed class MonitorServer : IAsyncDisposable
_connzHandler = new ConnzHandler(server);
_subszHandler = new SubszHandler(server);
_jszHandler = new JszHandler(server, options);
_routezHandler = new RoutezHandler(server);
_gatewayzHandler = new GatewayzHandler(server);
_leafzHandler = new LeafzHandler(server);
_accountzHandler = new AccountzHandler(server);
_app.MapGet(basePath + "/", () =>
{
@@ -63,21 +71,20 @@ public sealed class MonitorServer : IAsyncDisposable
return Results.Ok(_connzHandler.HandleConnz(ctx));
});
// Stubs for unimplemented endpoints
_app.MapGet(basePath + "/routez", () =>
{
stats.HttpReqStats.AddOrUpdate("/routez", 1, (_, v) => v + 1);
return Results.Ok(new { });
return Results.Ok(_routezHandler.Build());
});
_app.MapGet(basePath + "/gatewayz", () =>
{
stats.HttpReqStats.AddOrUpdate("/gatewayz", 1, (_, v) => v + 1);
return Results.Ok(new { });
return Results.Ok(_gatewayzHandler.Build());
});
_app.MapGet(basePath + "/leafz", () =>
{
stats.HttpReqStats.AddOrUpdate("/leafz", 1, (_, v) => v + 1);
return Results.Ok(new { });
return Results.Ok(_leafzHandler.Build());
});
_app.MapGet(basePath + "/subz", (HttpContext ctx) =>
{
@@ -92,12 +99,12 @@ public sealed class MonitorServer : IAsyncDisposable
_app.MapGet(basePath + "/accountz", () =>
{
stats.HttpReqStats.AddOrUpdate("/accountz", 1, (_, v) => v + 1);
return Results.Ok(new { });
return Results.Ok(_accountzHandler.Build());
});
_app.MapGet(basePath + "/accstatz", () =>
{
stats.HttpReqStats.AddOrUpdate("/accstatz", 1, (_, v) => v + 1);
return Results.Ok(new { });
return Results.Ok(_accountzHandler.BuildStats());
});
_app.MapGet(basePath + "/jsz", () =>
{

21
src/NATS.Server/Monitoring/RoutezHandler.cs Normal file
View File

@@ -0,0 +1,21 @@
namespace NATS.Server.Monitoring;
public sealed class RoutezHandler
{
private readonly NatsServer _server;
public RoutezHandler(NatsServer server)
{
_server = server;
}
public object Build()
{
var routes = _server.Stats.Routes;
return new
{
routes,
num_routes = routes,
};
}
}

2
src/NATS.Server/NatsClient.cs
View File

@@ -562,6 +562,8 @@ public sealed class NatsClient : INatsClient, IDisposable
Account?.DecrementSubscriptions();
Account?.SubList.Remove(sub);
if (Router is NatsServer server)
server.OnLocalUnsubscription(sub.Subject, sub.Queue);
}
private void ProcessPub(ParsedCommand cmd, ref long localInMsgs, ref long localInBytes)

57
src/NATS.Server/NatsServer.cs
View File

@@ -95,6 +95,8 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
public bool IsShuttingDown => Volatile.Read(ref _shutdown) != 0;
public bool IsLameDuckMode => Volatile.Read(ref _lameDuck) != 0;
public string? ClusterListen => _routeManager?.ListenEndpoint;
public string? GatewayListen => _gatewayManager?.ListenEndpoint;
public string? LeafListen => _leafNodeManager?.ListenEndpoint;
public JetStreamApiRouter? JetStreamApiRouter => _jetStreamApiRouter;
public int JetStreamStreams => _jetStreamStreamManager?.StreamNames.Count ?? 0;
public int JetStreamConsumers => _jetStreamConsumerManager?.ConsumerCount ?? 0;
@@ -366,18 +368,21 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
if (options.Cluster != null)
{
_routeManager = new RouteManager(options.Cluster, _stats, _serverInfo.ServerId, ApplyRemoteSubscription,
ProcessRoutedMessage,
_loggerFactory.CreateLogger<RouteManager>());
}
if (options.Gateway != null)
{
_gatewayManager = new GatewayManager(options.Gateway, _stats,
_gatewayManager = new GatewayManager(options.Gateway, _stats, _serverInfo.ServerId, ApplyRemoteSubscription,
ProcessGatewayMessage,
_loggerFactory.CreateLogger<GatewayManager>());
}
if (options.LeafNode != null)
{
_leafNodeManager = new LeafNodeManager(options.LeafNode, _stats,
_leafNodeManager = new LeafNodeManager(options.LeafNode, _stats, _serverInfo.ServerId, ApplyRemoteSubscription,
ProcessLeafMessage,
_loggerFactory.CreateLogger<LeafNodeManager>());
}
@@ -796,6 +801,15 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
public void OnLocalSubscription(string subject, string? queue)
{
_routeManager?.PropagateLocalSubscription(subject, queue);
_gatewayManager?.PropagateLocalSubscription(subject, queue);
_leafNodeManager?.PropagateLocalSubscription(subject, queue);
}
public void OnLocalUnsubscription(string subject, string? queue)
{
_routeManager?.PropagateLocalUnsubscription(subject, queue);
_gatewayManager?.PropagateLocalUnsubscription(subject, queue);
_leafNodeManager?.PropagateLocalUnsubscription(subject, queue);
}
private void ApplyRemoteSubscription(RemoteSubscription sub)
@@ -803,6 +817,38 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
_globalAccount.SubList.ApplyRemoteSub(sub);
}
private void ProcessRoutedMessage(RouteMessage message)
{
DeliverRemoteMessage(message.Subject, message.ReplyTo, message.Payload);
}
private void ProcessGatewayMessage(GatewayMessage message)
{
DeliverRemoteMessage(message.Subject, message.ReplyTo, message.Payload);
}
private void ProcessLeafMessage(LeafMessage message)
{
DeliverRemoteMessage(message.Subject, message.ReplyTo, message.Payload);
}
private void DeliverRemoteMessage(string subject, string? replyTo, ReadOnlyMemory<byte> payload)
{
var result = _globalAccount.SubList.Match(subject);
foreach (var sub in result.PlainSubs)
DeliverMessage(sub, subject, replyTo, default, payload);
foreach (var queueGroup in result.QueueSubs)
{
if (queueGroup.Length == 0)
continue;
var sub = queueGroup[0];
DeliverMessage(sub, subject, replyTo, default, payload);
}
}
public void ProcessMessage(string subject, string? replyTo, ReadOnlyMemory<byte> headers,
ReadOnlyMemory<byte> payload, NatsClient sender)
{
@@ -837,6 +883,13 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
}
}
if (_routeManager != null && _globalAccount.SubList.HasRemoteInterest(subject))
_routeManager.ForwardRoutedMessageAsync(subject, replyTo, payload, default).GetAwaiter().GetResult();
if (_gatewayManager != null && _globalAccount.SubList.HasRemoteInterest(subject))
_gatewayManager.ForwardMessageAsync(subject, replyTo, payload, default).GetAwaiter().GetResult();
if (_leafNodeManager != null && _globalAccount.SubList.HasRemoteInterest(subject))
_leafNodeManager.ForwardMessageAsync(subject, replyTo, payload, default).GetAwaiter().GetResult();
var subList = sender.Account?.SubList ?? _globalAccount.SubList;
var result = subList.Match(subject);
var delivered = false;

14
src/NATS.Server/Protocol/ClientCommandMatrix.cs
View File

@@ -10,7 +10,21 @@ public sealed class ClientCommandMatrix
return (kind, op.ToUpperInvariant()) switch
{
(global::NATS.Server.ClientKind.Router, "RS+") => true,
(global::NATS.Server.ClientKind.Router, "RS-") => true,
(global::NATS.Server.ClientKind.Router, "RMSG") => true,
(global::NATS.Server.ClientKind.Gateway, "A+") => true,
(global::NATS.Server.ClientKind.Gateway, "A-") => true,
(global::NATS.Server.ClientKind.Leaf, "LS+") => true,
(global::NATS.Server.ClientKind.Leaf, "LS-") => true,
(global::NATS.Server.ClientKind.Leaf, "LMSG") => true,
(_, "RS+") => false,
(_, "RS-") => false,
(_, "RMSG") => false,
(_, "A+") => false,
(_, "A-") => false,
(_, "LS+") => false,
(_, "LS-") => false,
(_, "LMSG") => false,
_ => true,
};
}

30
src/NATS.Server/Raft/RaftLog.cs
View File

@@ -27,6 +27,36 @@ public sealed class RaftLog
_entries.Clear();
_baseIndex = snapshot.LastIncludedIndex;
}
public async Task PersistAsync(string path, CancellationToken ct)
{
Directory.CreateDirectory(Path.GetDirectoryName(path)!);
var model = new PersistedLog
{
BaseIndex = _baseIndex,
Entries = [.. _entries],
};
await File.WriteAllTextAsync(path, System.Text.Json.JsonSerializer.Serialize(model), ct);
}
public static async Task<RaftLog> LoadAsync(string path, CancellationToken ct)
{
var log = new RaftLog();
if (!File.Exists(path))
return log;
var json = await File.ReadAllTextAsync(path, ct);
var model = System.Text.Json.JsonSerializer.Deserialize<PersistedLog>(json) ?? new PersistedLog();
log._baseIndex = model.BaseIndex;
log._entries.AddRange(model.Entries);
return log;
}
private sealed class PersistedLog
{
public long BaseIndex { get; set; }
public List<RaftLogEntry> Entries { get; set; } = [];
}
}
public sealed record RaftLogEntry(long Index, int Term, string Command);

43
src/NATS.Server/Raft/RaftNode.cs
View File

@@ -6,6 +6,8 @@ public sealed class RaftNode
private readonly List<RaftNode> _cluster = [];
private readonly RaftReplicator _replicator = new();
private readonly RaftSnapshotStore _snapshotStore = new();
private readonly IRaftTransport? _transport;
private readonly string? _persistDirectory;
public string Id { get; }
public int Term => TermState.CurrentTerm;
@@ -13,11 +15,13 @@ public sealed class RaftNode
public RaftRole Role { get; private set; } = RaftRole.Follower;
public RaftTermState TermState { get; } = new();
public long AppliedIndex { get; set; }
public RaftLog Log { get; } = new();
public RaftLog Log { get; private set; } = new();
public RaftNode(string id)
public RaftNode(string id, IRaftTransport? transport = null, string? persistDirectory = null)
{
Id = id;
_transport = transport;
_persistDirectory = persistDirectory;
}
public void ConfigureCluster(IEnumerable<RaftNode> peers)
@@ -60,7 +64,8 @@ public sealed class RaftNode
var entry = Log.Append(TermState.CurrentTerm, command);
var followers = _cluster.Where(n => n.Id != Id).ToList();
var acknowledgements = _replicator.Replicate(entry, followers);
var results = await _replicator.ReplicateAsync(Id, entry, followers, _transport, ct);
var acknowledgements = results.Count(r => r.Success);
var quorum = (_cluster.Count / 2) + 1;
if (acknowledgements + 1 >= quorum)
@@ -68,9 +73,14 @@ public sealed class RaftNode
AppliedIndex = entry.Index;
foreach (var node in _cluster)
node.AppliedIndex = Math.Max(node.AppliedIndex, entry.Index);
foreach (var node in _cluster.Where(n => n._persistDirectory != null))
await node.PersistAsync(ct);
}
await Task.CompletedTask;
if (_persistDirectory != null)
await PersistAsync(ct);
return entry.Index;
}
@@ -120,4 +130,29 @@ public sealed class RaftNode
if (_votesReceived >= quorum)
Role = RaftRole.Leader;
}
public async Task PersistAsync(CancellationToken ct)
{
var dir = _persistDirectory ?? Path.Combine(Path.GetTempPath(), "natsdotnet-raft", Id);
Directory.CreateDirectory(dir);
await Log.PersistAsync(Path.Combine(dir, "log.json"), ct);
await File.WriteAllTextAsync(Path.Combine(dir, "term.txt"), TermState.CurrentTerm.ToString(), ct);
await File.WriteAllTextAsync(Path.Combine(dir, "applied.txt"), AppliedIndex.ToString(), ct);
}
public async Task LoadPersistedStateAsync(CancellationToken ct)
{
var dir = _persistDirectory ?? Path.Combine(Path.GetTempPath(), "natsdotnet-raft", Id);
Log = await RaftLog.LoadAsync(Path.Combine(dir, "log.json"), ct);
var termPath = Path.Combine(dir, "term.txt");
if (File.Exists(termPath) && int.TryParse(await File.ReadAllTextAsync(termPath, ct), out var term))
TermState.CurrentTerm = term;
var appliedPath = Path.Combine(dir, "applied.txt");
if (File.Exists(appliedPath) && long.TryParse(await File.ReadAllTextAsync(appliedPath, ct), out var applied))
AppliedIndex = applied;
else if (Log.Entries.Count > 0)
AppliedIndex = Log.Entries[^1].Index;
}
}

20
src/NATS.Server/Raft/RaftReplicator.cs
View File

@@ -13,4 +13,24 @@ public sealed class RaftReplicator
return acknowledgements;
}
public async Task<IReadOnlyList<AppendResult>> ReplicateAsync(
string leaderId,
RaftLogEntry entry,
IReadOnlyList<RaftNode> followers,
IRaftTransport? transport,
CancellationToken ct)
{
if (transport != null)
return await transport.AppendEntriesAsync(leaderId, followers.Select(f => f.Id).ToArray(), entry, ct);
var results = new List<AppendResult>(followers.Count);
foreach (var follower in followers)
{
follower.ReceiveReplicatedEntry(entry);
results.Add(new AppendResult { FollowerId = follower.Id, Success = true });
}
return results;
}
}

6
src/NATS.Server/Raft/RaftRpcContracts.cs
View File

@@ -10,3 +10,9 @@ public sealed class VoteResponse
{
public bool Granted { get; init; }
}
public sealed class AppendResult
{
public string FollowerId { get; init; } = string.Empty;
public bool Success { get; init; }
}

44
src/NATS.Server/Raft/RaftTransport.cs Normal file
View File

@@ -0,0 +1,44 @@
namespace NATS.Server.Raft;
public interface IRaftTransport
{
Task<IReadOnlyList<AppendResult>> AppendEntriesAsync(string leaderId, IReadOnlyList<string> followerIds, RaftLogEntry entry, CancellationToken ct);
Task<VoteResponse> RequestVoteAsync(string candidateId, string voterId, VoteRequest request, CancellationToken ct);
}
public sealed class InMemoryRaftTransport : IRaftTransport
{
private readonly Dictionary<string, RaftNode> _nodes = new(StringComparer.Ordinal);
public void Register(RaftNode node)
{
_nodes[node.Id] = node;
}
public Task<IReadOnlyList<AppendResult>> AppendEntriesAsync(string leaderId, IReadOnlyList<string> followerIds, RaftLogEntry entry, CancellationToken ct)
{
var results = new List<AppendResult>(followerIds.Count);
foreach (var followerId in followerIds)
{
if (_nodes.TryGetValue(followerId, out var node))
{
node.ReceiveReplicatedEntry(entry);
results.Add(new AppendResult { FollowerId = followerId, Success = true });
}
else
{
results.Add(new AppendResult { FollowerId = followerId, Success = false });
}
}
return Task.FromResult<IReadOnlyList<AppendResult>>(results);
}
public Task<VoteResponse> RequestVoteAsync(string candidateId, string voterId, VoteRequest request, CancellationToken ct)
{
if (_nodes.TryGetValue(voterId, out var node))
return Task.FromResult(node.GrantVote(request.Term));
return Task.FromResult(new VoteResponse { Granted = false });
}
}

171
src/NATS.Server/Routes/RouteConnection.cs
View File

@@ -1,5 +1,6 @@
using System.Net.Sockets;
using System.Text;
using NATS.Server.Subscriptions;
namespace NATS.Server.Routes;
@@ -7,9 +8,14 @@ public sealed class RouteConnection(Socket socket) : IAsyncDisposable
{
private readonly Socket _socket = socket;
private readonly NetworkStream _stream = new(socket, ownsSocket: true);
private readonly SemaphoreSlim _writeGate = new(1, 1);
private readonly CancellationTokenSource _closedCts = new();
private Task? _frameLoopTask;
public string? RemoteServerId { get; private set; }
public string RemoteEndpoint => _socket.RemoteEndPoint?.ToString() ?? Guid.NewGuid().ToString("N");
public Func<RemoteSubscription, Task>? RemoteSubscriptionReceived { get; set; }
public Func<RouteMessage, Task>? RoutedMessageReceived { get; set; }
public async Task PerformOutboundHandshakeAsync(string serverId, CancellationToken ct)
{
@@ -25,27 +31,168 @@ public sealed class RouteConnection(Socket socket) : IAsyncDisposable
await WriteLineAsync($"ROUTE {serverId}", ct);
}
public void StartFrameLoop(CancellationToken ct)
{
if (_frameLoopTask != null)
return;
var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, _closedCts.Token);
_frameLoopTask = Task.Run(() => ReadFramesAsync(linked.Token), linked.Token);
}
public async Task SendRsPlusAsync(string subject, string? queue, CancellationToken ct)
{
var frame = queue is { Length: > 0 }
? $"RS+ {subject} {queue}"
: $"RS+ {subject}";
await WriteLineAsync(frame, ct);
}
public async Task SendRsMinusAsync(string subject, string? queue, CancellationToken ct)
{
var frame = queue is { Length: > 0 }
? $"RS- {subject} {queue}"
: $"RS- {subject}";
await WriteLineAsync(frame, ct);
}
public async Task SendRmsgAsync(string subject, string? replyTo, ReadOnlyMemory<byte> payload, CancellationToken ct)
{
var replyToken = string.IsNullOrEmpty(replyTo) ? "-" : replyTo;
await _writeGate.WaitAsync(ct);
try
{
var control = Encoding.ASCII.GetBytes($"RMSG {subject} {replyToken} {payload.Length}\r\n");
await _stream.WriteAsync(control, ct);
if (!payload.IsEmpty)
await _stream.WriteAsync(payload, ct);
await _stream.WriteAsync("\r\n"u8.ToArray(), ct);
await _stream.FlushAsync(ct);
}
finally
{
_writeGate.Release();
}
}
public async Task WaitUntilClosedAsync(CancellationToken ct)
{
var buffer = new byte[1024];
while (!ct.IsCancellationRequested)
{
var bytesRead = await _stream.ReadAsync(buffer, ct);
if (bytesRead == 0)
return;
}
if (_frameLoopTask == null)
return;
using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, _closedCts.Token);
await _frameLoopTask.WaitAsync(linked.Token);
}
public async ValueTask DisposeAsync()
{
await _closedCts.CancelAsync();
if (_frameLoopTask != null)
await _frameLoopTask.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
_closedCts.Dispose();
_writeGate.Dispose();
await _stream.DisposeAsync();
}
private async Task ReadFramesAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
string line;
try
{
line = await ReadLineAsync(ct);
}
catch (OperationCanceledException)
{
break;
}
catch (IOException)
{
break;
}
if (line.StartsWith("RS+ ", StringComparison.Ordinal))
{
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (parts.Length >= 2 && RemoteSubscriptionReceived != null)
{
var queue = parts.Length >= 3 ? parts[2] : null;
await RemoteSubscriptionReceived(new RemoteSubscription(parts[1], queue, RemoteServerId ?? string.Empty));
}
continue;
}
if (line.StartsWith("RS- ", StringComparison.Ordinal))
{
var parts = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (parts.Length >= 2 && RemoteSubscriptionReceived != null)
{
var queue = parts.Length >= 3 ? parts[2] : null;
await RemoteSubscriptionReceived(RemoteSubscription.Removal(parts[1], queue, RemoteServerId ?? string.Empty));
}
continue;
}
if (!line.StartsWith("RMSG ", StringComparison.Ordinal))
continue;
var args = line.Split(' ', StringSplitOptions.RemoveEmptyEntries);
if (args.Length < 4)
continue;
var subject = args[1];
var reply = args[2] == "-" ? null : args[2];
if (!int.TryParse(args[3], out var size) || size < 0)
continue;
var payload = await ReadPayloadAsync(size, ct);
if (RoutedMessageReceived != null)
await RoutedMessageReceived(new RouteMessage(subject, reply, payload));
}
}
private async Task<ReadOnlyMemory<byte>> ReadPayloadAsync(int size, CancellationToken ct)
{
var payload = new byte[size];
var offset = 0;
while (offset < size)
{
var read = await _stream.ReadAsync(payload.AsMemory(offset, size - offset), ct);
if (read == 0)
throw new IOException("Route connection closed during payload read");
offset += read;
}
var trailer = new byte[2];
var trailerRead = 0;
while (trailerRead < 2)
{
var read = await _stream.ReadAsync(trailer.AsMemory(trailerRead, 2 - trailerRead), ct);
if (read == 0)
throw new IOException("Route connection closed during payload trailer read");
trailerRead += read;
}
if (trailer[0] != (byte)'\r' || trailer[1] != (byte)'\n')
throw new IOException("Invalid route payload trailer");
return payload;
}
private async Task WriteLineAsync(string line, CancellationToken ct)
{
var bytes = Encoding.ASCII.GetBytes($"{line}\r\n");
await _stream.WriteAsync(bytes, ct);
await _stream.FlushAsync(ct);
await _writeGate.WaitAsync(ct);
try
{
var bytes = Encoding.ASCII.GetBytes($"{line}\r\n");
await _stream.WriteAsync(bytes, ct);
await _stream.FlushAsync(ct);
}
finally
{
_writeGate.Release();
}
}
private async Task<string> ReadLineAsync(CancellationToken ct)
@@ -56,7 +203,7 @@ public sealed class RouteConnection(Socket socket) : IAsyncDisposable
{
var read = await _stream.ReadAsync(single, ct);
if (read == 0)
throw new IOException("Route connection closed during handshake");
throw new IOException("Route connection closed");
if (single[0] == (byte)'\n')
break;
@@ -79,3 +226,5 @@ public sealed class RouteConnection(Socket socket) : IAsyncDisposable
return id;
}
}
public sealed record RouteMessage(string Subject, string? ReplyTo, ReadOnlyMemory<byte> Payload);

54
src/NATS.Server/Routes/RouteManager.cs
View File

@@ -15,6 +15,7 @@ public sealed class RouteManager : IAsyncDisposable
private readonly string _serverId;
private readonly ILogger<RouteManager> _logger;
private readonly Action<RemoteSubscription> _remoteSubSink;
private readonly Action<RouteMessage> _routedMessageSink;
private readonly ConcurrentDictionary<string, RouteConnection> _routes = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, byte> _connectedServerIds = new(StringComparer.Ordinal);
@@ -29,12 +30,14 @@ public sealed class RouteManager : IAsyncDisposable
ServerStats stats,
string serverId,
Action<RemoteSubscription> remoteSubSink,
Action<RouteMessage> routedMessageSink,
ILogger<RouteManager> logger)
{
_options = options;
_stats = stats;
_serverId = serverId;
_remoteSubSink = remoteSubSink;
_routedMessageSink = routedMessageSink;
_logger = logger;
}
@@ -51,8 +54,12 @@ public sealed class RouteManager : IAsyncDisposable
_options.Port = ((IPEndPoint)_listener.LocalEndPoint!).Port;
_acceptLoopTask = Task.Run(() => AcceptLoopAsync(_cts.Token));
var poolSize = Math.Max(_options.PoolSize, 1);
foreach (var route in _options.Routes.Distinct(StringComparer.OrdinalIgnoreCase))
_ = Task.Run(() => ConnectToRouteWithRetryAsync(route, _cts.Token));
{
for (var i = 0; i < poolSize; i++)
_ = Task.Run(() => ConnectToRouteWithRetryAsync(route, _cts.Token));
}
return Task.CompletedTask;
}
@@ -81,17 +88,33 @@ public sealed class RouteManager : IAsyncDisposable
public void PropagateLocalSubscription(string subject, string? queue)
{
if (_connectedServerIds.IsEmpty)
if (_routes.IsEmpty)
return;
var remoteSub = new RemoteSubscription(subject, queue, _serverId);
foreach (var peerId in _connectedServerIds.Keys)
foreach (var route in _routes.Values)
{
if (Managers.TryGetValue(peerId, out var peer))
peer.ReceiveRemoteSubscription(remoteSub);
_ = route.SendRsPlusAsync(subject, queue, _cts?.Token ?? CancellationToken.None);
}
}
public void PropagateLocalUnsubscription(string subject, string? queue)
{
if (_routes.IsEmpty)
return;
foreach (var route in _routes.Values)
_ = route.SendRsMinusAsync(subject, queue, _cts?.Token ?? CancellationToken.None);
}
public async Task ForwardRoutedMessageAsync(string subject, string? replyTo, ReadOnlyMemory<byte> payload, CancellationToken ct)
{
if (_routes.IsEmpty)
return;
foreach (var route in _routes.Values)
await route.SendRmsgAsync(subject, replyTo, payload, ct);
}
private async Task AcceptLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
@@ -170,7 +193,7 @@ public sealed class RouteManager : IAsyncDisposable
private void Register(RouteConnection route)
{
var key = $"{route.RemoteServerId}:{route.RemoteEndpoint}";
var key = $"{route.RemoteServerId}:{route.RemoteEndpoint}:{Guid.NewGuid():N}";
if (!_routes.TryAdd(key, route))
{
_ = route.DisposeAsync();
@@ -180,6 +203,18 @@ public sealed class RouteManager : IAsyncDisposable
if (route.RemoteServerId is { Length: > 0 } remoteServerId)
_connectedServerIds[remoteServerId] = 0;
route.RemoteSubscriptionReceived = sub =>
{
_remoteSubSink(sub);
return Task.CompletedTask;
};
route.RoutedMessageReceived = msg =>
{
_routedMessageSink(msg);
return Task.CompletedTask;
};
route.StartFrameLoop(_cts!.Token);
Interlocked.Increment(ref _stats.Routes);
_ = Task.Run(() => WatchRouteAsync(key, route, _cts!.Token));
}
@@ -217,8 +252,5 @@ public sealed class RouteManager : IAsyncDisposable
return new IPEndPoint(IPAddress.Parse(parts[0]), int.Parse(parts[1]));
}
private void ReceiveRemoteSubscription(RemoteSubscription sub)
{
_remoteSubSink(sub);
}
public int RouteCount => _routes.Count;
}

6
src/NATS.Server/Subscriptions/RemoteSubscription.cs
View File

@@ -1,3 +1,7 @@
namespace NATS.Server.Subscriptions;
public sealed record RemoteSubscription(string Subject, string? Queue, string RouteId);
public sealed record RemoteSubscription(string Subject, string? Queue, string RouteId, bool IsRemoval = false)
{
public static RemoteSubscription Removal(string subject, string? queue, string routeId)
=> new(subject, queue, routeId, IsRemoval: true);
}

8
src/NATS.Server/Subscriptions/SubList.cs
View File

@@ -103,7 +103,10 @@ public sealed class SubList : IDisposable
try
{
var key = $"{sub.RouteId}|{sub.Subject}|{sub.Queue}";
_remoteSubs[key] = sub;
if (sub.IsRemoval)
_remoteSubs.Remove(key);
else
_remoteSubs[key] = sub;
Interlocked.Increment(ref _generation);
}
finally
@@ -119,6 +122,9 @@ public sealed class SubList : IDisposable
{
foreach (var remoteSub in _remoteSubs.Values)
{
if (remoteSub.IsRemoval)
continue;
if (SubjectMatch.MatchLiteral(subject, remoteSub.Subject))
return true;
}