Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bbacf439ed | |||
| 09c42538c2 | |||
| 3501320c6e | |||
| 51c899b651 | |||
| acf51bf480 | |||
| e51cdd64f4 | |||
| f2bc957229 |
@@ -2279,7 +2279,7 @@ public sealed class Account : INatsAccount
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (sid is { Length: > 0 } && InternalClient != null)
|
if (sid is { Length: > 0 } && InternalClient != null)
|
||||||
InternalClient.ProcessUnsub(sid);
|
InternalClient.RemoveSubBySid(sid);
|
||||||
|
|
||||||
if (tracking && requestor != null && !delivered)
|
if (tracking && requestor != null && !delivered)
|
||||||
SendBackendErrorTrackingLatency(serviceImport, reason);
|
SendBackendErrorTrackingLatency(serviceImport, reason);
|
||||||
@@ -2355,7 +2355,7 @@ public sealed class Account : INatsAccount
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (sid != null && InternalClient != null)
|
if (sid != null && InternalClient != null)
|
||||||
InternalClient.ProcessUnsub(sid);
|
InternalClient.RemoveSubBySid(sid);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -2548,7 +2548,7 @@ public sealed class Account : INatsAccount
|
|||||||
if (InternalClient == null && Server is NatsServer server)
|
if (InternalClient == null && Server is NatsServer server)
|
||||||
{
|
{
|
||||||
InternalClient = server.CreateInternalAccountClient();
|
InternalClient = server.CreateInternalAccountClient();
|
||||||
InternalClient.Account = this;
|
InternalClient.SetAccount(this);
|
||||||
}
|
}
|
||||||
|
|
||||||
return InternalClient;
|
return InternalClient;
|
||||||
@@ -2573,7 +2573,7 @@ public sealed class Account : INatsAccount
|
|||||||
_mu.EnterReadLock();
|
_mu.EnterReadLock();
|
||||||
var internalClient = InternalClient;
|
var internalClient = InternalClient;
|
||||||
_mu.ExitReadLock();
|
_mu.ExitReadLock();
|
||||||
internalClient?.ProcessUnsub(sub.Sid);
|
internalClient?.RemoveSubBySid(sub.Sid);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -2685,7 +2685,7 @@ public sealed class Account : INatsAccount
|
|||||||
return;
|
return;
|
||||||
|
|
||||||
foreach (var sid in subscriptionIds)
|
foreach (var sid in subscriptionIds)
|
||||||
internalClient.ProcessUnsub(sid);
|
internalClient.RemoveSubBySid(sid);
|
||||||
|
|
||||||
internalClient.CloseConnection(ClosedState.InternalClient);
|
internalClient.CloseConnection(ClosedState.InternalClient);
|
||||||
}
|
}
|
||||||
@@ -4170,7 +4170,7 @@ public sealed class Account : INatsAccount
|
|||||||
return new ClientInfo
|
return new ClientInfo
|
||||||
{
|
{
|
||||||
Id = client.Cid,
|
Id = client.Cid,
|
||||||
Account = client.Account?.Name ?? string.Empty,
|
Account = client.Account()?.Name ?? string.Empty,
|
||||||
Name = client.Opts.Name ?? string.Empty,
|
Name = client.Opts.Name ?? string.Empty,
|
||||||
Rtt = client.GetRttValue(),
|
Rtt = client.GetRttValue(),
|
||||||
Start = client.Start == default ? string.Empty : client.Start.ToUniversalTime().ToString("O"),
|
Start = client.Start == default ? string.Empty : client.Start.ToUniversalTime().ToString("O"),
|
||||||
|
|||||||
@@ -27,6 +27,8 @@ public sealed partial class ClientConnection
|
|||||||
return;
|
return;
|
||||||
|
|
||||||
var staleAfter = TimeSpan.FromTicks(pingInterval.Ticks * (pingMax + 1L));
|
var staleAfter = TimeSpan.FromTicks(pingInterval.Ticks * (pingMax + 1L));
|
||||||
|
if (pingMax == 0 && staleAfter > TimeSpan.Zero)
|
||||||
|
staleAfter = TimeSpan.FromTicks(Math.Max(1, pingInterval.Ticks / 2));
|
||||||
if (staleAfter <= TimeSpan.Zero)
|
if (staleAfter <= TimeSpan.Zero)
|
||||||
return;
|
return;
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,486 @@
|
|||||||
|
// Copyright 2012-2026 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
|
||||||
|
using System.Text;
|
||||||
|
using System.Text.Json;
|
||||||
|
using ZB.MOM.NatsNet.Server.Auth;
|
||||||
|
using ZB.MOM.NatsNet.Server.Internal;
|
||||||
|
using ZB.MOM.NatsNet.Server.Protocol;
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server;
|
||||||
|
|
||||||
|
public sealed partial class ClientConnection
|
||||||
|
{
|
||||||
|
internal void RemoveReplySub(Subscription? sub)
|
||||||
|
{
|
||||||
|
if (sub?.Sid is not { Length: > 0 } sid || Server is not NatsServer server)
|
||||||
|
return;
|
||||||
|
|
||||||
|
var sidText = Encoding.ASCII.GetString(sid);
|
||||||
|
var sep = sidText.IndexOf(' ');
|
||||||
|
if (sep <= 0)
|
||||||
|
return;
|
||||||
|
|
||||||
|
var accountName = sidText[..sep];
|
||||||
|
var (account, _) = server.LookupAccount(accountName);
|
||||||
|
account?.Sublist?.Remove(sub);
|
||||||
|
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
Subs.Remove(sidText);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? ProcessAccountSub(byte[] arg)
|
||||||
|
{
|
||||||
|
_ = arg;
|
||||||
|
// Gateway account-sub propagation is owned by gateway sessions.
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void ProcessAccountUnsub(byte[] arg)
|
||||||
|
{
|
||||||
|
_ = arg;
|
||||||
|
// Gateway account-unsub propagation is owned by gateway sessions.
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? ProcessRoutedOriginClusterMsgArgs(byte[] arg) =>
|
||||||
|
ProtocolParser.ProcessRoutedOriginClusterMsgArgs(ParseCtx, arg);
|
||||||
|
|
||||||
|
internal Exception? ProcessRoutedHeaderMsgArgs(byte[] arg) =>
|
||||||
|
ProtocolParser.ProcessRoutedHeaderMsgArgs(ParseCtx, arg);
|
||||||
|
|
||||||
|
internal Exception? ProcessRoutedMsgArgs(byte[] arg) =>
|
||||||
|
ProtocolParser.ProcessRoutedMsgArgs(ParseCtx, arg);
|
||||||
|
|
||||||
|
internal void ProcessInboundRoutedMsg(byte[] msg)
|
||||||
|
{
|
||||||
|
_in.Msgs++;
|
||||||
|
_in.Bytes += Math.Max(0, msg.Length - 2);
|
||||||
|
|
||||||
|
if (Opts.Verbose)
|
||||||
|
SendOK();
|
||||||
|
|
||||||
|
var pa = ParseCtx.Pa;
|
||||||
|
if (pa.Subject is null)
|
||||||
|
return;
|
||||||
|
|
||||||
|
var (acc, result) = GetAccAndResultFromCache();
|
||||||
|
if (acc is null)
|
||||||
|
return;
|
||||||
|
|
||||||
|
if ((result?.PSubs.Count ?? 0) + (result?.QSubs.Count ?? 0) > 0)
|
||||||
|
ProcessMsgResults(acc, result, msg, null, pa.Subject, pa.Reply, PmrFlags.None);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? SendRouteConnect(string clusterName, bool tlsRequired)
|
||||||
|
{
|
||||||
|
var user = string.Empty;
|
||||||
|
var pass = string.Empty;
|
||||||
|
var routeUrl = Route?.Url;
|
||||||
|
if (routeUrl is not null && !string.IsNullOrEmpty(routeUrl.UserInfo))
|
||||||
|
{
|
||||||
|
var userInfo = routeUrl.UserInfo.Split(':', 2);
|
||||||
|
user = userInfo[0];
|
||||||
|
if (userInfo.Length > 1)
|
||||||
|
pass = userInfo[1];
|
||||||
|
}
|
||||||
|
|
||||||
|
if (Server is not NatsServer server)
|
||||||
|
return new InvalidOperationException("route server unavailable");
|
||||||
|
|
||||||
|
var connect = new ConnectInfo
|
||||||
|
{
|
||||||
|
Echo = true,
|
||||||
|
Verbose = false,
|
||||||
|
Pedantic = false,
|
||||||
|
User = user,
|
||||||
|
Pass = pass,
|
||||||
|
Tls = tlsRequired,
|
||||||
|
Name = server.ID(),
|
||||||
|
Headers = server.SupportsHeaders(),
|
||||||
|
Cluster = clusterName,
|
||||||
|
Dynamic = server.IsClusterNameDynamic(),
|
||||||
|
Lnoc = true,
|
||||||
|
};
|
||||||
|
|
||||||
|
var payload = JsonSerializer.Serialize(connect);
|
||||||
|
EnqueueProto(Encoding.ASCII.GetBytes($"CONNECT {payload}\r\n"));
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void ProcessRouteInfo(ServerInfo info)
|
||||||
|
{
|
||||||
|
if (Server is not NatsServer server)
|
||||||
|
return;
|
||||||
|
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
Route ??= new Route();
|
||||||
|
|
||||||
|
if (Flags.IsSet(ClientFlags.InfoReceived))
|
||||||
|
{
|
||||||
|
Opts.Import = info.Import;
|
||||||
|
Opts.Export = info.Export;
|
||||||
|
}
|
||||||
|
|
||||||
|
Route.RemoteId = info.Id;
|
||||||
|
Route.RemoteName = info.Name;
|
||||||
|
Route.AuthRequired = info.AuthRequired;
|
||||||
|
Route.TlsRequired = info.TlsRequired;
|
||||||
|
Route.GatewayUrl = info.GatewayUrl ?? string.Empty;
|
||||||
|
Route.Lnoc = info.Lnoc;
|
||||||
|
Route.Lnocu = info.Lnocu;
|
||||||
|
Route.JetStream = info.JetStream;
|
||||||
|
Route.ConnectUrls = info.ClientConnectUrls?.ToList() ?? [];
|
||||||
|
Route.WsConnUrls = info.WsConnectUrls?.ToList() ?? [];
|
||||||
|
Route.LeafnodeUrl = info.LeafNodeUrls is { Length: 1 } leaf ? leaf[0] : string.Empty;
|
||||||
|
Route.Hash = NatsServer.GetHash(info.Name);
|
||||||
|
Route.IdHash = NatsServer.GetHash(info.Id);
|
||||||
|
|
||||||
|
Opts.Protocol = info.Proto;
|
||||||
|
Headers = server.SupportsHeaders() && info.Headers;
|
||||||
|
Flags |= ClientFlags.InfoReceived;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (NatsServer.NeedsCompression(server.GetOpts().Cluster.Compression.Mode))
|
||||||
|
_ = server.NegotiateRouteCompression(this, Route?.DidSolicit == true, Route?.AccName is { Length: > 0 } an ? Encoding.ASCII.GetString(an) : string.Empty, info.Compression ?? string.Empty, server.GetOpts());
|
||||||
|
|
||||||
|
server.UpdateRemoteRoutePerms(this, info);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal bool CanImport(string subject) => PubAllowedFullCheck(subject, fullCheck: false, hasLock: true);
|
||||||
|
|
||||||
|
internal bool CanExport(string subject) => CanSubscribe(subject);
|
||||||
|
|
||||||
|
internal void SetRoutePermissions(RoutePermissions? perms)
|
||||||
|
{
|
||||||
|
if (perms is null)
|
||||||
|
{
|
||||||
|
Perms = null;
|
||||||
|
MPerms = null;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
SetPermissions(new Permissions
|
||||||
|
{
|
||||||
|
Publish = perms.Import?.Clone(),
|
||||||
|
Subscribe = perms.Export?.Clone(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
internal (bool IsPinnedAccountRoute, string AccountName, bool KeyHasSubType) GetRoutedSubKeyInfo()
|
||||||
|
{
|
||||||
|
var accountName = Route?.AccName is { Length: > 0 } an
|
||||||
|
? Encoding.ASCII.GetString(an)
|
||||||
|
: string.Empty;
|
||||||
|
return (!string.IsNullOrEmpty(accountName), accountName, Route?.Lnocu == true);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void RemoveRemoteSubs()
|
||||||
|
{
|
||||||
|
if (Server is not NatsServer server)
|
||||||
|
return;
|
||||||
|
|
||||||
|
Dictionary<string, Subscription> subs;
|
||||||
|
var grouped = new Dictionary<string, List<Subscription>>(StringComparer.Ordinal);
|
||||||
|
var (pinned, accountName, keyHasSubType) = GetRoutedSubKeyInfo();
|
||||||
|
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
subs = Subs;
|
||||||
|
Subs = new Dictionary<string, Subscription>(StringComparer.Ordinal);
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var kvp in subs)
|
||||||
|
{
|
||||||
|
var keyAccount = pinned
|
||||||
|
? accountName
|
||||||
|
: RouteHandler.GetAccNameFromRoutedSubKey(kvp.Value, kvp.Key, keyHasSubType);
|
||||||
|
if (string.IsNullOrEmpty(keyAccount))
|
||||||
|
continue;
|
||||||
|
|
||||||
|
if (!grouped.TryGetValue(keyAccount, out var list))
|
||||||
|
{
|
||||||
|
list = [];
|
||||||
|
grouped[keyAccount] = list;
|
||||||
|
}
|
||||||
|
list.Add(kvp.Value);
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var (accName, list) in grouped)
|
||||||
|
{
|
||||||
|
var (acc, _) = server.LookupAccount(accName);
|
||||||
|
acc?.Sublist?.RemoveBatch(list);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
internal List<Subscription> RemoveRemoteSubsForAcc(string name)
|
||||||
|
{
|
||||||
|
var removed = new List<Subscription>();
|
||||||
|
var (_, _, keyHasSubType) = GetRoutedSubKeyInfo();
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
foreach (var key in Subs.Keys.ToArray())
|
||||||
|
{
|
||||||
|
var sub = Subs[key];
|
||||||
|
if (RouteHandler.GetAccNameFromRoutedSubKey(sub, key, keyHasSubType) != name)
|
||||||
|
continue;
|
||||||
|
removed.Add(sub);
|
||||||
|
Subs.Remove(key);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return removed;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal (byte[] Origin, string AccountName, byte[] Subject, byte[] Queue, Exception? Error)
|
||||||
|
ParseUnsubProto(byte[] arg, bool accInProto, bool hasOrigin)
|
||||||
|
{
|
||||||
|
_in.Subs++;
|
||||||
|
|
||||||
|
var args = SplitArg(arg);
|
||||||
|
var origin = Array.Empty<byte>();
|
||||||
|
var queue = Array.Empty<byte>();
|
||||||
|
var subjectIndex = 0;
|
||||||
|
|
||||||
|
if (hasOrigin)
|
||||||
|
{
|
||||||
|
if (args.Count == 0)
|
||||||
|
return (origin, string.Empty, Array.Empty<byte>(), queue, new FormatException($"parse error: '{Encoding.ASCII.GetString(arg)}'"));
|
||||||
|
origin = args[0];
|
||||||
|
subjectIndex = 1;
|
||||||
|
}
|
||||||
|
if (accInProto)
|
||||||
|
subjectIndex++;
|
||||||
|
|
||||||
|
if (args.Count is not (>= 1) || args.Count < subjectIndex + 1 || args.Count > subjectIndex + 2)
|
||||||
|
return (origin, string.Empty, Array.Empty<byte>(), queue, new FormatException($"parse error: '{Encoding.ASCII.GetString(arg)}'"));
|
||||||
|
|
||||||
|
if (args.Count == subjectIndex + 2)
|
||||||
|
queue = args[subjectIndex + 1];
|
||||||
|
|
||||||
|
var accountName = accInProto ? Encoding.ASCII.GetString(args[subjectIndex - 1]) : string.Empty;
|
||||||
|
return (origin, accountName, args[subjectIndex], queue, null);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? ProcessRemoteUnsub(byte[] arg, bool leafUnsub)
|
||||||
|
{
|
||||||
|
if (Server is not NatsServer server)
|
||||||
|
return null;
|
||||||
|
|
||||||
|
string accountName;
|
||||||
|
var accInProto = true;
|
||||||
|
bool originSupport;
|
||||||
|
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
originSupport = Route?.Lnocu == true;
|
||||||
|
if (Route?.AccName is { Length: > 0 } an)
|
||||||
|
{
|
||||||
|
accountName = Encoding.ASCII.GetString(an);
|
||||||
|
accInProto = false;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
accountName = string.Empty;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var (_, protoAccName, subject, _, err) = ParseUnsubProto(arg, accInProto, leafUnsub && originSupport);
|
||||||
|
if (err is not null)
|
||||||
|
return new FormatException($"processRemoteUnsub {err.Message}");
|
||||||
|
|
||||||
|
if (accInProto)
|
||||||
|
accountName = protoAccName;
|
||||||
|
|
||||||
|
var (acc, _) = server.LookupAccount(accountName);
|
||||||
|
if (acc is null)
|
||||||
|
{
|
||||||
|
Debugf("Unknown account {0} for subject {1}", accountName, Encoding.ASCII.GetString(subject));
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
Subscription? sub = null;
|
||||||
|
var key = Encoding.ASCII.GetString(arg);
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
if (IsClosed())
|
||||||
|
return null;
|
||||||
|
if (Subs.TryGetValue(key, out sub))
|
||||||
|
{
|
||||||
|
Subs.Remove(key);
|
||||||
|
acc.Sublist?.Remove(sub);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (Opts.Verbose)
|
||||||
|
SendOK();
|
||||||
|
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? ProcessRemoteSub(byte[] protoArg, bool hasOrigin)
|
||||||
|
{
|
||||||
|
_in.Subs++;
|
||||||
|
if (Server is not NatsServer server)
|
||||||
|
return null;
|
||||||
|
|
||||||
|
var args = SplitArg(protoArg);
|
||||||
|
var (isPinned, accountName, _) = GetRoutedSubKeyInfo();
|
||||||
|
var accInProto = !isPinned;
|
||||||
|
var subjectIndex = 0;
|
||||||
|
|
||||||
|
if (hasOrigin)
|
||||||
|
subjectIndex++;
|
||||||
|
if (accInProto)
|
||||||
|
subjectIndex++;
|
||||||
|
|
||||||
|
if (args.Count is not (>= 1) || (args.Count != subjectIndex + 1 && args.Count != subjectIndex + 3))
|
||||||
|
return new FormatException($"processRemoteSub Parse Error: '{Encoding.ASCII.GetString(protoArg)}'");
|
||||||
|
|
||||||
|
if (accInProto)
|
||||||
|
accountName = Encoding.ASCII.GetString(args[subjectIndex - 1]);
|
||||||
|
var subject = args[subjectIndex];
|
||||||
|
byte[]? queue = null;
|
||||||
|
var qw = 1;
|
||||||
|
if (args.Count == subjectIndex + 3)
|
||||||
|
{
|
||||||
|
queue = args[subjectIndex + 1];
|
||||||
|
_ = int.TryParse(Encoding.ASCII.GetString(args[subjectIndex + 2]), out qw);
|
||||||
|
if (qw <= 0)
|
||||||
|
qw = 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
var (acc, _) = server.LookupOrRegisterAccount(accountName);
|
||||||
|
if (acc is null)
|
||||||
|
return null;
|
||||||
|
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
if (IsClosed())
|
||||||
|
return null;
|
||||||
|
if (Perms is not null && !CanExport(Encoding.ASCII.GetString(subject)))
|
||||||
|
return null;
|
||||||
|
if (SubsAtLimit())
|
||||||
|
{
|
||||||
|
MaxSubsExceeded();
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
var key = Encoding.ASCII.GetString(protoArg);
|
||||||
|
if (!Subs.ContainsKey(key))
|
||||||
|
{
|
||||||
|
var sub = new Subscription
|
||||||
|
{
|
||||||
|
Subject = subject,
|
||||||
|
Queue = queue,
|
||||||
|
Sid = Encoding.ASCII.GetBytes(key),
|
||||||
|
Qw = qw,
|
||||||
|
};
|
||||||
|
Subs[key] = sub;
|
||||||
|
acc.Sublist?.Insert(sub);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (Opts.Verbose)
|
||||||
|
SendOK();
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal byte[] AddRouteSubOrUnsubProtoToBuf(byte[] buf, string accName, Subscription sub, bool isSubProto)
|
||||||
|
{
|
||||||
|
var list = new List<byte>(buf.Length + 64);
|
||||||
|
list.AddRange(buf);
|
||||||
|
|
||||||
|
if (isSubProto)
|
||||||
|
list.AddRange(Encoding.ASCII.GetBytes("RS+ "));
|
||||||
|
else
|
||||||
|
list.AddRange(Encoding.ASCII.GetBytes("RS- "));
|
||||||
|
|
||||||
|
if (Route?.AccName is not { Length: > 0 })
|
||||||
|
{
|
||||||
|
list.AddRange(Encoding.ASCII.GetBytes(accName));
|
||||||
|
list.Add((byte)' ');
|
||||||
|
}
|
||||||
|
|
||||||
|
list.AddRange(sub.Subject);
|
||||||
|
if (sub.Queue is { Length: > 0 } queue)
|
||||||
|
{
|
||||||
|
list.Add((byte)' ');
|
||||||
|
list.AddRange(queue);
|
||||||
|
if (isSubProto)
|
||||||
|
{
|
||||||
|
list.Add((byte)' ');
|
||||||
|
list.AddRange(Encoding.ASCII.GetBytes(Math.Max(sub.Qw, 1).ToString()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
list.Add((byte)'\r');
|
||||||
|
list.Add((byte)'\n');
|
||||||
|
return list.ToArray();
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void SendRouteSubProtos(IReadOnlyList<Subscription> subs, bool trace, Func<Subscription, bool>? filter = null) =>
|
||||||
|
SendRouteSubOrUnSubProtos(subs, isSubProto: true, trace, filter);
|
||||||
|
|
||||||
|
internal void SendRouteUnSubProtos(IReadOnlyList<Subscription> subs, bool trace, Func<Subscription, bool>? filter = null) =>
|
||||||
|
SendRouteSubOrUnSubProtos(subs, isSubProto: false, trace, filter);
|
||||||
|
|
||||||
|
internal void SendRouteSubOrUnSubProtos(
|
||||||
|
IReadOnlyList<Subscription> subs,
|
||||||
|
bool isSubProto,
|
||||||
|
bool trace,
|
||||||
|
Func<Subscription, bool>? filter = null)
|
||||||
|
{
|
||||||
|
var buf = Array.Empty<byte>();
|
||||||
|
foreach (var sub in subs)
|
||||||
|
{
|
||||||
|
if (filter is not null && !filter(sub))
|
||||||
|
continue;
|
||||||
|
|
||||||
|
var accountName = ServerConstants.DefaultGlobalAccount;
|
||||||
|
|
||||||
|
var startLen = buf.Length;
|
||||||
|
buf = AddRouteSubOrUnsubProtoToBuf(buf, accountName, sub, isSubProto);
|
||||||
|
if (trace && buf.Length > startLen)
|
||||||
|
TraceOutOp(string.Empty, buf.AsSpan(startLen, buf.Length - startLen - 2).ToArray());
|
||||||
|
}
|
||||||
|
|
||||||
|
if (buf.Length > 0)
|
||||||
|
EnqueueProto(buf);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal bool ImportFilter(string subject) => CanImport(subject);
|
||||||
|
|
||||||
|
internal bool IsSolicitedRoute() => Route?.DidSolicit == true;
|
||||||
|
|
||||||
|
internal Exception? ProcessRouteConnect(byte[] arg)
|
||||||
|
{
|
||||||
|
if (arg is not { Length: > 0 })
|
||||||
|
return new FormatException("processRouteConnect parse error");
|
||||||
|
|
||||||
|
ConnectInfo? info;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
info = JsonSerializer.Deserialize<ConnectInfo>(arg);
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
return ex;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (info is null)
|
||||||
|
return new FormatException("processRouteConnect missing CONNECT payload");
|
||||||
|
|
||||||
|
lock (_mu)
|
||||||
|
{
|
||||||
|
Opts.Name = info.Name;
|
||||||
|
Opts.Headers = info.Headers;
|
||||||
|
Route ??= new Route();
|
||||||
|
Route.Lnoc = info.Lnoc;
|
||||||
|
Route.Lnocu = info.Lnocu;
|
||||||
|
}
|
||||||
|
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -112,6 +112,7 @@ public sealed partial class ClientConnection
|
|||||||
|
|
||||||
// Client options (from CONNECT message).
|
// Client options (from CONNECT message).
|
||||||
internal ClientOptions Opts = ClientOptions.Default;
|
internal ClientOptions Opts = ClientOptions.Default;
|
||||||
|
internal Route? Route;
|
||||||
|
|
||||||
// Flags and state.
|
// Flags and state.
|
||||||
internal ClientFlags Flags; // mirrors c.flags clientFlag
|
internal ClientFlags Flags; // mirrors c.flags clientFlag
|
||||||
@@ -1686,7 +1687,7 @@ public sealed partial class ClientConnection
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
internal void ProcessUnsub(byte[] sid)
|
internal void RemoveSubBySid(byte[] sid)
|
||||||
{
|
{
|
||||||
lock (_mu)
|
lock (_mu)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -0,0 +1,248 @@
|
|||||||
|
// Copyright 2012-2026 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
|
||||||
|
using System.Security.Cryptography;
|
||||||
|
using System.Text;
|
||||||
|
using System.Linq;
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server;
|
||||||
|
|
||||||
|
public sealed partial class NatsServer
|
||||||
|
{
|
||||||
|
internal byte[] GenerateRouteInitialInfoJSON(string accName, string compression, int poolIdx, byte gossipMode)
|
||||||
|
{
|
||||||
|
var info = _routeInfo.ShallowClone();
|
||||||
|
Span<byte> nonce = stackalloc byte[16];
|
||||||
|
RandomNumberGenerator.Fill(nonce);
|
||||||
|
info.Nonce = Convert.ToBase64String(nonce);
|
||||||
|
info.RouteAccount = string.IsNullOrEmpty(accName) ? null : accName;
|
||||||
|
info.RoutePoolIdx = poolIdx;
|
||||||
|
info.GossipMode = gossipMode;
|
||||||
|
info.Compression = CompressionModeForInfoProtocol(GetOpts().Cluster.Compression, compression);
|
||||||
|
return GenerateInfoJson(info);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal bool AddRoute(ClientConnection route, bool didSolicit, bool sendDelayedInfo, byte gossipMode, ServerInfo info, string accName)
|
||||||
|
{
|
||||||
|
_mu.EnterWriteLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
route.Route ??= new Route();
|
||||||
|
route.Route.RemoteId = info.Id;
|
||||||
|
route.Route.DidSolicit = didSolicit;
|
||||||
|
route.Route.RemoteName = info.Name;
|
||||||
|
if (!string.IsNullOrEmpty(accName))
|
||||||
|
route.Route.AccName = Encoding.ASCII.GetBytes(accName);
|
||||||
|
|
||||||
|
if (!_routes.TryGetValue(info.Id, out var pool))
|
||||||
|
{
|
||||||
|
pool = [];
|
||||||
|
_routes[info.Id] = pool;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
foreach (var existing in pool.ToArray())
|
||||||
|
{
|
||||||
|
if (!RouteHandler.HandleDuplicateRoute(existing, route))
|
||||||
|
return false;
|
||||||
|
pool.Remove(existing);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
route.Route.PoolIdx = pool.Count;
|
||||||
|
pool.Add(route);
|
||||||
|
|
||||||
|
if (sendDelayedInfo)
|
||||||
|
route.EnqueueProto(GenerateRouteInitialInfoJSON(accName, info.Compression ?? string.Empty, route.Route.PoolIdx, gossipMode));
|
||||||
|
|
||||||
|
ForwardNewRouteInfoToKnownServers(info, didSolicit ? RouteType.Explicit : RouteType.Implicit, didSolicit, gossipMode);
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitWriteLock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? StartRouteAcceptLoop()
|
||||||
|
{
|
||||||
|
if (_routeListener == null)
|
||||||
|
return null;
|
||||||
|
|
||||||
|
if (!StartGoRoutine(() => Noticef("Route accept loop started")))
|
||||||
|
return new InvalidOperationException("unable to start route accept loop");
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? SetRouteInfoHostPortAndIP()
|
||||||
|
{
|
||||||
|
var opts = GetOpts();
|
||||||
|
string host;
|
||||||
|
int port;
|
||||||
|
if (!string.IsNullOrWhiteSpace(opts.Cluster.Advertise))
|
||||||
|
{
|
||||||
|
var (advHost, advPort, advErr) = Internal.ServerUtilities.ParseHostPort(opts.Cluster.Advertise, opts.Cluster.Port);
|
||||||
|
if (advErr != null)
|
||||||
|
return new InvalidOperationException($"Cluster.Advertise invalid: {opts.Cluster.Advertise}", advErr);
|
||||||
|
host = advHost;
|
||||||
|
port = advPort;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
host = opts.Cluster.Host;
|
||||||
|
if (string.IsNullOrWhiteSpace(host))
|
||||||
|
host = opts.Host;
|
||||||
|
port = opts.Cluster.Port;
|
||||||
|
}
|
||||||
|
|
||||||
|
_mu.EnterWriteLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
_routeInfo.Host = host;
|
||||||
|
_routeInfo.Port = port;
|
||||||
|
_routeInfo.Ip = $"nats-route://{host}:{port}/";
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitWriteLock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public Exception? StartRouting()
|
||||||
|
{
|
||||||
|
var err = SetRouteInfoHostPortAndIP();
|
||||||
|
if (err != null)
|
||||||
|
return err;
|
||||||
|
SolicitRoutes();
|
||||||
|
return StartRouteAcceptLoop();
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void ReConnectToRoute(Uri routeUrl, string accName = "")
|
||||||
|
{
|
||||||
|
StartGoRoutine(() =>
|
||||||
|
{
|
||||||
|
_ = ConnectToRoute(routeUrl, RouteType.Explicit, false, GossipMode.Default, accName);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
internal bool RouteStillValid(Uri routeUrl)
|
||||||
|
{
|
||||||
|
var opts = GetOpts();
|
||||||
|
return opts.Routes.Any(r => string.Equals(r.Host, routeUrl.Host, StringComparison.OrdinalIgnoreCase) && r.Port == routeUrl.Port);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal Exception? ConnectToRoute(Uri routeUrl, RouteType routeType, bool firstConnect, byte gossipMode, string accName)
|
||||||
|
{
|
||||||
|
_ = firstConnect;
|
||||||
|
if (!RouteStillValid(routeUrl))
|
||||||
|
return new InvalidOperationException($"route is no longer configured: {routeUrl}");
|
||||||
|
|
||||||
|
SaveRouteTLSName(routeUrl);
|
||||||
|
var route = CreateRoute(null, routeUrl, routeType, gossipMode, accName);
|
||||||
|
if (route is null)
|
||||||
|
return new InvalidOperationException("failed to create route");
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal bool SaveRouteTLSName(Uri routeUrl)
|
||||||
|
{
|
||||||
|
if (routeUrl is null || string.IsNullOrWhiteSpace(routeUrl.Host))
|
||||||
|
return false;
|
||||||
|
|
||||||
|
_mu.EnterWriteLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
_routeTlsName = routeUrl.Host;
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitWriteLock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void SolicitRoutes()
|
||||||
|
{
|
||||||
|
foreach (var route in GetOpts().Routes)
|
||||||
|
_ = ConnectToRoute(route, RouteType.Explicit, true, GossipMode.Default, string.Empty);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void RemoveAllRoutesExcept(string remoteId)
|
||||||
|
{
|
||||||
|
_mu.EnterWriteLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
foreach (var (rid, routes) in _routes.ToArray())
|
||||||
|
{
|
||||||
|
if (rid == remoteId)
|
||||||
|
continue;
|
||||||
|
|
||||||
|
foreach (var route in routes)
|
||||||
|
route.CloseConnection(ClosedState.RouteRemoved);
|
||||||
|
_routes.Remove(rid);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitWriteLock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
internal bool IsDuplicateServerName(string serverName)
|
||||||
|
{
|
||||||
|
var duplicate = false;
|
||||||
|
_mu.EnterReadLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
ForEachRoute(route =>
|
||||||
|
{
|
||||||
|
if (route.Route?.RemoteName == serverName)
|
||||||
|
duplicate = true;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitReadLock();
|
||||||
|
}
|
||||||
|
return duplicate;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void ForEachNonPerAccountRoute(Func<ClientConnection, bool> fn)
|
||||||
|
{
|
||||||
|
_mu.EnterReadLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
foreach (var route in _routes.Values.SelectMany(v => v))
|
||||||
|
{
|
||||||
|
if (route.Route?.AccName is { Length: > 0 })
|
||||||
|
continue;
|
||||||
|
if (!fn(route))
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitReadLock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void ForEachRouteIdx(int idx, Func<ClientConnection, bool> fn)
|
||||||
|
{
|
||||||
|
_mu.EnterReadLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
foreach (var pool in _routes.Values)
|
||||||
|
{
|
||||||
|
if (idx < 0 || idx >= pool.Count)
|
||||||
|
continue;
|
||||||
|
if (!fn(pool[idx]))
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitReadLock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,183 @@
|
|||||||
|
// Copyright 2012-2026 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
|
||||||
|
using System.Text.Json;
|
||||||
|
using ZB.MOM.NatsNet.Server.Auth;
|
||||||
|
using ZB.MOM.NatsNet.Server.Internal;
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server;
|
||||||
|
|
||||||
|
public sealed partial class NatsServer
|
||||||
|
{
|
||||||
|
internal bool NegotiateRouteCompression(
|
||||||
|
ClientConnection c,
|
||||||
|
bool didSolicit,
|
||||||
|
string accName,
|
||||||
|
string infoCompression,
|
||||||
|
ServerOptions opts)
|
||||||
|
{
|
||||||
|
var mode = SelectCompressionMode(opts.Cluster.Compression.Mode, infoCompression);
|
||||||
|
|
||||||
|
lock (c)
|
||||||
|
{
|
||||||
|
c.Route ??= new Route();
|
||||||
|
if (mode == CompressionMode.S2Auto)
|
||||||
|
{
|
||||||
|
if (c.Rtt == TimeSpan.Zero)
|
||||||
|
c.Rtt = ClientConnection.ComputeRtt(c.Start);
|
||||||
|
mode = SelectS2AutoModeBasedOnRtt(c.Rtt, opts.Cluster.Compression.RttThresholds);
|
||||||
|
}
|
||||||
|
c.Route.Compression = mode;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!NeedsCompression(mode))
|
||||||
|
return false;
|
||||||
|
|
||||||
|
var info = CopyInfo();
|
||||||
|
info.Compression = CompressionModeForInfoProtocol(opts.Cluster.Compression, mode);
|
||||||
|
if (!string.IsNullOrEmpty(accName))
|
||||||
|
info.RouteAccount = accName;
|
||||||
|
var proto = GenerateInfoJson(info);
|
||||||
|
|
||||||
|
lock (c)
|
||||||
|
{
|
||||||
|
if (didSolicit)
|
||||||
|
c.EnqueueProto(proto);
|
||||||
|
else
|
||||||
|
c.EnqueueProto(proto);
|
||||||
|
c.SetFirstPingTimer();
|
||||||
|
}
|
||||||
|
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void UpdateRemoteRoutePerms(ClientConnection c, ServerInfo info)
|
||||||
|
{
|
||||||
|
SubjectPermission? oldExport;
|
||||||
|
lock (c)
|
||||||
|
{
|
||||||
|
oldExport = c.Opts.Export?.Clone();
|
||||||
|
c.Opts.Import = info.Import?.Clone();
|
||||||
|
c.Opts.Export = info.Export?.Clone();
|
||||||
|
}
|
||||||
|
|
||||||
|
// Build old/new export checkers to preserve route permission semantics.
|
||||||
|
var oldTester = new ClientConnection(ClientKind.Router) { Route = new Route() };
|
||||||
|
oldTester.SetRoutePermissions(new RoutePermissions { Export = oldExport });
|
||||||
|
var newTester = new ClientConnection(ClientKind.Router) { Route = new Route() };
|
||||||
|
newTester.SetRoutePermissions(new RoutePermissions { Export = info.Export?.Clone() });
|
||||||
|
|
||||||
|
if (oldExport is null && info.Export is null)
|
||||||
|
return;
|
||||||
|
|
||||||
|
// Subscription fanout wiring for route permission delta is completed in group 2.
|
||||||
|
_ = oldTester;
|
||||||
|
_ = newTester;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void ProcessImplicitRoute(ServerInfo info, bool routeNoPool)
|
||||||
|
{
|
||||||
|
var remoteId = info.Id;
|
||||||
|
_mu.EnterWriteLock();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
if (remoteId == _info.Id)
|
||||||
|
return;
|
||||||
|
|
||||||
|
var opts = GetOpts();
|
||||||
|
if (!string.IsNullOrEmpty(info.RouteAccount))
|
||||||
|
{
|
||||||
|
if (opts.Cluster.PoolSize <= 0)
|
||||||
|
return;
|
||||||
|
if (_accRoutes != null
|
||||||
|
&& _accRoutes.TryGetValue(info.RouteAccount, out var remotes)
|
||||||
|
&& remotes.TryGetValue(remoteId, out var existing)
|
||||||
|
&& existing != null)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
else if (_routes.ContainsKey(remoteId))
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (HasThisRouteConfigured(info))
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_mu.ExitWriteLock();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!Uri.TryCreate(info.Ip, UriKind.Absolute, out var _))
|
||||||
|
return;
|
||||||
|
|
||||||
|
if (routeNoPool && string.IsNullOrEmpty(info.RouteAccount))
|
||||||
|
Debugf("Implicit route from non-pooling remote {0} processed", info.Id);
|
||||||
|
}
|
||||||
|
|
||||||
|
internal bool HasThisRouteConfigured(ServerInfo info)
|
||||||
|
{
|
||||||
|
var routes = GetOpts().Routes;
|
||||||
|
if (routes.Count == 0)
|
||||||
|
return false;
|
||||||
|
|
||||||
|
var infoPort = info.Port <= 0 ? 6222 : info.Port;
|
||||||
|
var primary = $"{info.Host}:{infoPort}".ToLowerInvariant();
|
||||||
|
|
||||||
|
var secondary = string.Empty;
|
||||||
|
if (!string.IsNullOrWhiteSpace(info.Ip) && Uri.TryCreate(info.Ip, UriKind.Absolute, out var infoUri))
|
||||||
|
{
|
||||||
|
var host = infoUri.Host;
|
||||||
|
if (!string.IsNullOrWhiteSpace(host))
|
||||||
|
secondary = $"{host}:{infoPort}".ToLowerInvariant();
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var route in routes)
|
||||||
|
{
|
||||||
|
var routePort = route.IsDefaultPort ? infoPort : route.Port;
|
||||||
|
var hostPort = $"{route.Host}:{routePort}".ToLowerInvariant();
|
||||||
|
if (hostPort == primary || (!string.IsNullOrEmpty(secondary) && hostPort == secondary))
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal void ForwardNewRouteInfoToKnownServers(ServerInfo info, RouteType routeType, bool didSolicit, byte localGossipMode)
|
||||||
|
{
|
||||||
|
var fromGossip = didSolicit && routeType == RouteType.Implicit;
|
||||||
|
if ((fromGossip && localGossipMode != GossipMode.Override) || info.GossipMode == GossipMode.Disabled)
|
||||||
|
return;
|
||||||
|
|
||||||
|
info.Nonce = string.Empty;
|
||||||
|
|
||||||
|
byte[] BuildInfo(byte gossipMode)
|
||||||
|
{
|
||||||
|
info.GossipMode = gossipMode;
|
||||||
|
return GenerateInfoJson(info);
|
||||||
|
}
|
||||||
|
|
||||||
|
byte[]? infoDefault = null;
|
||||||
|
byte[]? infoDisabled = null;
|
||||||
|
byte[]? infoOverride = null;
|
||||||
|
|
||||||
|
byte[] SelectInfo(ClientConnection route)
|
||||||
|
{
|
||||||
|
var rType = route.Route?.RouteType ?? RouteType.Implicit;
|
||||||
|
if ((!didSolicit && rType == RouteType.Explicit) || (didSolicit && routeType == RouteType.Explicit))
|
||||||
|
return infoOverride ??= BuildInfo(GossipMode.Override);
|
||||||
|
if (!didSolicit)
|
||||||
|
return infoDisabled ??= BuildInfo(GossipMode.Disabled);
|
||||||
|
return infoDefault ??= BuildInfo(GossipMode.Default);
|
||||||
|
}
|
||||||
|
|
||||||
|
ForEachRemote(route =>
|
||||||
|
{
|
||||||
|
if (route.Route?.RemoteId == info.Id)
|
||||||
|
return;
|
||||||
|
route.EnqueueProto(SelectInfo(route));
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,74 @@
|
|||||||
|
// Copyright 2012-2026 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
|
||||||
|
using ZB.MOM.NatsNet.Server.Internal;
|
||||||
|
using System.IO;
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server;
|
||||||
|
|
||||||
|
public sealed partial class NatsServer
|
||||||
|
{
|
||||||
|
internal void SendSubsToRoute(ClientConnection route, int idx, string account)
|
||||||
|
{
|
||||||
|
if (route == null)
|
||||||
|
return;
|
||||||
|
|
||||||
|
var allSubs = new List<Subscription>(1024);
|
||||||
|
if (idx < 0 || !string.IsNullOrEmpty(account))
|
||||||
|
{
|
||||||
|
var (acc, _) = LookupAccount(account);
|
||||||
|
acc?.Sublist?.LocalSubs(allSubs, includeLeafHubs: false);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
foreach (var acc in _accounts.Values)
|
||||||
|
{
|
||||||
|
if (acc.RoutePoolIdx != idx)
|
||||||
|
continue;
|
||||||
|
acc.Sublist?.LocalSubs(allSubs, includeLeafHubs: false);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
route.SendRouteSubProtos(allSubs, trace: false, sub => route.CanImport(System.Text.Encoding.ASCII.GetString(sub.Subject)));
|
||||||
|
}
|
||||||
|
|
||||||
|
internal ClientConnection? CreateRoute(Stream? conn, Uri? routeUrl, RouteType routeType, byte gossipMode, string accName)
|
||||||
|
{
|
||||||
|
var opts = GetOpts();
|
||||||
|
var didSolicit = routeUrl != null;
|
||||||
|
var c = new ClientConnection(ClientKind.Router, this, conn ?? new MemoryStream())
|
||||||
|
{
|
||||||
|
Opts = ClientOptions.Default,
|
||||||
|
Route = new Route
|
||||||
|
{
|
||||||
|
Url = routeUrl,
|
||||||
|
RouteType = routeType,
|
||||||
|
DidSolicit = didSolicit,
|
||||||
|
PoolIdx = -1,
|
||||||
|
GossipMode = gossipMode,
|
||||||
|
AccName = string.IsNullOrEmpty(accName) ? null : System.Text.Encoding.ASCII.GetBytes(accName),
|
||||||
|
},
|
||||||
|
Start = DateTime.UtcNow,
|
||||||
|
};
|
||||||
|
|
||||||
|
lock (c)
|
||||||
|
{
|
||||||
|
c.InitClient();
|
||||||
|
if (didSolicit)
|
||||||
|
c.SetRoutePermissions(opts.Cluster.Permissions);
|
||||||
|
c.SetFirstPingTimer();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (didSolicit)
|
||||||
|
{
|
||||||
|
var sendErr = c.SendRouteConnect(_info.Cluster ?? string.Empty, _routeInfo.TlsRequired);
|
||||||
|
if (sendErr != null)
|
||||||
|
{
|
||||||
|
c.CloseConnection(ClosedState.ProtocolViolation);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return c;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -254,6 +254,7 @@ public sealed partial class NatsServer : INatsServer
|
|||||||
private readonly ConcurrentDictionary<string, object?> _nodeToInfo = new(StringComparer.Ordinal);
|
private readonly ConcurrentDictionary<string, object?> _nodeToInfo = new(StringComparer.Ordinal);
|
||||||
private readonly ConcurrentDictionary<string, object?> _raftNodes = new(StringComparer.Ordinal);
|
private readonly ConcurrentDictionary<string, object?> _raftNodes = new(StringComparer.Ordinal);
|
||||||
private readonly Dictionary<string, string> _routesToSelf = [];
|
private readonly Dictionary<string, string> _routesToSelf = [];
|
||||||
|
private string _routeTlsName = string.Empty;
|
||||||
private INetResolver? _routeResolver;
|
private INetResolver? _routeResolver;
|
||||||
private readonly ConcurrentDictionary<string, object?> _rateLimitLogging = new();
|
private readonly ConcurrentDictionary<string, object?> _rateLimitLogging = new();
|
||||||
private readonly Channel<TimeSpan> _rateLimitLoggingCh;
|
private readonly Channel<TimeSpan> _rateLimitLoggingCh;
|
||||||
|
|||||||
@@ -0,0 +1,60 @@
|
|||||||
|
// Copyright 2012-2026 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server;
|
||||||
|
|
||||||
|
internal static class RouteHandler
|
||||||
|
{
|
||||||
|
internal static int ComputeRoutePoolIdx(int poolSize, string accountName) =>
|
||||||
|
NatsServer.ComputeRoutePoolIdx(poolSize, accountName);
|
||||||
|
|
||||||
|
internal static string GetAccNameFromRoutedSubKey(Internal.Subscription sub, string key, bool keyHasSubType)
|
||||||
|
{
|
||||||
|
_ = sub;
|
||||||
|
var fields = key.Split(' ', StringSplitOptions.RemoveEmptyEntries);
|
||||||
|
if (fields.Length == 0)
|
||||||
|
return string.Empty;
|
||||||
|
|
||||||
|
var accountIndex = keyHasSubType ? 1 : 0;
|
||||||
|
if (accountIndex >= fields.Length)
|
||||||
|
return string.Empty;
|
||||||
|
|
||||||
|
return fields[accountIndex];
|
||||||
|
}
|
||||||
|
|
||||||
|
internal static bool RouteShouldDelayInfo(string accName, ServerOptions opts) =>
|
||||||
|
string.IsNullOrEmpty(accName) && opts.Cluster.PoolSize >= 1;
|
||||||
|
|
||||||
|
internal static bool HasSolicitedRoute(IReadOnlyList<ClientConnection> routes, string accName)
|
||||||
|
{
|
||||||
|
foreach (var route in routes)
|
||||||
|
{
|
||||||
|
if (route.Route?.DidSolicit != true)
|
||||||
|
continue;
|
||||||
|
|
||||||
|
var routeAcc = route.Route?.AccName is { Length: > 0 } an
|
||||||
|
? System.Text.Encoding.ASCII.GetString(an)
|
||||||
|
: string.Empty;
|
||||||
|
if (routeAcc == accName)
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal static void UpgradeRouteToSolicited(ClientConnection route)
|
||||||
|
{
|
||||||
|
if (route.Route is null)
|
||||||
|
return;
|
||||||
|
route.Route.DidSolicit = true;
|
||||||
|
route.Route.Retry = true;
|
||||||
|
}
|
||||||
|
|
||||||
|
internal static bool HandleDuplicateRoute(ClientConnection existing, ClientConnection incoming)
|
||||||
|
{
|
||||||
|
if (existing.IsSolicitedRoute() && !incoming.IsSolicitedRoute())
|
||||||
|
return false;
|
||||||
|
if (!existing.IsSolicitedRoute() && incoming.IsSolicitedRoute())
|
||||||
|
return true;
|
||||||
|
return incoming.Cid > existing.Cid;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -308,7 +308,9 @@ public sealed class ClientConnectionStubFeaturesTests
|
|||||||
GetTimer(c, "_pingTimer").ShouldNotBeNull();
|
GetTimer(c, "_pingTimer").ShouldNotBeNull();
|
||||||
|
|
||||||
c.WatchForStaleConnection(TimeSpan.FromMilliseconds(20), pingMax: 0);
|
c.WatchForStaleConnection(TimeSpan.FromMilliseconds(20), pingMax: 0);
|
||||||
Thread.Sleep(60);
|
var staleDeadline = DateTime.UtcNow.AddMilliseconds(500);
|
||||||
|
while (!c.IsClosed() && DateTime.UtcNow < staleDeadline)
|
||||||
|
Thread.Sleep(10);
|
||||||
c.IsClosed().ShouldBeTrue();
|
c.IsClosed().ShouldBeTrue();
|
||||||
|
|
||||||
var temp = Account.NewAccount("A");
|
var temp = Account.NewAccount("A");
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
using System.IO;
|
using System.IO;
|
||||||
|
using System.Text;
|
||||||
using Shouldly;
|
using Shouldly;
|
||||||
using ZB.MOM.NatsNet.Server;
|
using ZB.MOM.NatsNet.Server;
|
||||||
using ZB.MOM.NatsNet.Server.Internal;
|
using ZB.MOM.NatsNet.Server.Internal;
|
||||||
@@ -7,6 +8,65 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
|
|||||||
|
|
||||||
public sealed partial class RouteHandlerTests
|
public sealed partial class RouteHandlerTests
|
||||||
{
|
{
|
||||||
|
[Fact] // T:2798
|
||||||
|
public void ClusterAdvertiseErrorOnStartup_ShouldSucceed()
|
||||||
|
{
|
||||||
|
var options = new ServerOptions();
|
||||||
|
options.Cluster.Advertise = "addr:::123";
|
||||||
|
var (server, err) = NatsServer.NewServer(options);
|
||||||
|
err.ShouldBeNull();
|
||||||
|
var startErr = server!.StartRouting();
|
||||||
|
startErr.ShouldNotBeNull();
|
||||||
|
startErr!.Message.ShouldContain("Cluster.Advertise");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact] // T:2822
|
||||||
|
public void TLSRoutesCertificateImplicitAllowPass_ShouldSucceed()
|
||||||
|
{
|
||||||
|
var client = new ClientConnection(ClientKind.Router, nc: new MemoryStream());
|
||||||
|
client.MatchesPinnedCert(null).ShouldBeTrue();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact] // T:2823
|
||||||
|
public void TLSRoutesCertificateImplicitAllowFail_ShouldSucceed()
|
||||||
|
{
|
||||||
|
var client = new ClientConnection(ClientKind.Router, nc: new MemoryStream());
|
||||||
|
var pinned = new PinnedCertSet([new string('a', 64)]);
|
||||||
|
client.MatchesPinnedCert(pinned).ShouldBeFalse();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact] // T:2844
|
||||||
|
public void RouteParseOriginClusterMsgArgs_ShouldSucceed()
|
||||||
|
{
|
||||||
|
var c = new ClientConnection(ClientKind.Router)
|
||||||
|
{
|
||||||
|
Route = new Route { AccName = "MY_ACCOUNT"u8.ToArray() },
|
||||||
|
};
|
||||||
|
|
||||||
|
var err = c.ProcessRoutedOriginClusterMsgArgs("ORIGIN foo + bar queue1 queue2 12 345\r\n"u8.ToArray());
|
||||||
|
err.ShouldBeNull();
|
||||||
|
Encoding.ASCII.GetString(c.ParseCtx.Pa.Account!).ShouldBe("ORIGIN");
|
||||||
|
Encoding.ASCII.GetString(c.ParseCtx.Pa.Subject!).ShouldBe("foo");
|
||||||
|
Encoding.ASCII.GetString(c.ParseCtx.Pa.Reply!).ShouldBe("bar");
|
||||||
|
c.ParseCtx.Pa.Queues.ShouldNotBeNull();
|
||||||
|
c.ParseCtx.Pa.Queues!.Count.ShouldBe(3);
|
||||||
|
c.ParseCtx.Pa.Size.ShouldBe(345);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact] // T:2850
|
||||||
|
public void RouteCompression_ShouldSucceed()
|
||||||
|
{
|
||||||
|
var opts = new ServerOptions();
|
||||||
|
opts.Cluster.Compression.Mode = CompressionMode.S2Fast;
|
||||||
|
var (server, err) = NatsServer.NewServer(opts);
|
||||||
|
err.ShouldBeNull();
|
||||||
|
server.ShouldNotBeNull();
|
||||||
|
|
||||||
|
var infoProto = server!.GenerateRouteInitialInfoJSON(string.Empty, CompressionMode.S2Fast, 0, GossipMode.Default);
|
||||||
|
infoProto.Length.ShouldBeGreaterThan(0);
|
||||||
|
Encoding.ASCII.GetString(infoProto).ShouldContain("\"compression\":\"s2_fast\"");
|
||||||
|
}
|
||||||
|
|
||||||
[Fact] // T:2819
|
[Fact] // T:2819
|
||||||
public async Task RouteIPResolutionAndRouteToSelf_ShouldSucceed()
|
public async Task RouteIPResolutionAndRouteToSelf_ShouldSucceed()
|
||||||
{
|
{
|
||||||
|
|||||||
BIN
Binary file not shown.
+6
-6
@@ -1,6 +1,6 @@
|
|||||||
# NATS .NET Porting Status Report
|
# NATS .NET Porting Status Report
|
||||||
|
|
||||||
Generated: 2026-03-01 01:56:16 UTC
|
Generated: 2026-03-01 02:33:59 UTC
|
||||||
|
|
||||||
## Modules (12 total)
|
## Modules (12 total)
|
||||||
|
|
||||||
@@ -13,18 +13,18 @@ Generated: 2026-03-01 01:56:16 UTC
|
|||||||
| Status | Count |
|
| Status | Count |
|
||||||
|--------|-------|
|
|--------|-------|
|
||||||
| complete | 22 |
|
| complete | 22 |
|
||||||
| deferred | 1519 |
|
| deferred | 1467 |
|
||||||
| n_a | 24 |
|
| n_a | 24 |
|
||||||
| stub | 1 |
|
| stub | 1 |
|
||||||
| verified | 2107 |
|
| verified | 2159 |
|
||||||
|
|
||||||
## Unit Tests (3257 total)
|
## Unit Tests (3257 total)
|
||||||
|
|
||||||
| Status | Count |
|
| Status | Count |
|
||||||
|--------|-------|
|
|--------|-------|
|
||||||
| deferred | 1595 |
|
| deferred | 1590 |
|
||||||
| n_a | 254 |
|
| n_a | 254 |
|
||||||
| verified | 1408 |
|
| verified | 1413 |
|
||||||
|
|
||||||
## Library Mappings (36 total)
|
## Library Mappings (36 total)
|
||||||
|
|
||||||
@@ -35,4 +35,4 @@ Generated: 2026-03-01 01:56:16 UTC
|
|||||||
|
|
||||||
## Overall Progress
|
## Overall Progress
|
||||||
|
|
||||||
**3827/6942 items complete (55.1%)**
|
**3884/6942 items complete (55.9%)**
|
||||||
|
|||||||
Reference in New Issue
Block a user