7fbffffd05
- Rename tests/NATS.Server.Tests -> tests/NATS.Server.Core.Tests - Update solution file, InternalsVisibleTo, and csproj references - Remove JETSTREAM_INTEGRATION_MATRIX and NATS.NKeys from csproj (moved to JetStream.Tests and Auth.Tests) - Update all namespaces from NATS.Server.Tests.* to NATS.Server.Core.Tests.* - Replace private GetFreePort/ReadUntilAsync helpers with TestUtilities calls - Fix stale namespace in Transport.Tests/NetworkingGoParityTests.cs
1216 lines
46 KiB
C#
1216 lines
46 KiB
C#
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Text;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using NATS.Client.Core;
|
|
using NATS.Server.Auth;
|
|
using NATS.Server.Configuration;
|
|
using NATS.Server.Gateways;
|
|
using NATS.Server.LeafNodes;
|
|
using NATS.Server.Routes;
|
|
using NATS.Server.Subscriptions;
|
|
using NATS.Server.TestUtilities;
|
|
|
|
namespace NATS.Server.Transport.Tests.Networking;
|
|
|
|
/// <summary>
|
|
/// Ported Go networking tests for gateway interest mode, route pool accounting,
|
|
/// and leaf node connections. Each test references the Go function name and file.
|
|
/// </summary>
|
|
public class NetworkingGoParityTests
|
|
{
|
|
// ════════════════════════════════════════════════════════════════════
|
|
// GATEWAY INTEREST MODE (~20 tests from gateway_test.go)
|
|
// ════════════════════════════════════════════════════════════════════
|
|
|
|
// Go: TestGatewayDontSendSubInterest server/gateway_test.go:1755
|
|
[Fact]
|
|
public void Tracker_starts_in_optimistic_mode()
|
|
{
|
|
var tracker = new GatewayInterestTracker();
|
|
tracker.GetMode("$G").ShouldBe(GatewayInterestMode.Optimistic);
|
|
}
|
|
|
|
// Go: TestGatewayAccountInterest server/gateway_test.go:1794
|
|
[Fact]
|
|
public void Tracker_no_interest_accumulates_in_optimistic_mode()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 5);
|
|
for (var i = 0; i < 4; i++)
|
|
tracker.TrackNoInterest("$G", $"subj.{i}");
|
|
|
|
tracker.GetMode("$G").ShouldBe(GatewayInterestMode.Optimistic);
|
|
tracker.ShouldForward("$G", "subj.0").ShouldBeFalse();
|
|
tracker.ShouldForward("$G", "other").ShouldBeTrue();
|
|
}
|
|
|
|
// Go: TestGatewaySwitchToInterestOnlyModeImmediately server/gateway_test.go:6934
|
|
[Fact]
|
|
public void Tracker_switches_to_interest_only_at_threshold()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 3);
|
|
tracker.TrackNoInterest("$G", "a");
|
|
tracker.TrackNoInterest("$G", "b");
|
|
tracker.TrackNoInterest("$G", "c");
|
|
|
|
tracker.GetMode("$G").ShouldBe(GatewayInterestMode.InterestOnly);
|
|
}
|
|
|
|
// Go: TestGatewayAccountInterest server/gateway_test.go:1794
|
|
[Fact]
|
|
public void Tracker_interest_only_blocks_unknown_subjects()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 1);
|
|
tracker.TrackNoInterest("$G", "trigger");
|
|
|
|
tracker.GetMode("$G").ShouldBe(GatewayInterestMode.InterestOnly);
|
|
tracker.ShouldForward("$G", "unknown.subject").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayAccountInterest server/gateway_test.go:1794
|
|
[Fact]
|
|
public void Tracker_interest_only_forwards_tracked_subjects()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 1);
|
|
tracker.TrackNoInterest("$G", "trigger");
|
|
tracker.TrackInterest("$G", "orders.>");
|
|
|
|
tracker.ShouldForward("$G", "orders.created").ShouldBeTrue();
|
|
tracker.ShouldForward("$G", "users.created").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayAccountUnsub server/gateway_test.go:1912
|
|
[Fact]
|
|
public void Tracker_removing_interest_in_io_mode_stops_forwarding()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 1);
|
|
tracker.TrackNoInterest("$G", "trigger");
|
|
tracker.TrackInterest("$G", "foo");
|
|
tracker.ShouldForward("$G", "foo").ShouldBeTrue();
|
|
|
|
tracker.TrackNoInterest("$G", "foo");
|
|
tracker.ShouldForward("$G", "foo").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayAccountInterest server/gateway_test.go:1794
|
|
[Fact]
|
|
public void Tracker_accounts_are_independent()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 1);
|
|
tracker.TrackNoInterest("ACCT_A", "trigger");
|
|
|
|
tracker.GetMode("ACCT_A").ShouldBe(GatewayInterestMode.InterestOnly);
|
|
tracker.GetMode("ACCT_B").ShouldBe(GatewayInterestMode.Optimistic);
|
|
tracker.ShouldForward("ACCT_B", "any.subject").ShouldBeTrue();
|
|
}
|
|
|
|
// Go: TestGatewaySwitchToInterestOnlyModeImmediately server/gateway_test.go:6934
|
|
[Fact]
|
|
public void Tracker_explicit_switch_to_interest_only()
|
|
{
|
|
var tracker = new GatewayInterestTracker();
|
|
tracker.SwitchToInterestOnly("$G");
|
|
tracker.GetMode("$G").ShouldBe(GatewayInterestMode.InterestOnly);
|
|
tracker.ShouldForward("$G", "anything").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayAccountInterest server/gateway_test.go:1794
|
|
[Fact]
|
|
public void Tracker_optimistic_mode_interest_add_removes_from_no_interest()
|
|
{
|
|
var tracker = new GatewayInterestTracker();
|
|
tracker.TrackNoInterest("$G", "foo");
|
|
tracker.ShouldForward("$G", "foo").ShouldBeFalse();
|
|
|
|
tracker.TrackInterest("$G", "foo");
|
|
tracker.ShouldForward("$G", "foo").ShouldBeTrue();
|
|
}
|
|
|
|
// Go: TestGatewaySubjectInterest server/gateway_test.go:1972
|
|
[Fact]
|
|
public void Tracker_wildcard_interest_matches_in_io_mode()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 1);
|
|
tracker.TrackNoInterest("$G", "trigger");
|
|
tracker.TrackInterest("$G", "events.>");
|
|
|
|
tracker.ShouldForward("$G", "events.created").ShouldBeTrue();
|
|
tracker.ShouldForward("$G", "events.a.b.c").ShouldBeTrue();
|
|
tracker.ShouldForward("$G", "other").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayAccountInterest server/gateway_test.go:1794
|
|
[Fact]
|
|
public void ShouldForwardInterestOnly_uses_SubList_remote_interest()
|
|
{
|
|
using var subList = new SubList();
|
|
subList.ApplyRemoteSub(new RemoteSubscription("orders.*", null, "gw1", "$G"));
|
|
|
|
GatewayManager.ShouldForwardInterestOnly(subList, "$G", "orders.created").ShouldBeTrue();
|
|
GatewayManager.ShouldForwardInterestOnly(subList, "$G", "users.created").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayAccountUnsub server/gateway_test.go:1912
|
|
[Fact]
|
|
public void ShouldForwardInterestOnly_respects_removal()
|
|
{
|
|
using var subList = new SubList();
|
|
subList.ApplyRemoteSub(new RemoteSubscription("orders.*", null, "gw1", "$G"));
|
|
GatewayManager.ShouldForwardInterestOnly(subList, "$G", "orders.created").ShouldBeTrue();
|
|
|
|
subList.ApplyRemoteSub(RemoteSubscription.Removal("orders.*", null, "gw1", "$G"));
|
|
GatewayManager.ShouldForwardInterestOnly(subList, "$G", "orders.created").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewaySubjectInterest server/gateway_test.go:1972
|
|
[Fact]
|
|
public async Task Gateway_propagates_subject_interest_end_to_end()
|
|
{
|
|
await using var fixture = await TwoGatewayFixture.StartAsync();
|
|
|
|
await using var conn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Remote.Port}",
|
|
});
|
|
await conn.ConnectAsync();
|
|
|
|
await using var sub = await conn.SubscribeCoreAsync<string>("gw.interest.test");
|
|
await conn.PingAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => fixture.Local.HasRemoteInterest("gw.interest.test"));
|
|
|
|
fixture.Local.HasRemoteInterest("gw.interest.test").ShouldBeTrue();
|
|
}
|
|
|
|
// Go: TestGatewayDontSendSubInterest server/gateway_test.go:1755
|
|
[Fact]
|
|
public async Task Gateway_message_forwarded_to_remote_subscriber()
|
|
{
|
|
await using var fixture = await TwoGatewayFixture.StartAsync();
|
|
|
|
await using var remoteConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Remote.Port}",
|
|
});
|
|
await remoteConn.ConnectAsync();
|
|
await using var localConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Local.Port}",
|
|
});
|
|
await localConn.ConnectAsync();
|
|
|
|
await using var sub = await remoteConn.SubscribeCoreAsync<string>("gw.fwd.test");
|
|
await remoteConn.PingAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => fixture.Local.HasRemoteInterest("gw.fwd.test"));
|
|
|
|
await localConn.PublishAsync("gw.fwd.test", "gateway-msg");
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
(await sub.Msgs.ReadAsync(cts.Token)).Data.ShouldBe("gateway-msg");
|
|
}
|
|
|
|
// Go: TestGatewayAccountUnsub server/gateway_test.go:1912
|
|
[Fact]
|
|
public async Task Gateway_unsubscribe_removes_remote_interest()
|
|
{
|
|
await using var fixture = await TwoGatewayFixture.StartAsync();
|
|
|
|
await using var conn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Remote.Port}",
|
|
});
|
|
await conn.ConnectAsync();
|
|
|
|
var sub = await conn.SubscribeCoreAsync<string>("gw.unsub.test");
|
|
await conn.PingAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => fixture.Local.HasRemoteInterest("gw.unsub.test"));
|
|
|
|
fixture.Local.HasRemoteInterest("gw.unsub.test").ShouldBeTrue();
|
|
|
|
await sub.DisposeAsync();
|
|
await conn.PingAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => !(fixture.Local.HasRemoteInterest("gw.unsub.test")));
|
|
|
|
fixture.Local.HasRemoteInterest("gw.unsub.test").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayNoAccInterestThenQSubThenRegularSub server/gateway_test.go:5643
|
|
[Fact]
|
|
public async Task Gateway_wildcard_interest_propagates()
|
|
{
|
|
await using var fixture = await TwoGatewayFixture.StartAsync();
|
|
|
|
await using var conn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Remote.Port}",
|
|
});
|
|
await conn.ConnectAsync();
|
|
|
|
await using var sub = await conn.SubscribeCoreAsync<string>("gw.wild.>");
|
|
await conn.PingAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => fixture.Local.HasRemoteInterest("gw.wild.test"));
|
|
|
|
fixture.Local.HasRemoteInterest("gw.wild.test").ShouldBeTrue();
|
|
fixture.Local.HasRemoteInterest("gw.wild.deep.nested").ShouldBeTrue();
|
|
}
|
|
|
|
// Go: TestGatewayNoCrashOnInvalidSubject server/gateway_test.go:6279
|
|
[Fact]
|
|
public void Invalid_subject_does_not_crash_SubList()
|
|
{
|
|
using var subList = new SubList();
|
|
// Should handle gracefully, not throw
|
|
subList.HasRemoteInterest("$G", "valid.subject").ShouldBeFalse();
|
|
subList.HasRemoteInterest("$G", "").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestGatewayLogAccountInterestModeSwitch server/gateway_test.go:5843
|
|
[Fact]
|
|
public void Tracker_default_threshold_is_1000()
|
|
{
|
|
GatewayInterestTracker.DefaultNoInterestThreshold.ShouldBe(1000);
|
|
}
|
|
|
|
// Go: TestGatewayAccountInterestModeSwitchOnlyOncePerAccount server/gateway_test.go:5932
|
|
[Fact]
|
|
public void Tracker_switch_is_idempotent()
|
|
{
|
|
var tracker = new GatewayInterestTracker(noInterestThreshold: 1);
|
|
tracker.TrackNoInterest("$G", "a");
|
|
tracker.GetMode("$G").ShouldBe(GatewayInterestMode.InterestOnly);
|
|
|
|
// Switching again should not change state
|
|
tracker.SwitchToInterestOnly("$G");
|
|
tracker.GetMode("$G").ShouldBe(GatewayInterestMode.InterestOnly);
|
|
}
|
|
|
|
// Go: TestGatewayReplyMappingBasic server/gateway_test.go:3200
|
|
[Fact]
|
|
public void Reply_mapper_round_trips()
|
|
{
|
|
var mapped = ReplyMapper.ToGatewayReply("INBOX.abc123", "SERVERID1");
|
|
mapped.ShouldNotBeNull();
|
|
mapped!.ShouldStartWith("_GR_.");
|
|
|
|
ReplyMapper.HasGatewayReplyPrefix(mapped).ShouldBeTrue();
|
|
ReplyMapper.TryRestoreGatewayReply(mapped, out var restored).ShouldBeTrue();
|
|
restored.ShouldBe("INBOX.abc123");
|
|
}
|
|
|
|
// Go: TestGatewayReplyMappingBasic server/gateway_test.go:3200
|
|
[Fact]
|
|
public void Reply_mapper_null_input_returns_null()
|
|
{
|
|
var result = ReplyMapper.ToGatewayReply(null, "S1");
|
|
result.ShouldBeNull();
|
|
}
|
|
|
|
// ════════════════════════════════════════════════════════════════════
|
|
// ROUTE POOL ACCOUNTING (~15 tests from routes_test.go)
|
|
// ════════════════════════════════════════════════════════════════════
|
|
|
|
// Go: TestRoutePool server/routes_test.go:1966
|
|
[Fact]
|
|
public void Route_pool_idx_deterministic_for_same_account()
|
|
{
|
|
var idx1 = RouteManager.ComputeRoutePoolIdx(3, "$G");
|
|
var idx2 = RouteManager.ComputeRoutePoolIdx(3, "$G");
|
|
idx1.ShouldBe(idx2);
|
|
}
|
|
|
|
// Go: TestRoutePool server/routes_test.go:1966
|
|
[Fact]
|
|
public void Route_pool_idx_in_range()
|
|
{
|
|
for (var poolSize = 1; poolSize <= 10; poolSize++)
|
|
{
|
|
var idx = RouteManager.ComputeRoutePoolIdx(poolSize, "$G");
|
|
idx.ShouldBeGreaterThanOrEqualTo(0);
|
|
idx.ShouldBeLessThan(poolSize);
|
|
}
|
|
}
|
|
|
|
// Go: TestRoutePool server/routes_test.go:1966
|
|
[Fact]
|
|
public void Route_pool_idx_distributes_accounts()
|
|
{
|
|
var accounts = new[] { "$G", "ACCT_A", "ACCT_B", "ACCT_C", "ACCT_D" };
|
|
var poolSize = 3;
|
|
var indices = new HashSet<int>();
|
|
foreach (var account in accounts)
|
|
indices.Add(RouteManager.ComputeRoutePoolIdx(poolSize, account));
|
|
|
|
// With 5 accounts and pool of 3, we should use at least 2 different indices
|
|
indices.Count.ShouldBeGreaterThanOrEqualTo(2);
|
|
}
|
|
|
|
// Go: TestRoutePool server/routes_test.go:1966
|
|
[Fact]
|
|
public void Route_pool_idx_single_pool_always_zero()
|
|
{
|
|
RouteManager.ComputeRoutePoolIdx(1, "$G").ShouldBe(0);
|
|
RouteManager.ComputeRoutePoolIdx(1, "ACCT_A").ShouldBe(0);
|
|
RouteManager.ComputeRoutePoolIdx(1, "ACCT_B").ShouldBe(0);
|
|
}
|
|
|
|
// Go: TestRoutePoolConnectRace server/routes_test.go:2100
|
|
[Fact]
|
|
public async Task Route_pool_default_three_connections_per_peer()
|
|
{
|
|
var optsA = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Cluster = new ClusterOptions
|
|
{
|
|
Name = Guid.NewGuid().ToString("N"),
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
},
|
|
};
|
|
|
|
var serverA = new NatsServer(optsA, NullLoggerFactory.Instance);
|
|
var ctsA = new CancellationTokenSource();
|
|
_ = serverA.StartAsync(ctsA.Token);
|
|
await serverA.WaitForReadyAsync();
|
|
|
|
try
|
|
{
|
|
var optsB = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Cluster = new ClusterOptions
|
|
{
|
|
Name = Guid.NewGuid().ToString("N"),
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Routes = [serverA.ClusterListen!],
|
|
},
|
|
};
|
|
|
|
var serverB = new NatsServer(optsB, NullLoggerFactory.Instance);
|
|
var ctsB = new CancellationTokenSource();
|
|
_ = serverB.StartAsync(ctsB.Token);
|
|
await serverB.WaitForReadyAsync();
|
|
|
|
try
|
|
{
|
|
await PollHelper.WaitUntilAsync(() => !(serverA.Stats.Routes < 3));
|
|
|
|
serverA.Stats.Routes.ShouldBeGreaterThanOrEqualTo(3);
|
|
}
|
|
finally
|
|
{
|
|
await ctsB.CancelAsync();
|
|
serverB.Dispose();
|
|
ctsB.Dispose();
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
await ctsA.CancelAsync();
|
|
serverA.Dispose();
|
|
ctsA.Dispose();
|
|
}
|
|
}
|
|
|
|
// Go: TestRoutePoolRouteStoredSameIndexBothSides server/routes_test.go:2180
|
|
[Fact]
|
|
public void Route_pool_idx_uses_FNV1a_hash()
|
|
{
|
|
// Go uses fnv.New32a() — FNV-1a 32-bit
|
|
// Verify we produce the same hash for known inputs
|
|
var idx = RouteManager.ComputeRoutePoolIdx(10, "$G");
|
|
idx.ShouldBeGreaterThanOrEqualTo(0);
|
|
idx.ShouldBeLessThan(10);
|
|
|
|
// Same input always produces same output
|
|
RouteManager.ComputeRoutePoolIdx(10, "$G").ShouldBe(idx);
|
|
}
|
|
|
|
// Go: TestRoutePoolPerAccountSubUnsubProtoParsing server/routes_test.go:3104
|
|
[Fact]
|
|
public async Task Route_subscription_propagation_between_peers()
|
|
{
|
|
var optsA = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Cluster = new ClusterOptions
|
|
{
|
|
Name = Guid.NewGuid().ToString("N"),
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
},
|
|
};
|
|
|
|
var serverA = new NatsServer(optsA, NullLoggerFactory.Instance);
|
|
var ctsA = new CancellationTokenSource();
|
|
_ = serverA.StartAsync(ctsA.Token);
|
|
await serverA.WaitForReadyAsync();
|
|
|
|
try
|
|
{
|
|
var optsB = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Cluster = new ClusterOptions
|
|
{
|
|
Name = Guid.NewGuid().ToString("N"),
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Routes = [serverA.ClusterListen!],
|
|
},
|
|
};
|
|
|
|
var serverB = new NatsServer(optsB, NullLoggerFactory.Instance);
|
|
var ctsB = new CancellationTokenSource();
|
|
_ = serverB.StartAsync(ctsB.Token);
|
|
await serverB.WaitForReadyAsync();
|
|
|
|
try
|
|
{
|
|
await PollHelper.WaitUntilAsync(() => !(serverA.Stats.Routes < 3));
|
|
|
|
await using var conn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{serverB.Port}",
|
|
});
|
|
await conn.ConnectAsync();
|
|
|
|
await using var sub = await conn.SubscribeCoreAsync<string>("route.sub.test");
|
|
await conn.PingAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => serverA.HasRemoteInterest("route.sub.test"));
|
|
|
|
serverA.HasRemoteInterest("route.sub.test").ShouldBeTrue();
|
|
}
|
|
finally
|
|
{
|
|
await ctsB.CancelAsync();
|
|
serverB.Dispose();
|
|
ctsB.Dispose();
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
await ctsA.CancelAsync();
|
|
serverA.Dispose();
|
|
ctsA.Dispose();
|
|
}
|
|
}
|
|
|
|
// Go: TestRoutePerAccount server/routes_test.go:2539
|
|
[Fact]
|
|
public void Route_pool_different_accounts_can_get_different_indices()
|
|
{
|
|
// With a large pool, different accounts should hash to different slots
|
|
var indices = new Dictionary<string, int>();
|
|
for (var i = 0; i < 100; i++)
|
|
{
|
|
var acct = $"account_{i}";
|
|
indices[acct] = RouteManager.ComputeRoutePoolIdx(100, acct);
|
|
}
|
|
|
|
// With 100 accounts and pool size 100, we should have decent distribution
|
|
var uniqueIndices = indices.Values.Distinct().Count();
|
|
uniqueIndices.ShouldBeGreaterThan(20);
|
|
}
|
|
|
|
// Go: TestRouteSendLocalSubsWithLowMaxPending server/routes_test.go:1098
|
|
[Fact]
|
|
public async Task Route_message_forwarded_to_subscriber_on_peer()
|
|
{
|
|
var optsA = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Cluster = new ClusterOptions
|
|
{
|
|
Name = Guid.NewGuid().ToString("N"),
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
},
|
|
};
|
|
|
|
var serverA = new NatsServer(optsA, NullLoggerFactory.Instance);
|
|
var ctsA = new CancellationTokenSource();
|
|
_ = serverA.StartAsync(ctsA.Token);
|
|
await serverA.WaitForReadyAsync();
|
|
|
|
try
|
|
{
|
|
var optsB = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Cluster = new ClusterOptions
|
|
{
|
|
Name = Guid.NewGuid().ToString("N"),
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Routes = [serverA.ClusterListen!],
|
|
},
|
|
};
|
|
|
|
var serverB = new NatsServer(optsB, NullLoggerFactory.Instance);
|
|
var ctsB = new CancellationTokenSource();
|
|
_ = serverB.StartAsync(ctsB.Token);
|
|
await serverB.WaitForReadyAsync();
|
|
|
|
try
|
|
{
|
|
await PollHelper.WaitUntilAsync(() => !(serverA.Stats.Routes < 3));
|
|
|
|
await using var subConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{serverB.Port}",
|
|
});
|
|
await subConn.ConnectAsync();
|
|
|
|
await using var pubConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{serverA.Port}",
|
|
});
|
|
await pubConn.ConnectAsync();
|
|
|
|
await using var sub = await subConn.SubscribeCoreAsync<string>("route.fwd.test");
|
|
await subConn.PingAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => serverA.HasRemoteInterest("route.fwd.test"));
|
|
|
|
await pubConn.PublishAsync("route.fwd.test", "routed-msg");
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
(await sub.Msgs.ReadAsync(cts.Token)).Data.ShouldBe("routed-msg");
|
|
}
|
|
finally
|
|
{
|
|
await ctsB.CancelAsync();
|
|
serverB.Dispose();
|
|
ctsB.Dispose();
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
await ctsA.CancelAsync();
|
|
serverA.Dispose();
|
|
ctsA.Dispose();
|
|
}
|
|
}
|
|
|
|
// Go: TestRoutePoolAndPerAccountErrors server/routes_test.go:1906
|
|
[Fact]
|
|
public void Route_pool_idx_zero_pool_returns_zero()
|
|
{
|
|
RouteManager.ComputeRoutePoolIdx(0, "$G").ShouldBe(0);
|
|
}
|
|
|
|
// Go: TestRoutePoolSizeDifferentOnEachServer server/routes_test.go:2254
|
|
[Fact]
|
|
public void Route_pool_idx_consistent_across_sizes()
|
|
{
|
|
// The hash should be deterministic regardless of pool size
|
|
var hashSmall = RouteManager.ComputeRoutePoolIdx(3, "test");
|
|
var hashLarge = RouteManager.ComputeRoutePoolIdx(100, "test");
|
|
|
|
hashSmall.ShouldBeGreaterThanOrEqualTo(0);
|
|
hashLarge.ShouldBeGreaterThanOrEqualTo(0);
|
|
}
|
|
|
|
// ════════════════════════════════════════════════════════════════════
|
|
// LEAF NODE CONNECTIONS (~20 tests from leafnode_test.go)
|
|
// ════════════════════════════════════════════════════════════════════
|
|
|
|
// Go: TestLeafNodeLoop server/leafnode_test.go:837
|
|
[Fact]
|
|
public void Leaf_loop_detector_marks_and_detects()
|
|
{
|
|
var marked = LeafLoopDetector.Mark("test.subject", "S1");
|
|
LeafLoopDetector.HasLoopMarker(marked).ShouldBeTrue();
|
|
LeafLoopDetector.IsLooped(marked, "S1").ShouldBeTrue();
|
|
LeafLoopDetector.IsLooped(marked, "S2").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestLeafNodeLoop server/leafnode_test.go:837
|
|
[Fact]
|
|
public void Leaf_loop_detector_unmarks()
|
|
{
|
|
var marked = LeafLoopDetector.Mark("orders.created", "SERVER1");
|
|
LeafLoopDetector.TryUnmark(marked, out var unmarked).ShouldBeTrue();
|
|
unmarked.ShouldBe("orders.created");
|
|
}
|
|
|
|
// Go: TestLeafNodeLoop server/leafnode_test.go:837
|
|
[Fact]
|
|
public void Leaf_loop_detector_non_marked_returns_false()
|
|
{
|
|
LeafLoopDetector.HasLoopMarker("plain.subject").ShouldBeFalse();
|
|
LeafLoopDetector.IsLooped("plain.subject", "S1").ShouldBeFalse();
|
|
LeafLoopDetector.TryUnmark("plain.subject", out _).ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestLeafNodeBasicAuthSingleton server/leafnode_test.go:602
|
|
[Fact]
|
|
public async Task Leaf_connection_handshake_succeeds()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("LOCAL1", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldBe("LEAF LOCAL1");
|
|
await WriteLineAsync(remoteSocket, "LEAF REMOTE1", cts.Token);
|
|
await handshakeTask;
|
|
|
|
leaf.RemoteId.ShouldBe("REMOTE1");
|
|
}
|
|
|
|
// Go: TestLeafNodeRTT server/leafnode_test.go:488
|
|
[Fact]
|
|
public async Task Leaf_connection_inbound_handshake()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformInboundHandshakeAsync("SERVER1", cts.Token);
|
|
await WriteLineAsync(remoteSocket, "LEAF REMOTE2", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldBe("LEAF SERVER1");
|
|
await handshakeTask;
|
|
|
|
leaf.RemoteId.ShouldBe("REMOTE2");
|
|
}
|
|
|
|
// Go: TestLeafNodePermissions server/leafnode_test.go:1267
|
|
[Fact]
|
|
public async Task Leaf_LS_plus_sends_subscription_interest()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldStartWith("LEAF ");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE", cts.Token);
|
|
await handshakeTask;
|
|
|
|
await leaf.SendLsPlusAsync("$G", "test.subject", null, cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldBe("LS+ $G test.subject");
|
|
}
|
|
|
|
// Go: TestLeafNodePermissions server/leafnode_test.go:1267
|
|
[Fact]
|
|
public async Task Leaf_LS_minus_sends_unsubscription()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldStartWith("LEAF ");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE", cts.Token);
|
|
await handshakeTask;
|
|
|
|
await leaf.SendLsMinusAsync("$G", "test.subject", null, cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldBe("LS- $G test.subject");
|
|
}
|
|
|
|
// Go: TestLeafNodePermissions server/leafnode_test.go:1267
|
|
[Fact]
|
|
public async Task Leaf_LS_plus_with_queue_group()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldStartWith("LEAF ");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE", cts.Token);
|
|
await handshakeTask;
|
|
|
|
await leaf.SendLsPlusAsync("$G", "queue.subject", "workers", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldBe("LS+ $G queue.subject workers");
|
|
}
|
|
|
|
// Go: TestLeafNodeInterestPropagationDaisychain server/leafnode_test.go:3953
|
|
[Fact]
|
|
public async Task Leaf_receives_remote_subscription()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldStartWith("LEAF ");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE", cts.Token);
|
|
await handshakeTask;
|
|
|
|
var received = new TaskCompletionSource<RemoteSubscription>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
leaf.RemoteSubscriptionReceived = sub =>
|
|
{
|
|
received.TrySetResult(sub);
|
|
return Task.CompletedTask;
|
|
};
|
|
leaf.StartLoop(cts.Token);
|
|
|
|
await WriteLineAsync(remoteSocket, "LS+ $G events.>", cts.Token);
|
|
var result = await received.Task.WaitAsync(cts.Token);
|
|
result.Account.ShouldBe("$G");
|
|
result.Subject.ShouldBe("events.>");
|
|
result.IsRemoval.ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestLeafNodeInterestPropagationDaisychain server/leafnode_test.go:3953
|
|
[Fact]
|
|
public async Task Leaf_receives_remote_unsubscription()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldStartWith("LEAF ");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE", cts.Token);
|
|
await handshakeTask;
|
|
|
|
var received = new TaskCompletionSource<RemoteSubscription>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
leaf.RemoteSubscriptionReceived = sub =>
|
|
{
|
|
if (sub.IsRemoval)
|
|
received.TrySetResult(sub);
|
|
return Task.CompletedTask;
|
|
};
|
|
leaf.StartLoop(cts.Token);
|
|
|
|
await WriteLineAsync(remoteSocket, "LS+ $G events.>", cts.Token);
|
|
await PollHelper.YieldForAsync(100);
|
|
await WriteLineAsync(remoteSocket, "LS- $G events.>", cts.Token);
|
|
|
|
var result = await received.Task.WaitAsync(cts.Token);
|
|
result.IsRemoval.ShouldBeTrue();
|
|
result.Subject.ShouldBe("events.>");
|
|
}
|
|
|
|
// Go: TestLeafNodeOriginClusterInfo server/leafnode_test.go:1942
|
|
[Fact]
|
|
public async Task Leaf_handshake_propagates_JetStream_domain()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket) { JetStreamDomain = "hub-domain" };
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
var line = await ReadLineAsync(remoteSocket, cts.Token);
|
|
line.ShouldBe("LEAF HUB domain=hub-domain");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE domain=spoke-domain", cts.Token);
|
|
await handshakeTask;
|
|
|
|
leaf.RemoteJetStreamDomain.ShouldBe("spoke-domain");
|
|
}
|
|
|
|
// Go: TestLeafNodeRemoteIsHub server/leafnode_test.go:1177
|
|
[Fact]
|
|
public async Task Leaf_manager_solicited_connection_backoff()
|
|
{
|
|
// Verify the exponential backoff computation
|
|
LeafNodeManager.ComputeBackoff(0).ShouldBe(TimeSpan.FromSeconds(1));
|
|
LeafNodeManager.ComputeBackoff(1).ShouldBe(TimeSpan.FromSeconds(2));
|
|
LeafNodeManager.ComputeBackoff(2).ShouldBe(TimeSpan.FromSeconds(4));
|
|
LeafNodeManager.ComputeBackoff(3).ShouldBe(TimeSpan.FromSeconds(8));
|
|
LeafNodeManager.ComputeBackoff(4).ShouldBe(TimeSpan.FromSeconds(16));
|
|
LeafNodeManager.ComputeBackoff(5).ShouldBe(TimeSpan.FromSeconds(32));
|
|
LeafNodeManager.ComputeBackoff(6).ShouldBe(TimeSpan.FromSeconds(60));
|
|
LeafNodeManager.ComputeBackoff(7).ShouldBe(TimeSpan.FromSeconds(60));
|
|
LeafNodeManager.ComputeBackoff(-1).ShouldBe(TimeSpan.FromSeconds(1));
|
|
}
|
|
|
|
// Go: TestLeafNodeHubWithGateways server/leafnode_test.go:1584
|
|
[Fact]
|
|
public async Task Leaf_hub_spoke_message_round_trip()
|
|
{
|
|
await using var fixture = await LeafFixture.StartAsync();
|
|
|
|
await using var hubConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Hub.Port}",
|
|
});
|
|
await hubConn.ConnectAsync();
|
|
await using var spokeConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Spoke.Port}",
|
|
});
|
|
await spokeConn.ConnectAsync();
|
|
|
|
await using var sub = await spokeConn.SubscribeCoreAsync<string>("leaf.roundtrip");
|
|
await spokeConn.PingAsync();
|
|
await fixture.WaitForRemoteInterestOnHubAsync("leaf.roundtrip");
|
|
|
|
await hubConn.PublishAsync("leaf.roundtrip", "round-trip-msg");
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
(await sub.Msgs.ReadAsync(cts.Token)).Data.ShouldBe("round-trip-msg");
|
|
}
|
|
|
|
// Go: TestLeafNodeStreamAndShadowSubs server/leafnode_test.go:6176
|
|
[Fact]
|
|
public async Task Leaf_spoke_to_hub_message_delivery()
|
|
{
|
|
await using var fixture = await LeafFixture.StartAsync();
|
|
|
|
await using var hubConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Hub.Port}",
|
|
});
|
|
await hubConn.ConnectAsync();
|
|
await using var spokeConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Spoke.Port}",
|
|
});
|
|
await spokeConn.ConnectAsync();
|
|
|
|
await using var sub = await hubConn.SubscribeCoreAsync<string>("leaf.reverse");
|
|
await hubConn.PingAsync();
|
|
await fixture.WaitForRemoteInterestOnSpokeAsync("leaf.reverse");
|
|
|
|
await spokeConn.PublishAsync("leaf.reverse", "reverse-msg");
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
(await sub.Msgs.ReadAsync(cts.Token)).Data.ShouldBe("reverse-msg");
|
|
}
|
|
|
|
// Go: TestLeafNodeQueueGroupDistribution server/leafnode_test.go:4021
|
|
[Fact]
|
|
public async Task Leaf_queue_subscription_delivery()
|
|
{
|
|
await using var fixture = await LeafFixture.StartAsync();
|
|
|
|
await using var hubConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Hub.Port}",
|
|
});
|
|
await hubConn.ConnectAsync();
|
|
await using var spokeConn = new NatsConnection(new NatsOpts
|
|
{
|
|
Url = $"nats://127.0.0.1:{fixture.Spoke.Port}",
|
|
});
|
|
await spokeConn.ConnectAsync();
|
|
|
|
await using var sub = await spokeConn.SubscribeCoreAsync<string>("leaf.queue", queueGroup: "workers");
|
|
await spokeConn.PingAsync();
|
|
await fixture.WaitForRemoteInterestOnHubAsync("leaf.queue");
|
|
|
|
await hubConn.PublishAsync("leaf.queue", "queue-msg");
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
(await sub.Msgs.ReadAsync(cts.Token)).Data.ShouldBe("queue-msg");
|
|
}
|
|
|
|
// Go: TestLeafNodeDuplicateMsg server/leafnode_test.go:6513
|
|
[Fact]
|
|
public async Task Leaf_no_remote_interest_for_unsubscribed_subject()
|
|
{
|
|
await using var fixture = await LeafFixture.StartAsync();
|
|
fixture.Hub.HasRemoteInterest("nonexistent.leaf.subject").ShouldBeFalse();
|
|
fixture.Spoke.HasRemoteInterest("nonexistent.leaf.subject").ShouldBeFalse();
|
|
}
|
|
|
|
// Go: TestLeafNodePermissions server/leafnode_test.go:1267
|
|
[Fact]
|
|
public async Task Leaf_connection_LMSG_sends_message()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldStartWith("LEAF ");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE", cts.Token);
|
|
await handshakeTask;
|
|
|
|
var payload = Encoding.UTF8.GetBytes("hello-leaf");
|
|
await leaf.SendMessageAsync("$G", "test.msg", "reply.to", payload, cts.Token);
|
|
|
|
var line = await ReadLineAsync(remoteSocket, cts.Token);
|
|
line.ShouldBe("LMSG $G test.msg reply.to 10");
|
|
|
|
// Read payload + CRLF
|
|
var buf = new byte[12]; // 10 payload + 2 CRLF
|
|
var offset = 0;
|
|
while (offset < 12)
|
|
{
|
|
var n = await remoteSocket.ReceiveAsync(buf.AsMemory(offset), SocketFlags.None, cts.Token);
|
|
offset += n;
|
|
}
|
|
|
|
Encoding.UTF8.GetString(buf, 0, 10).ShouldBe("hello-leaf");
|
|
}
|
|
|
|
// Go: TestLeafNodeIsolatedLeafSubjectPropagationGlobal server/leafnode_test.go:10280
|
|
[Fact]
|
|
public async Task Leaf_LMSG_with_no_reply()
|
|
{
|
|
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
|
|
using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
|
await remoteSocket.ConnectAsync(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
|
|
using var leafSocket = await listener.AcceptSocketAsync();
|
|
await using var leaf = new LeafConnection(leafSocket);
|
|
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
var handshakeTask = leaf.PerformOutboundHandshakeAsync("HUB", cts.Token);
|
|
(await ReadLineAsync(remoteSocket, cts.Token)).ShouldStartWith("LEAF ");
|
|
await WriteLineAsync(remoteSocket, "LEAF SPOKE", cts.Token);
|
|
await handshakeTask;
|
|
|
|
await leaf.SendMessageAsync("$G", "no.reply", null, "data"u8.ToArray(), cts.Token);
|
|
var line = await ReadLineAsync(remoteSocket, cts.Token);
|
|
line.ShouldBe("LMSG $G no.reply - 4");
|
|
}
|
|
|
|
// ════════════════════════════════════════════════════════════════════
|
|
// Helpers
|
|
// ════════════════════════════════════════════════════════════════════
|
|
|
|
private static async Task<string> ReadLineAsync(Socket socket, CancellationToken ct)
|
|
{
|
|
var bytes = new List<byte>(64);
|
|
var single = new byte[1];
|
|
while (true)
|
|
{
|
|
var read = await socket.ReceiveAsync(single, SocketFlags.None, ct);
|
|
if (read == 0)
|
|
break;
|
|
if (single[0] == (byte)'\n')
|
|
break;
|
|
if (single[0] != (byte)'\r')
|
|
bytes.Add(single[0]);
|
|
}
|
|
|
|
return Encoding.ASCII.GetString([.. bytes]);
|
|
}
|
|
|
|
private static Task WriteLineAsync(Socket socket, string line, CancellationToken ct)
|
|
=> socket.SendAsync(Encoding.ASCII.GetBytes($"{line}\r\n"), SocketFlags.None, ct).AsTask();
|
|
}
|
|
|
|
// ════════════════════════════════════════════════════════════════════════
|
|
// Shared Fixtures
|
|
// ════════════════════════════════════════════════════════════════════════
|
|
|
|
internal sealed class TwoGatewayFixture : IAsyncDisposable
|
|
{
|
|
private readonly CancellationTokenSource _localCts;
|
|
private readonly CancellationTokenSource _remoteCts;
|
|
|
|
private TwoGatewayFixture(NatsServer local, NatsServer remote, CancellationTokenSource localCts, CancellationTokenSource remoteCts)
|
|
{
|
|
Local = local;
|
|
Remote = remote;
|
|
_localCts = localCts;
|
|
_remoteCts = remoteCts;
|
|
}
|
|
|
|
public NatsServer Local { get; }
|
|
public NatsServer Remote { get; }
|
|
|
|
public static async Task<TwoGatewayFixture> StartAsync()
|
|
{
|
|
var localOptions = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Gateway = new GatewayOptions
|
|
{
|
|
Name = "LOCAL",
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
},
|
|
};
|
|
|
|
var local = new NatsServer(localOptions, NullLoggerFactory.Instance);
|
|
var localCts = new CancellationTokenSource();
|
|
_ = local.StartAsync(localCts.Token);
|
|
await local.WaitForReadyAsync();
|
|
|
|
var remoteOptions = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Gateway = new GatewayOptions
|
|
{
|
|
Name = "REMOTE",
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Remotes = [local.GatewayListen!],
|
|
},
|
|
};
|
|
|
|
var remote = new NatsServer(remoteOptions, NullLoggerFactory.Instance);
|
|
var remoteCts = new CancellationTokenSource();
|
|
_ = remote.StartAsync(remoteCts.Token);
|
|
await remote.WaitForReadyAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => !((local.Stats.Gateways == 0 || remote.Stats.Gateways == 0)));
|
|
|
|
return new TwoGatewayFixture(local, remote, localCts, remoteCts);
|
|
}
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
await _localCts.CancelAsync();
|
|
await _remoteCts.CancelAsync();
|
|
Local.Dispose();
|
|
Remote.Dispose();
|
|
_localCts.Dispose();
|
|
_remoteCts.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Leaf fixture duplicated here to avoid cross-namespace dependencies.
|
|
/// Uses hub and spoke servers connected via leaf node protocol.
|
|
/// </summary>
|
|
internal sealed class LeafFixture : IAsyncDisposable
|
|
{
|
|
private readonly CancellationTokenSource _hubCts;
|
|
private readonly CancellationTokenSource _spokeCts;
|
|
|
|
private LeafFixture(NatsServer hub, NatsServer spoke, CancellationTokenSource hubCts, CancellationTokenSource spokeCts)
|
|
{
|
|
Hub = hub;
|
|
Spoke = spoke;
|
|
_hubCts = hubCts;
|
|
_spokeCts = spokeCts;
|
|
}
|
|
|
|
public NatsServer Hub { get; }
|
|
public NatsServer Spoke { get; }
|
|
|
|
public static async Task<LeafFixture> StartAsync()
|
|
{
|
|
var hubOptions = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
LeafNode = new LeafNodeOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
},
|
|
};
|
|
|
|
var hub = new NatsServer(hubOptions, NullLoggerFactory.Instance);
|
|
var hubCts = new CancellationTokenSource();
|
|
_ = hub.StartAsync(hubCts.Token);
|
|
await hub.WaitForReadyAsync();
|
|
|
|
var spokeOptions = new NatsOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
LeafNode = new LeafNodeOptions
|
|
{
|
|
Host = "127.0.0.1",
|
|
Port = 0,
|
|
Remotes = [hub.LeafListen!],
|
|
},
|
|
};
|
|
|
|
var spoke = new NatsServer(spokeOptions, NullLoggerFactory.Instance);
|
|
var spokeCts = new CancellationTokenSource();
|
|
_ = spoke.StartAsync(spokeCts.Token);
|
|
await spoke.WaitForReadyAsync();
|
|
|
|
await PollHelper.WaitUntilAsync(() => !((hub.Stats.Leafs == 0 || spoke.Stats.Leafs == 0)));
|
|
|
|
return new LeafFixture(hub, spoke, hubCts, spokeCts);
|
|
}
|
|
|
|
public async Task WaitForRemoteInterestOnHubAsync(string subject)
|
|
{
|
|
await PollHelper.WaitOrThrowAsync(
|
|
() => Hub.HasRemoteInterest(subject),
|
|
$"Timed out waiting for remote interest on hub for '{subject}'.");
|
|
}
|
|
|
|
public async Task WaitForRemoteInterestOnSpokeAsync(string subject)
|
|
{
|
|
await PollHelper.WaitOrThrowAsync(
|
|
() => Spoke.HasRemoteInterest(subject),
|
|
$"Timed out waiting for remote interest on spoke for '{subject}'.");
|
|
}
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
await _spokeCts.CancelAsync();
|
|
await _hubCts.CancelAsync();
|
|
Spoke.Dispose();
|
|
Hub.Dispose();
|
|
_spokeCts.Dispose();
|
|
_hubCts.Dispose();
|
|
}
|
|
}
|