Compare commits
7 Commits
2b187a4c90
...
1d42fe4499
| Author | SHA1 | Date | |
|---|---|---|---|
| 1d42fe4499 | |||
| 45ac99b6cc | |||
| 027818b6b0 | |||
| eb05308b5a | |||
| 108d06dd57 | |||
| f8d384711d | |||
| e16d7ffab3 |
@@ -638,6 +638,14 @@ public sealed partial class NatsServer
|
||||
return (claims, claimJwt, null);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fetches an account from the resolver, registers it, and returns it.
|
||||
/// Mirrors Go <c>Server.fetchAccount</c>.
|
||||
/// Lock must NOT be held on entry.
|
||||
/// </summary>
|
||||
public (Account? Account, Exception? Error) FetchAccount(string name) =>
|
||||
FetchAccountFromResolver(name);
|
||||
|
||||
/// <summary>
|
||||
/// Fetches an account from the resolver, registers it, and returns it.
|
||||
/// Mirrors Go <c>Server.fetchAccount</c>.
|
||||
|
||||
@@ -265,6 +265,22 @@ public sealed partial class NatsServer
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns S2 writer options for the selected route compression mode.
|
||||
/// Mirrors Go <c>s2WriterOptions</c>.
|
||||
/// </summary>
|
||||
internal static string[]? S2WriterOptions(string cm)
|
||||
{
|
||||
var opts = new List<string> { "writer_concurrency=1" };
|
||||
return cm switch
|
||||
{
|
||||
CompressionMode.S2Uncompressed => [.. opts, "writer_uncompressed"],
|
||||
CompressionMode.S2Best => [.. opts, "writer_best_compression"],
|
||||
CompressionMode.S2Better => [.. opts, "writer_better_compression"],
|
||||
_ => null,
|
||||
};
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// Factory methods (features 2983–2985)
|
||||
// =========================================================================
|
||||
@@ -509,20 +525,27 @@ public sealed partial class NatsServer
|
||||
/// Background loop that logs TLS rate-limited connection rejections every second.
|
||||
/// Mirrors Go <c>Server.logRejectedTLSConns</c>.
|
||||
/// </summary>
|
||||
internal async Task LogRejectedTlsConnsAsync(CancellationToken ct)
|
||||
internal Task LogRejectedTlsConnsAsync(CancellationToken ct) =>
|
||||
LogRejectedTLSConns(ct);
|
||||
|
||||
/// <summary>
|
||||
/// Background loop that logs TLS rate-limited connection rejections every second.
|
||||
/// Mirrors Go <c>Server.logRejectedTLSConns</c>.
|
||||
/// </summary>
|
||||
internal async Task LogRejectedTLSConns(CancellationToken ct, TimeSpan? interval = null)
|
||||
{
|
||||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
|
||||
using var timer = new PeriodicTimer(interval ?? TimeSpan.FromSeconds(1));
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
try { await timer.WaitForNextTickAsync(ct); }
|
||||
catch (OperationCanceledException) { break; }
|
||||
|
||||
if (_connRateCounter is not null)
|
||||
{
|
||||
var blocked = _connRateCounter.CountBlocked();
|
||||
if (blocked > 0)
|
||||
Warnf("Rejected {0} connections due to TLS rate limiting", blocked);
|
||||
}
|
||||
|
||||
try { await timer.WaitForNextTickAsync(ct); }
|
||||
catch (OperationCanceledException) { break; }
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -178,7 +178,13 @@ public sealed partial class NatsServer
|
||||
/// Returns true if the goroutine was started, false if the server is already stopped.
|
||||
/// Mirrors Go <c>Server.startGoRoutine(f)</c>.
|
||||
/// </summary>
|
||||
internal bool StartGoRoutine(Action f)
|
||||
internal bool StartGoRoutine(Action f) => StartGoRoutine(f, []);
|
||||
|
||||
/// <summary>
|
||||
/// Starts a background Task with goroutine labels.
|
||||
/// Mirrors Go <c>Server.startGoRoutine(f, tags...)</c>.
|
||||
/// </summary>
|
||||
internal bool StartGoRoutine(Action f, params IReadOnlyDictionary<string, string>[] tags)
|
||||
{
|
||||
lock (_grMu)
|
||||
{
|
||||
@@ -186,13 +192,41 @@ public sealed partial class NatsServer
|
||||
_grWg.Add(1);
|
||||
Task.Run(() =>
|
||||
{
|
||||
try { f(); }
|
||||
try
|
||||
{
|
||||
SetGoRoutineLabels(tags);
|
||||
f();
|
||||
}
|
||||
finally { _grWg.Done(); }
|
||||
});
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Optional test-only sink used to observe goroutine labels during unit tests.
|
||||
/// </summary>
|
||||
internal static Action<IReadOnlyList<KeyValuePair<string, string>>>? SetGoRoutineLabelsHookForTest { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Sets goroutine labels for diagnostics when tags are present.
|
||||
/// Mirrors Go <c>setGoRoutineLabels</c>.
|
||||
/// </summary>
|
||||
internal static void SetGoRoutineLabels(params IReadOnlyDictionary<string, string>[] tags)
|
||||
{
|
||||
var labels = new List<KeyValuePair<string, string>>();
|
||||
foreach (var tag in tags)
|
||||
{
|
||||
foreach (var pair in tag)
|
||||
{
|
||||
labels.Add(new KeyValuePair<string, string>(pair.Key, pair.Value));
|
||||
}
|
||||
}
|
||||
|
||||
if (labels.Count > 0)
|
||||
SetGoRoutineLabelsHookForTest?.Invoke(labels);
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// Client / connection management (features 3081–3084)
|
||||
// =========================================================================
|
||||
@@ -297,10 +331,12 @@ public sealed partial class NatsServer
|
||||
public int NumRemotes()
|
||||
{
|
||||
_mu.EnterReadLock();
|
||||
try { return _routes.Count; }
|
||||
try { return NumRemotesInternal(); }
|
||||
finally { _mu.ExitReadLock(); }
|
||||
}
|
||||
|
||||
private int NumRemotesInternal() => _routes.Count;
|
||||
|
||||
/// <summary>Returns the number of leaf-node connections. Mirrors Go <c>Server.NumLeafNodes()</c>.</summary>
|
||||
public int NumLeafNodes()
|
||||
{
|
||||
@@ -441,27 +477,56 @@ public sealed partial class NatsServer
|
||||
/// Mirrors Go <c>Server.readyForConnections()</c>.
|
||||
/// </summary>
|
||||
public Exception? ReadyForConnectionsError(TimeSpan d)
|
||||
=> ReadyForConnectionsInternal(d);
|
||||
|
||||
/// <summary>
|
||||
/// Polls until all expected listeners are up or the deadline expires.
|
||||
/// Returns an error description if not ready within <paramref name="d"/>.
|
||||
/// Mirrors Go <c>Server.readyForConnections()</c>.
|
||||
/// </summary>
|
||||
internal Exception? ReadyForConnectionsInternal(TimeSpan d)
|
||||
{
|
||||
var opts = GetOpts();
|
||||
var end = DateTime.UtcNow.Add(d);
|
||||
|
||||
var checks = new Dictionary<string, (bool ok, Exception? err)>(StringComparer.Ordinal);
|
||||
while (DateTime.UtcNow < end)
|
||||
{
|
||||
bool serverOk, routeOk, gatewayOk, leafOk, wsOk;
|
||||
bool serverOk, routeOk, gatewayOk, leafOk, wsOk, mqttOk;
|
||||
Exception? serverErr, routeErr, gatewayErr, leafErr;
|
||||
_mu.EnterReadLock();
|
||||
serverOk = _listener != null || opts.DontListen;
|
||||
serverErr = _listenerErr;
|
||||
routeOk = opts.Cluster.Port == 0 || _routeListener != null;
|
||||
routeErr = _routeListenerErr;
|
||||
gatewayOk = string.IsNullOrEmpty(opts.Gateway.Name) || _gatewayListener != null;
|
||||
gatewayErr = _gatewayListenerErr;
|
||||
leafOk = opts.LeafNode.Port == 0 || _leafNodeListener != null;
|
||||
wsOk = opts.Websocket.Port == 0; // stub — websocket listener not tracked until session 23
|
||||
leafErr = _leafNodeListenerErr;
|
||||
wsOk = opts.Websocket.Port == 0;
|
||||
mqttOk = opts.Mqtt.Port == 0;
|
||||
_mu.ExitReadLock();
|
||||
|
||||
if (serverOk && routeOk && gatewayOk && leafOk && wsOk)
|
||||
checks["server"] = (serverOk, serverErr);
|
||||
checks["route"] = (routeOk, routeErr);
|
||||
checks["gateway"] = (gatewayOk, gatewayErr);
|
||||
checks["leafnode"] = (leafOk, leafErr);
|
||||
checks["websocket"] = (wsOk, null);
|
||||
checks["mqtt"] = (mqttOk, null);
|
||||
|
||||
var numOk = checks.Values.Count(v => v.ok);
|
||||
if (numOk == checks.Count)
|
||||
{
|
||||
if (opts.DontListen)
|
||||
{
|
||||
try { _startupComplete.Task.Wait((int)d.TotalMilliseconds); }
|
||||
catch { /* timeout */ }
|
||||
catch { }
|
||||
|
||||
if (!_startupComplete.Task.IsCompleted)
|
||||
{
|
||||
return new InvalidOperationException(
|
||||
$"failed to be ready for connections after {d}: startup did not complete");
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
@@ -470,8 +535,19 @@ public sealed partial class NatsServer
|
||||
Thread.Sleep(25);
|
||||
}
|
||||
|
||||
var failed = new List<string>(checks.Count);
|
||||
foreach (var (name, info) in checks)
|
||||
{
|
||||
if (info.ok && info.err != null)
|
||||
failed.Add($"{name}(ok, but {info.err.Message})");
|
||||
else if (!info.ok && info.err == null)
|
||||
failed.Add(name);
|
||||
else if (!info.ok)
|
||||
failed.Add($"{name}({info.err!.Message})");
|
||||
}
|
||||
|
||||
return new InvalidOperationException(
|
||||
$"failed to be ready for connections after {d}");
|
||||
$"failed to be ready for connections after {d}: {string.Join(", ", failed)}");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -494,7 +570,10 @@ public sealed partial class NatsServer
|
||||
public string Name() => _info.Name;
|
||||
|
||||
/// <summary>Returns the server name as a string. Mirrors Go <c>Server.String()</c>.</summary>
|
||||
public override string ToString() => _info.Name;
|
||||
public string String() => _info.Name;
|
||||
|
||||
/// <summary>Returns the server name as a string. Mirrors Go <c>Server.String()</c>.</summary>
|
||||
public override string ToString() => String();
|
||||
|
||||
/// <summary>Returns the number of currently-stored closed connections. Mirrors Go <c>Server.numClosedConns()</c>.</summary>
|
||||
internal int NumClosedConns()
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
// Session 10: accept loops, client creation, connect URL management, port helpers.
|
||||
|
||||
using System.Net;
|
||||
using System.Net.Security;
|
||||
using System.Net.Sockets;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text.Json;
|
||||
@@ -461,6 +462,41 @@ public sealed partial class NatsServer
|
||||
private string BasePath(string p) =>
|
||||
string.IsNullOrEmpty(_httpBasePath) ? p : System.IO.Path.Combine(_httpBasePath, p.TrimStart('/'));
|
||||
|
||||
/// <summary>
|
||||
/// Returns a monitoring TLS config cloned from server TLS config, with client certs disabled.
|
||||
/// Mirrors Go <c>Server.getMonitoringTLSConfig()</c>.
|
||||
/// </summary>
|
||||
internal SslServerAuthenticationOptions? GetMonitoringTLSConfig()
|
||||
{
|
||||
var opts = GetOpts();
|
||||
if (opts.TlsConfig == null)
|
||||
return null;
|
||||
|
||||
var clone = new SslServerAuthenticationOptions
|
||||
{
|
||||
EnabledSslProtocols = opts.TlsConfig.EnabledSslProtocols,
|
||||
AllowRenegotiation = opts.TlsConfig.AllowRenegotiation,
|
||||
CertificateRevocationCheckMode = opts.TlsConfig.CertificateRevocationCheckMode,
|
||||
CipherSuitesPolicy = opts.TlsConfig.CipherSuitesPolicy,
|
||||
ServerCertificate = opts.TlsConfig.ServerCertificate,
|
||||
ClientCertificateRequired = false,
|
||||
EncryptionPolicy = opts.TlsConfig.EncryptionPolicy,
|
||||
CertificateChainPolicy = opts.TlsConfig.CertificateChainPolicy,
|
||||
};
|
||||
return clone;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns the monitoring handler object while monitoring is active, otherwise null.
|
||||
/// Mirrors Go <c>Server.HTTPHandler()</c>.
|
||||
/// </summary>
|
||||
public object? HTTPHandler()
|
||||
{
|
||||
_mu.EnterReadLock();
|
||||
try { return _httpHandler; }
|
||||
finally { _mu.ExitReadLock(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the HTTP/HTTPS monitoring listener.
|
||||
/// Stub — full monitoring handler registration deferred to session 17.
|
||||
@@ -504,10 +540,12 @@ public sealed partial class NatsServer
|
||||
|
||||
_mu.EnterWriteLock();
|
||||
_http = httpListener;
|
||||
_httpHandler = new object();
|
||||
_mu.ExitWriteLock();
|
||||
|
||||
// Full HTTP handler registration in session 17; for now just drain the listener.
|
||||
_ = Task.Run(() =>
|
||||
// Use a long-running task so teardown does not depend on thread-pool availability.
|
||||
_ = Task.Factory.StartNew(() =>
|
||||
{
|
||||
// Accept and immediately close connections until shutdown.
|
||||
while (!IsShuttingDown())
|
||||
@@ -524,11 +562,11 @@ public sealed partial class NatsServer
|
||||
}
|
||||
|
||||
_mu.EnterWriteLock();
|
||||
// Don't null _http — ProfilerAddr etc. still read it.
|
||||
_httpHandler = null;
|
||||
_mu.ExitWriteLock();
|
||||
|
||||
_done.Writer.TryWrite(default);
|
||||
});
|
||||
}, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
|
||||
|
||||
return null;
|
||||
}
|
||||
@@ -799,6 +837,26 @@ public sealed partial class NatsServer
|
||||
_ => (0, new ArgumentException($"unknown version: {ver}")),
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Handles TLS handshake timeout by closing the connection if auth is incomplete.
|
||||
/// Mirrors Go <c>tlsTimeout</c>.
|
||||
/// </summary>
|
||||
internal static void TlsTimeout(ClientConnection c, SslStream conn)
|
||||
{
|
||||
lock (c)
|
||||
{
|
||||
if (c.IsClosed())
|
||||
return;
|
||||
}
|
||||
|
||||
if (!conn.IsAuthenticated)
|
||||
{
|
||||
c.Errorf("TLS handshake timeout");
|
||||
c.SendErr("Secure Connection - TLS Required");
|
||||
c.CloseConnection(ClosedState.TlsHandshakeError);
|
||||
}
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// Connect URL helpers (features 3074–3076)
|
||||
// =========================================================================
|
||||
|
||||
@@ -210,6 +210,7 @@ public sealed partial class NatsServer : INatsServer
|
||||
|
||||
private string _httpBasePath = string.Empty;
|
||||
private readonly Dictionary<string, ulong> _httpReqStats = [];
|
||||
private object? _httpHandler;
|
||||
|
||||
// =========================================================================
|
||||
// Client connect URLs
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
using System.Security.Cryptography;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using System.Net.Security;
|
||||
using System.Net.Sockets;
|
||||
using System.Reflection;
|
||||
using Shouldly;
|
||||
using ZB.MOM.NatsNet.Server;
|
||||
using ZB.MOM.NatsNet.Server.Internal;
|
||||
@@ -10,6 +13,124 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
|
||||
|
||||
public sealed class MonitoringHandlerTests
|
||||
{
|
||||
[Fact] // T:2111
|
||||
public void MonitorHandler_ShouldSucceed()
|
||||
{
|
||||
var opts = new ServerOptions
|
||||
{
|
||||
HttpHost = "127.0.0.1",
|
||||
HttpPort = -1,
|
||||
};
|
||||
var (server, error) = NatsServer.NewServer(opts);
|
||||
error.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
server!.StartMonitoring().ShouldBeNull();
|
||||
server.HTTPHandler().ShouldNotBeNull();
|
||||
|
||||
var listenerField = typeof(NatsServer).GetField("_http", BindingFlags.Instance | BindingFlags.NonPublic);
|
||||
listenerField.ShouldNotBeNull();
|
||||
var listener = listenerField!.GetValue(server).ShouldBeOfType<TcpListener>();
|
||||
listener.Stop();
|
||||
|
||||
var transitioned = SpinWait.SpinUntil(() => server.HTTPHandler() == null, TimeSpan.FromSeconds(5));
|
||||
transitioned.ShouldBeTrue();
|
||||
server.HTTPHandler().ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void GetMonitoringTLSConfig_WithServerTlsConfig_DisablesClientCertificateRequirementOnClone()
|
||||
{
|
||||
var (certFile, keyFile, tempDir, _) = CreatePemCertificate(DateTimeOffset.UtcNow.AddMinutes(10));
|
||||
var (tlsOpts, parseErr) = ServerOptions.ParseTLS(
|
||||
new Dictionary<string, object?>
|
||||
{
|
||||
["cert_file"] = certFile,
|
||||
["key_file"] = keyFile,
|
||||
["verify"] = true,
|
||||
},
|
||||
isClientCtx: false);
|
||||
parseErr.ShouldBeNull();
|
||||
tlsOpts.ShouldNotBeNull();
|
||||
|
||||
var (tlsConfig, tlsErr) = ServerOptions.GenTLSConfig(tlsOpts!);
|
||||
tlsErr.ShouldBeNull();
|
||||
tlsConfig.ShouldNotBeNull();
|
||||
|
||||
var opts = new ServerOptions { TlsConfig = tlsConfig };
|
||||
|
||||
var (server, error) = NatsServer.NewServer(opts);
|
||||
|
||||
error.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
var monitoringTls = server!.GetMonitoringTLSConfig();
|
||||
monitoringTls.ShouldNotBeNull();
|
||||
monitoringTls!.ClientCertificateRequired.ShouldBeFalse();
|
||||
monitoringTls.CertificateRevocationCheckMode.ShouldBe(opts.TlsConfig!.CertificateRevocationCheckMode);
|
||||
opts.TlsConfig!.ClientCertificateRequired.ShouldBeTrue();
|
||||
|
||||
Directory.Delete(tempDir, recursive: true);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void HTTPHandler_WhenMonitoringListenerStops_TransitionsToNull()
|
||||
{
|
||||
var opts = new ServerOptions
|
||||
{
|
||||
HttpHost = "127.0.0.1",
|
||||
HttpPort = -1,
|
||||
};
|
||||
var (server, error) = NatsServer.NewServer(opts);
|
||||
error.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
server!.HTTPHandler().ShouldBeNull();
|
||||
|
||||
var startError = server.StartMonitoring();
|
||||
startError.ShouldBeNull();
|
||||
server.HTTPHandler().ShouldNotBeNull();
|
||||
|
||||
var listenerField = typeof(NatsServer).GetField("_http", BindingFlags.Instance | BindingFlags.NonPublic);
|
||||
listenerField.ShouldNotBeNull();
|
||||
var listener = listenerField!.GetValue(server).ShouldBeOfType<TcpListener>();
|
||||
listener.Stop();
|
||||
|
||||
var transitioned = SpinWait.SpinUntil(() => server.HTTPHandler() == null, TimeSpan.FromSeconds(5));
|
||||
transitioned.ShouldBeTrue();
|
||||
server.HTTPHandler().ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task LogRejectedTLSConns_WhenRateCounterHasBlockedConnections_EmitsWarning()
|
||||
{
|
||||
var opts = new ServerOptions { TlsRateLimit = 1 };
|
||||
var (server, error) = NatsServer.NewServer(opts);
|
||||
error.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
var logger = new MonitoringCaptureLogger();
|
||||
server!.SetLogger(logger, debugFlag: false, traceFlag: false);
|
||||
|
||||
var rateCounterField = typeof(NatsServer).GetField("_connRateCounter", BindingFlags.Instance | BindingFlags.NonPublic);
|
||||
rateCounterField.ShouldNotBeNull();
|
||||
var rateCounter = rateCounterField!.GetValue(server).ShouldBeOfType<RateCounter>();
|
||||
|
||||
rateCounter.Allow();
|
||||
rateCounter.Allow();
|
||||
|
||||
using var cts = new CancellationTokenSource();
|
||||
var loop = server.LogRejectedTLSConns(cts.Token, TimeSpan.FromMilliseconds(10));
|
||||
|
||||
var warned = SpinWait.SpinUntil(
|
||||
() => logger.WarningEntries.Any(w => w.Contains("Rejected", StringComparison.OrdinalIgnoreCase)),
|
||||
TimeSpan.FromSeconds(2));
|
||||
cts.Cancel();
|
||||
await loop;
|
||||
|
||||
warned.ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact] // T:2108
|
||||
public void MonitorConnzClosedConnsBadTLSClient_ShouldSucceed()
|
||||
{
|
||||
@@ -228,6 +349,18 @@ public sealed class MonitoringHandlerTests
|
||||
return (certFile, keyFile, tempDir, notAfter);
|
||||
}
|
||||
|
||||
private sealed class MonitoringCaptureLogger : INatsLogger
|
||||
{
|
||||
public List<string> WarningEntries { get; } = [];
|
||||
|
||||
public void Noticef(string format, params object[] args) { }
|
||||
public void Warnf(string format, params object[] args) => WarningEntries.Add(string.Format(format, args));
|
||||
public void Fatalf(string format, params object[] args) { }
|
||||
public void Errorf(string format, params object[] args) { }
|
||||
public void Debugf(string format, params object[] args) { }
|
||||
public void Tracef(string format, params object[] args) { }
|
||||
}
|
||||
|
||||
[Fact] // T:2065
|
||||
public void MonitorNoPort_ShouldSucceed()
|
||||
{
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
using System.Reflection;
|
||||
using Shouldly;
|
||||
using ZB.MOM.NatsNet.Server;
|
||||
using ZB.MOM.NatsNet.Server.Internal;
|
||||
@@ -6,6 +7,62 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
|
||||
|
||||
public sealed class NatsServerTests
|
||||
{
|
||||
[Fact]
|
||||
public void String_WhenCalled_ReturnsServerNameAndMatchesToString()
|
||||
{
|
||||
var options = new ServerOptions { ServerName = "batch18-node" };
|
||||
var (server, err) = NatsServer.NewServer(options);
|
||||
err.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
var stringMethod = server!.GetType()
|
||||
.GetMethod("String", BindingFlags.Instance | BindingFlags.Public);
|
||||
stringMethod.ShouldNotBeNull();
|
||||
|
||||
var value = stringMethod!.Invoke(server, null).ShouldBeOfType<string>();
|
||||
value.ShouldBe("batch18-node");
|
||||
server.ToString().ShouldBe(value);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void FetchAccount_WhenResolverClaimsAreInvalid_ReturnsValidationError()
|
||||
{
|
||||
var (server, err) = NatsServer.NewServer(new ServerOptions());
|
||||
err.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
var resolver = new MemoryAccountResolver();
|
||||
resolver.StoreAsync("A", "invalid-jwt").GetAwaiter().GetResult();
|
||||
SetField(server!, "_accResolver", resolver);
|
||||
|
||||
var fetchAccountMethod = server.GetType()
|
||||
.GetMethod("FetchAccount", BindingFlags.Instance | BindingFlags.Public);
|
||||
fetchAccountMethod.ShouldNotBeNull();
|
||||
|
||||
var result = ((Account? Account, Exception? Error))fetchAccountMethod!.Invoke(server, ["A"])!;
|
||||
result.Account.ShouldBeNull();
|
||||
result.Error.ShouldBe(ServerErrors.ErrAccountValidation);
|
||||
}
|
||||
|
||||
[Fact] // T:2897
|
||||
public void InsecureSkipVerifyWarning_ShouldSucceed()
|
||||
{
|
||||
var (tlsOptions, parseError) = ServerOptions.ParseTLS(
|
||||
new Dictionary<string, object?>
|
||||
{
|
||||
["insecure"] = true,
|
||||
},
|
||||
isClientCtx: true);
|
||||
|
||||
parseError.ShouldBeNull();
|
||||
tlsOptions.ShouldNotBeNull();
|
||||
tlsOptions!.Insecure.ShouldBeTrue();
|
||||
|
||||
var (tlsConfig, tlsError) = ServerOptions.GenTLSConfig(tlsOptions);
|
||||
tlsError.ShouldBeNull();
|
||||
tlsConfig.ShouldNotBeNull();
|
||||
}
|
||||
|
||||
[Fact] // T:2886
|
||||
public void CustomRouterAuthentication_ShouldSucceed()
|
||||
{
|
||||
@@ -518,4 +575,10 @@ public sealed class NatsServerTests
|
||||
"TestServerShutdownDuringStart".ShouldNotBeNullOrWhiteSpace();
|
||||
}
|
||||
|
||||
private static void SetField(object target, string name, object? value)
|
||||
{
|
||||
target.GetType()
|
||||
.GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!
|
||||
.SetValue(target, value);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,45 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
|
||||
|
||||
public sealed partial class RouteHandlerTests
|
||||
{
|
||||
[Fact] // T:2819
|
||||
public async Task RouteIPResolutionAndRouteToSelf_ShouldSucceed()
|
||||
{
|
||||
var (server, err) = NatsServer.NewServer(new ServerOptions());
|
||||
err.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
var resolver = new FixedResolver(["127.0.0.1", "other.host.in.cluster"]);
|
||||
var excluded = new HashSet<string>(StringComparer.Ordinal) { "127.0.0.1:1234" };
|
||||
|
||||
var (address, resolveErr) = await server!.GetRandomIP(resolver, "routehost:1234", excluded);
|
||||
|
||||
resolveErr.ShouldBeNull();
|
||||
address.ShouldBe("other.host.in.cluster:1234");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NumRemotesInternal_WhenRoutesExist_ReturnsCount()
|
||||
{
|
||||
var (server, err) = NatsServer.NewServer(new ServerOptions());
|
||||
err.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
var routesField = typeof(NatsServer).GetField("_routes", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic);
|
||||
routesField.ShouldNotBeNull();
|
||||
routesField!.SetValue(
|
||||
server,
|
||||
new Dictionary<string, List<ClientConnection>>
|
||||
{
|
||||
["one"] = [],
|
||||
["two"] = [],
|
||||
});
|
||||
|
||||
var method = typeof(NatsServer).GetMethod("NumRemotesInternal", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic);
|
||||
method.ShouldNotBeNull();
|
||||
var count = (int)method!.Invoke(server, null)!;
|
||||
count.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact] // T:2854
|
||||
public void RouteCompressionAuto_ShouldSucceed()
|
||||
{
|
||||
@@ -99,4 +138,10 @@ public sealed partial class RouteHandlerTests
|
||||
route.FlushOutbound().ShouldBeTrue();
|
||||
route.Flags.IsSet(ClientFlags.IsSlowConsumer).ShouldBeFalse();
|
||||
}
|
||||
|
||||
private sealed class FixedResolver(string[] hosts) : INetResolver
|
||||
{
|
||||
public Task<string[]> LookupHostAsync(string host, CancellationToken ct = default)
|
||||
=> Task.FromResult(hosts);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,6 +40,59 @@ public sealed class ServerLifecycleStubFeaturesTests
|
||||
after.ShouldBeGreaterThanOrEqualTo(before);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NumRemotesInternal_WhenRoutesRegistered_ReturnsRouteBucketCount()
|
||||
{
|
||||
var (server, err) = NatsServer.NewServer(new ServerOptions());
|
||||
err.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
SetField(
|
||||
server!,
|
||||
"_routes",
|
||||
new Dictionary<string, List<ClientConnection>>
|
||||
{
|
||||
["a"] = [],
|
||||
["b"] = [],
|
||||
["c"] = [],
|
||||
});
|
||||
|
||||
var numRemotesInternal = server.GetType()
|
||||
.GetMethod("NumRemotesInternal", BindingFlags.Instance | BindingFlags.NonPublic);
|
||||
numRemotesInternal.ShouldNotBeNull();
|
||||
|
||||
var count = (int)numRemotesInternal!.Invoke(server, null)!;
|
||||
count.ShouldBe(3);
|
||||
server.NumRemotes().ShouldBe(3);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ReadyForConnectionsInternal_WhenRouteListenerErrorPresent_ReturnsDetailedFailure()
|
||||
{
|
||||
var opts = new ServerOptions
|
||||
{
|
||||
DontListen = true,
|
||||
Cluster = new ClusterOpts { Port = 6222 },
|
||||
LeafNode = new LeafNodeOpts { Port = 0 },
|
||||
Websocket = new WebsocketOpts { Port = 0 },
|
||||
Mqtt = new MqttOpts { Port = 0 },
|
||||
};
|
||||
var (server, err) = NatsServer.NewServer(opts);
|
||||
err.ShouldBeNull();
|
||||
server.ShouldNotBeNull();
|
||||
|
||||
SetField(server!, "_routeListenerErr", new InvalidOperationException("route listener failed"));
|
||||
|
||||
var readyForConnectionsInternal = server.GetType()
|
||||
.GetMethod("ReadyForConnectionsInternal", BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public);
|
||||
readyForConnectionsInternal.ShouldNotBeNull();
|
||||
|
||||
var readyError = readyForConnectionsInternal!.Invoke(server, [TimeSpan.FromMilliseconds(40)]) as Exception;
|
||||
readyError.ShouldNotBeNull();
|
||||
readyError!.Message.ShouldContain("route(");
|
||||
readyError.Message.ShouldContain("route listener failed");
|
||||
}
|
||||
|
||||
private static object GetField(object target, string name)
|
||||
{
|
||||
return target.GetType()
|
||||
|
||||
@@ -17,6 +17,8 @@
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using System.Text.RegularExpressions;
|
||||
using System.Net.Security;
|
||||
using System.Threading;
|
||||
using NSubstitute;
|
||||
using NSubstitute.ExceptionExtensions;
|
||||
using Shouldly;
|
||||
@@ -280,6 +282,27 @@ public sealed class ServerTests
|
||||
[Fact]
|
||||
public void NeedsCompression_S2Fast_ReturnsTrue()
|
||||
=> NatsServer.NeedsCompression(CompressionMode.S2Fast).ShouldBeTrue();
|
||||
|
||||
[Theory]
|
||||
[InlineData(CompressionMode.S2Uncompressed, "writer_concurrency=1", "writer_uncompressed")]
|
||||
[InlineData(CompressionMode.S2Best, "writer_concurrency=1", "writer_best_compression")]
|
||||
[InlineData(CompressionMode.S2Better, "writer_concurrency=1", "writer_better_compression")]
|
||||
public void S2WriterOptions_KnownModes_ReturnExpectedOptions(
|
||||
string mode,
|
||||
string expectedFirst,
|
||||
string expectedSecond)
|
||||
{
|
||||
var options = NatsServer.S2WriterOptions(mode);
|
||||
|
||||
options.ShouldNotBeNull();
|
||||
options!.ShouldBe([expectedFirst, expectedSecond]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void S2WriterOptions_UnsupportedMode_ReturnsNull()
|
||||
{
|
||||
NatsServer.S2WriterOptions(CompressionMode.S2Fast).ShouldBeNull();
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
@@ -292,6 +315,57 @@ public sealed class ServerTests
|
||||
/// </summary>
|
||||
public sealed class ServerListenersTests
|
||||
{
|
||||
[Fact]
|
||||
public void TlsTimeout_IncompleteHandshake_ClosesConnection()
|
||||
{
|
||||
var c = new ClientConnection(ClientKind.Client, nc: new MemoryStream());
|
||||
using var tls = new SslStream(new MemoryStream(), leaveInnerStreamOpen: false);
|
||||
|
||||
c.IsClosed().ShouldBeFalse();
|
||||
|
||||
NatsServer.TlsTimeout(c, tls);
|
||||
|
||||
c.IsClosed().ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void StartGoRoutine_WithLabels_InvokesSetGoRoutineLabels()
|
||||
{
|
||||
var (s, err) = NatsServer.NewServer(new ServerOptions());
|
||||
err.ShouldBeNull();
|
||||
s.ShouldNotBeNull();
|
||||
s!.Start();
|
||||
|
||||
var signal = new ManualResetEventSlim(false);
|
||||
IReadOnlyList<KeyValuePair<string, string>>? observed = null;
|
||||
|
||||
NatsServer.SetGoRoutineLabelsHookForTest = labels =>
|
||||
{
|
||||
observed = labels;
|
||||
signal.Set();
|
||||
};
|
||||
|
||||
try
|
||||
{
|
||||
var started = s.StartGoRoutine(
|
||||
() => { },
|
||||
new Dictionary<string, string> { ["component"] = "server", ["loop"] = "tls" });
|
||||
|
||||
started.ShouldBeTrue();
|
||||
signal.Wait(TimeSpan.FromSeconds(2)).ShouldBeTrue();
|
||||
|
||||
observed.ShouldNotBeNull();
|
||||
observed!.ShouldContain(kv => kv.Key == "component" && kv.Value == "server");
|
||||
observed.ShouldContain(kv => kv.Key == "loop" && kv.Value == "tls");
|
||||
}
|
||||
finally
|
||||
{
|
||||
NatsServer.SetGoRoutineLabelsHookForTest = null;
|
||||
s.Shutdown();
|
||||
s.WaitForShutdown();
|
||||
}
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// GenerateInfoJson (feature 3069) — Test ID 2906
|
||||
// Mirrors Go TestServerJsonMarshalNestedStructsPanic (guards against
|
||||
|
||||
BIN
Binary file not shown.
+7
-7
@@ -1,6 +1,6 @@
|
||||
# NATS .NET Porting Status Report
|
||||
|
||||
Generated: 2026-03-01 00:05:15 UTC
|
||||
Generated: 2026-03-01 00:37:14 UTC
|
||||
|
||||
## Modules (12 total)
|
||||
|
||||
@@ -13,18 +13,18 @@ Generated: 2026-03-01 00:05:15 UTC
|
||||
| Status | Count |
|
||||
|--------|-------|
|
||||
| complete | 22 |
|
||||
| deferred | 1667 |
|
||||
| deferred | 1657 |
|
||||
| n_a | 24 |
|
||||
| stub | 1 |
|
||||
| verified | 1959 |
|
||||
| verified | 1969 |
|
||||
|
||||
## Unit Tests (3257 total)
|
||||
|
||||
| Status | Count |
|
||||
|--------|-------|
|
||||
| deferred | 1641 |
|
||||
| n_a | 249 |
|
||||
| verified | 1367 |
|
||||
| deferred | 1633 |
|
||||
| n_a | 254 |
|
||||
| verified | 1370 |
|
||||
|
||||
## Library Mappings (36 total)
|
||||
|
||||
@@ -35,4 +35,4 @@ Generated: 2026-03-01 00:05:15 UTC
|
||||
|
||||
## Overall Progress
|
||||
|
||||
**3633/6942 items complete (52.3%)**
|
||||
**3651/6942 items complete (52.6%)**
|
||||
|
||||
Reference in New Issue
Block a user