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

feat(batch18): implement group-a server core helpers

Browse Source
This commit is contained in:
Joseph Doherty
2026-02-28 19:17:53 -05:00
parent e16d7ffab3
commit f8d384711d
7 changed files with 308 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
dotnet/src/ZB.MOM.NatsNet.Server/NatsServer.Init.cs
View File

@@ -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; }
}
}

38
dotnet/src/ZB.MOM.NatsNet.Server/NatsServer.Lifecycle.cs
View File

@@ -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)
// =========================================================================

64
dotnet/src/ZB.MOM.NatsNet.Server/NatsServer.Listeners.cs
View File

@@ -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)
// =========================================================================

1
dotnet/src/ZB.MOM.NatsNet.Server/NatsServer.cs
View File

@@ -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