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

feat(batch17): port lifecycle, tls, and rate-limited logging features

Browse Source
This commit is contained in:
Joseph Doherty
2026-02-28 19:32:59 -05:00
parent 8d5964efff
commit 4974339b52
8 changed files with 694 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ClientConnectionStubFeaturesTests.cs
View File

@@ -283,4 +283,117 @@ public sealed class ClientConnectionStubFeaturesTests
result.PSubs.Add(new Subscription { Subject = Encoding.ASCII.GetBytes("foo"), Sid = Encoding.ASCII.GetBytes("1") });
c.ProcessMsgResults(null, result, "hello\r\n"u8.ToArray(), null, Encoding.ASCII.GetBytes("foo"), null, PmrFlags.None).didDeliver.ShouldBeTrue();
}
[Fact]
public void LifecycleAndTlsHelpers_GroupC_ShouldBehave()
{
var logger = new CaptureLogger();
var (server, err) = NatsServer.NewServer(new ServerOptions
{
PingInterval = TimeSpan.FromMilliseconds(120),
});
err.ShouldBeNull();
server.SetLogger(logger, debugFlag: true, traceFlag: true);
using var ms = new MemoryStream();
var c = new ClientConnection(ClientKind.Client, server, ms)
{
Cid = 42,
Host = "127.0.0.1",
Start = DateTime.UtcNow.AddSeconds(-2),
Rtt = TimeSpan.FromMilliseconds(5),
};
c.SetFirstPingTimer();
GetTimer(c, "_pingTimer").ShouldNotBeNull();
c.WatchForStaleConnection(TimeSpan.FromMilliseconds(20), pingMax: 0);
Thread.Sleep(60);
c.IsClosed().ShouldBeTrue();
var temp = Account.NewAccount("A");
temp.Sublist = SubscriptionIndex.NewSublistWithCache();
c.SetAccount(temp);
var registered = server.LookupOrRegisterAccount("A").Account;
registered.Sublist = SubscriptionIndex.NewSublistWithCache();
var inserted = new Subscription
{
Subject = Encoding.ASCII.GetBytes("foo.bar"),
Sid = Encoding.ASCII.GetBytes("11"),
};
registered.Sublist.Insert(inserted).ShouldBeNull();
c.SwapAccountAfterReload();
c.GetAccount().ShouldBe(registered);
c.Perms = new ClientPermissions();
c.Perms.Sub.Deny = SubscriptionIndex.NewSublistWithCache();
c.Perms.Sub.Deny.Insert(new Subscription { Subject = Encoding.ASCII.GetBytes(">") }).ShouldBeNull();
c.Subs["22"] = new Subscription
{
Subject = Encoding.ASCII.GetBytes("foo.bar"),
Sid = Encoding.ASCII.GetBytes("22"),
};
c.ProcessSubsOnConfigReload(new HashSet<string>(StringComparer.Ordinal) { registered.Name });
c.Subs.ContainsKey("22").ShouldBeFalse();
c.ParseCtx.Pa.Account = Encoding.ASCII.GetBytes("A");
c.ParseCtx.Pa.Subject = Encoding.ASCII.GetBytes("foo.bar");
c.ParseCtx.Pa.PaCache = Encoding.ASCII.GetBytes("A:foo.bar");
var cached = c.GetAccAndResultFromCache();
cached.Account.ShouldBe(registered);
cached.Result.ShouldNotBeNull();
cached.Result.PSubs.Count.ShouldBeGreaterThan(0);
var closedSub = new Subscription { Subject = Encoding.ASCII.GetBytes("foo.closed") };
closedSub.Close();
var inField = typeof(ClientConnection).GetField("_in", BindingFlags.Instance | BindingFlags.NonPublic)!;
var state = (ReadCacheState)inField.GetValue(c)!;
state.PaCache = new Dictionary<string, PerAccountCache>(StringComparer.Ordinal)
{
["closed"] = new PerAccountCache
{
Acc = registered,
Results = new SubscriptionIndexResult
{
PSubs = { closedSub },
},
GenId = 1,
},
};
inField.SetValue(c, state);
c.PruneClosedSubFromPerAccountCache();
state = (ReadCacheState)inField.GetValue(c)!;
state.PaCache.ShouldNotBeNull();
state.PaCache.Count.ShouldBe(0);
var info = c.GetClientInfo(detailed: true);
info.ShouldNotBeNull();
info!.Account.ShouldBe("A");
info.Server.ShouldNotBeNullOrWhiteSpace();
info.ServiceAccount().ShouldBe("A");
var (allowed, convertErr) = ClientConnection.ConvertAllowedConnectionTypes(
["standard", "mqtt", "bad"]);
allowed.ShouldContain(AuthHandler.ConnectionTypes.Standard);
allowed.ShouldContain(AuthHandler.ConnectionTypes.Mqtt);
convertErr.ShouldNotBeNull();
c.RateLimitWarnf("warn {0}", 1);
c.RateLimitWarnf("warn {0}", 1);
logger.Warnings.Count.ShouldBe(1);
}
private sealed class CaptureLogger : INatsLogger
{
public List<string> Warnings { get; } = [];
public void Noticef(string format, params object[] args) { }
public void Warnf(string format, params object[] args) => Warnings.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) { }
}
}

33
dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/NatsServerTests.cs
View File

@@ -6,6 +6,26 @@ namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
public sealed class NatsServerTests
{
[Fact]
public void RateLimitedClientLogging_ShouldSuppressDuplicates()
{
var logger = new NatsServerCaptureLogger();
var (server, err) = NatsServer.NewServer(new ServerOptions());
err.ShouldBeNull();
server.SetLogger(logger, debugFlag: true, traceFlag: true);
var c = new ClientConnection(ClientKind.Client, server, new MemoryStream());
c.RateLimitWarnf("duplicate warning {0}", "A");
c.RateLimitWarnf("duplicate warning {0}", "A");
c.RateLimitFormatWarnf("format warning {0}", "B");
c.RateLimitFormatWarnf("format warning {0}", "C");
c.RateLimitErrorf("duplicate error {0}", "X");
c.RateLimitErrorf("duplicate error {0}", "X");
logger.Warnings.Count.ShouldBe(2);
logger.Errors.Count.ShouldBe(1);
}
[Fact] // T:2886
public void CustomRouterAuthentication_ShouldSucceed()
{
@@ -518,4 +538,17 @@ public sealed class NatsServerTests
"TestServerShutdownDuringStart".ShouldNotBeNullOrWhiteSpace();
}
private sealed class NatsServerCaptureLogger : INatsLogger
{
public List<string> Warnings { get; } = [];
public List<string> Errors { get; } = [];
public void Noticef(string format, params object[] args) { }
public void Warnf(string format, params object[] args) => Warnings.Add(string.Format(format, args));
public void Fatalf(string format, params object[] args) { }
public void Errorf(string format, params object[] args) => Errors.Add(string.Format(format, args));
public void Debugf(string format, params object[] args) { }
public void Tracef(string format, params object[] args) { }
}
}