feat: Wave 6 batch 2 — accounts/auth, gateways, routes, JetStream API, JetStream cluster tests
Add comprehensive Go-parity test coverage across 5 subsystems: - Accounts/Auth: isolation, import/export, auth mechanisms, permissions (82 tests) - Gateways: connection, forwarding, interest mode, config (106 tests) - Routes: connection, subscription, forwarding, config validation (78 tests) - JetStream API: stream/consumer CRUD, pub/sub, features, admin (234 tests) - JetStream Cluster: streams, consumers, failover, meta (108 tests) Total: ~608 new test annotations across 22 files (+13,844 lines) All tests pass individually; suite total: 2,283 passing, 3 skipped
This commit is contained in:
Joseph Doherty
820
tests/NATS.Server.Tests/Routes/RouteForwardingTests.cs
Normal file
820
tests/NATS.Server.Tests/Routes/RouteForwardingTests.cs
Normal file
@@ -0,0 +1,820 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using NATS.Client.Core;
|
||||
using NATS.Server.Auth;
|
||||
using NATS.Server.Configuration;
|
||||
using NATS.Server.Routes;
|
||||
using NATS.Server.Subscriptions;
|
||||
|
||||
namespace NATS.Server.Tests.Routes;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for route message forwarding (RMSG), reply propagation, payload delivery,
|
||||
/// and cross-cluster message routing.
|
||||
/// Ported from Go: server/routes_test.go.
|
||||
/// </summary>
|
||||
public class RouteForwardingTests
|
||||
{
|
||||
// -- Helpers --
|
||||
|
||||
private static async Task<(NatsServer Server, CancellationTokenSource Cts)> StartServerAsync(
|
||||
NatsOptions opts)
|
||||
{
|
||||
var server = new NatsServer(opts, NullLoggerFactory.Instance);
|
||||
var cts = new CancellationTokenSource();
|
||||
_ = server.StartAsync(cts.Token);
|
||||
await server.WaitForReadyAsync();
|
||||
return (server, cts);
|
||||
}
|
||||
|
||||
private static NatsOptions MakeClusterOpts(string? clusterName = null, string? seed = null)
|
||||
{
|
||||
return new NatsOptions
|
||||
{
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
Cluster = new ClusterOptions
|
||||
{
|
||||
Name = clusterName ?? Guid.NewGuid().ToString("N"),
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
Routes = seed is null ? [] : [seed],
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private static async Task WaitForRouteFormation(NatsServer a, NatsServer b, int timeoutMs = 5000)
|
||||
{
|
||||
using var timeout = new CancellationTokenSource(timeoutMs);
|
||||
while (!timeout.IsCancellationRequested &&
|
||||
(Interlocked.Read(ref a.Stats.Routes) == 0 ||
|
||||
Interlocked.Read(ref b.Stats.Routes) == 0))
|
||||
{
|
||||
await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task WaitForCondition(Func<bool> predicate, int timeoutMs = 5000)
|
||||
{
|
||||
using var cts = new CancellationTokenSource(timeoutMs);
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
if (predicate()) return;
|
||||
await Task.Delay(20, cts.Token).ContinueWith(_ => { }, TaskScheduler.Default);
|
||||
}
|
||||
|
||||
throw new TimeoutException("Condition not met.");
|
||||
}
|
||||
|
||||
private static async Task DisposeServers(params (NatsServer Server, CancellationTokenSource Cts)[] servers)
|
||||
{
|
||||
foreach (var (server, cts) in servers)
|
||||
{
|
||||
await cts.CancelAsync();
|
||||
server.Dispose();
|
||||
cts.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
// -- Tests: RMSG forwarding --
|
||||
|
||||
// Go: TestSeedSolicitWorks server/routes_test.go:365 (message forwarding)
|
||||
[Fact]
|
||||
public async Task RMSG_forwards_published_message_to_remote_subscriber()
|
||||
{
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
var a = await StartServerAsync(MakeClusterOpts(cluster));
|
||||
var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
await using var subscriber = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{a.Server.Port}",
|
||||
});
|
||||
await subscriber.ConnectAsync();
|
||||
await using var sub = await subscriber.SubscribeCoreAsync<string>("rmsg.test");
|
||||
await subscriber.PingAsync();
|
||||
await WaitForCondition(() => b.Server.HasRemoteInterest("rmsg.test"));
|
||||
|
||||
await using var publisher = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{b.Server.Port}",
|
||||
});
|
||||
await publisher.ConnectAsync();
|
||||
await publisher.PublishAsync("rmsg.test", "routed-payload");
|
||||
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
var msg = await sub.Msgs.ReadAsync(timeout.Token);
|
||||
msg.Data.ShouldBe("routed-payload");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: Request-Reply across routes via raw socket with reply-to
|
||||
[Fact]
|
||||
public async Task Request_reply_works_across_routed_servers()
|
||||
{
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
var optsA = MakeClusterOpts(cluster);
|
||||
optsA.Cluster!.PoolSize = 1;
|
||||
var a = await StartServerAsync(optsA);
|
||||
var optsB = MakeClusterOpts(cluster, a.Server.ClusterListen!);
|
||||
optsB.Cluster!.PoolSize = 1;
|
||||
var b = await StartServerAsync(optsB);
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
// Responder on server A: subscribe via raw socket to get exact wire control
|
||||
using var responderSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await responderSock.ConnectAsync(IPAddress.Loopback, a.Server.Port);
|
||||
var buf = new byte[4096];
|
||||
_ = await responderSock.ReceiveAsync(buf); // INFO
|
||||
await responderSock.SendAsync(Encoding.ASCII.GetBytes("CONNECT {}\r\nSUB service.echo 1\r\nPING\r\n"));
|
||||
await ReadUntilAsync(responderSock, "PONG");
|
||||
await WaitForCondition(() => b.Server.HasRemoteInterest("service.echo"));
|
||||
|
||||
// Requester on server B: subscribe to reply inbox via raw socket
|
||||
using var requesterSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await requesterSock.ConnectAsync(IPAddress.Loopback, b.Server.Port);
|
||||
_ = await requesterSock.ReceiveAsync(buf); // INFO
|
||||
var replyInbox = $"_INBOX.{Guid.NewGuid():N}";
|
||||
await requesterSock.SendAsync(Encoding.ASCII.GetBytes(
|
||||
$"CONNECT {{}}\r\nSUB {replyInbox} 2\r\nPING\r\n"));
|
||||
await ReadUntilAsync(requesterSock, "PONG");
|
||||
await WaitForCondition(() => a.Server.HasRemoteInterest(replyInbox));
|
||||
|
||||
// Publish request with reply-to from B
|
||||
await requesterSock.SendAsync(Encoding.ASCII.GetBytes(
|
||||
$"PUB service.echo {replyInbox} 4\r\nping\r\nPING\r\n"));
|
||||
await ReadUntilAsync(requesterSock, "PONG");
|
||||
|
||||
// Read the request on A, verify reply-to
|
||||
var requestData = await ReadUntilAsync(responderSock, "ping");
|
||||
requestData.ShouldContain($"MSG service.echo 1 {replyInbox} 4");
|
||||
requestData.ShouldContain("ping");
|
||||
|
||||
// Publish reply from A to the reply-to subject
|
||||
await responderSock.SendAsync(Encoding.ASCII.GetBytes(
|
||||
$"PUB {replyInbox} 4\r\npong\r\nPING\r\n"));
|
||||
await ReadUntilAsync(responderSock, "PONG");
|
||||
|
||||
// Read the reply on B
|
||||
var replyData = await ReadUntilAsync(requesterSock, "pong");
|
||||
replyData.ShouldContain($"MSG {replyInbox} 2 4");
|
||||
replyData.ShouldContain("pong");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: RMSG wire-level parsing
|
||||
[Fact]
|
||||
public async Task RMSG_wire_frame_delivers_payload_to_handler()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
RouteMessage? receivedMsg = null;
|
||||
route.RoutedMessageReceived = msg =>
|
||||
{
|
||||
receivedMsg = msg;
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
route.StartFrameLoop(timeout.Token);
|
||||
|
||||
var payload = "hello-world";
|
||||
var frame = $"RMSG $G test.subject - {payload.Length}\r\n{payload}\r\n";
|
||||
await remote.SendAsync(Encoding.ASCII.GetBytes(frame), SocketFlags.None, timeout.Token);
|
||||
|
||||
await WaitForCondition(() => receivedMsg != null);
|
||||
receivedMsg.ShouldNotBeNull();
|
||||
receivedMsg!.Subject.ShouldBe("test.subject");
|
||||
receivedMsg.ReplyTo.ShouldBeNull();
|
||||
Encoding.UTF8.GetString(receivedMsg.Payload.Span).ShouldBe("hello-world");
|
||||
}
|
||||
|
||||
// Go: RMSG with reply subject
|
||||
[Fact]
|
||||
public async Task RMSG_wire_frame_includes_reply_to()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
RouteMessage? receivedMsg = null;
|
||||
route.RoutedMessageReceived = msg =>
|
||||
{
|
||||
receivedMsg = msg;
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
route.StartFrameLoop(timeout.Token);
|
||||
|
||||
var payload = "data";
|
||||
var frame = $"RMSG $G test.subject _INBOX.abc123 {payload.Length}\r\n{payload}\r\n";
|
||||
await remote.SendAsync(Encoding.ASCII.GetBytes(frame), SocketFlags.None, timeout.Token);
|
||||
|
||||
await WaitForCondition(() => receivedMsg != null);
|
||||
receivedMsg.ShouldNotBeNull();
|
||||
receivedMsg!.Subject.ShouldBe("test.subject");
|
||||
receivedMsg.ReplyTo.ShouldBe("_INBOX.abc123");
|
||||
}
|
||||
|
||||
// Go: RMSG with account
|
||||
[Fact]
|
||||
public async Task RMSG_wire_frame_with_account_scope()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
RouteMessage? receivedMsg = null;
|
||||
route.RoutedMessageReceived = msg =>
|
||||
{
|
||||
receivedMsg = msg;
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
route.StartFrameLoop(timeout.Token);
|
||||
|
||||
var payload = "acct-data";
|
||||
var frame = $"RMSG MYACCOUNT test.sub - {payload.Length}\r\n{payload}\r\n";
|
||||
await remote.SendAsync(Encoding.ASCII.GetBytes(frame), SocketFlags.None, timeout.Token);
|
||||
|
||||
await WaitForCondition(() => receivedMsg != null);
|
||||
receivedMsg.ShouldNotBeNull();
|
||||
receivedMsg!.Account.ShouldBe("MYACCOUNT");
|
||||
receivedMsg.Subject.ShouldBe("test.sub");
|
||||
}
|
||||
|
||||
// Go: RMSG with zero-length payload
|
||||
[Fact]
|
||||
public async Task RMSG_wire_frame_with_empty_payload()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
RouteMessage? receivedMsg = null;
|
||||
route.RoutedMessageReceived = msg =>
|
||||
{
|
||||
receivedMsg = msg;
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
route.StartFrameLoop(timeout.Token);
|
||||
|
||||
var frame = "RMSG $G empty.test - 0\r\n\r\n";
|
||||
await remote.SendAsync(Encoding.ASCII.GetBytes(frame), SocketFlags.None, timeout.Token);
|
||||
|
||||
await WaitForCondition(() => receivedMsg != null);
|
||||
receivedMsg.ShouldNotBeNull();
|
||||
receivedMsg!.Subject.ShouldBe("empty.test");
|
||||
receivedMsg.Payload.Length.ShouldBe(0);
|
||||
}
|
||||
|
||||
// Go: TestServerRoutesWithClients server/routes_test.go:216 (large payload)
|
||||
[Fact]
|
||||
public async Task Large_payload_forwarded_across_route()
|
||||
{
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
var a = await StartServerAsync(MakeClusterOpts(cluster));
|
||||
var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
await using var subscriber = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{a.Server.Port}",
|
||||
});
|
||||
await subscriber.ConnectAsync();
|
||||
await using var sub = await subscriber.SubscribeCoreAsync<byte[]>("large.payload");
|
||||
await subscriber.PingAsync();
|
||||
await WaitForCondition(() => b.Server.HasRemoteInterest("large.payload"));
|
||||
|
||||
await using var publisher = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{b.Server.Port}",
|
||||
});
|
||||
await publisher.ConnectAsync();
|
||||
|
||||
var data = new byte[8192];
|
||||
Random.Shared.NextBytes(data);
|
||||
await publisher.PublishAsync("large.payload", data);
|
||||
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
var msg = await sub.Msgs.ReadAsync(timeout.Token);
|
||||
msg.Data.ShouldBe(data);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: TestRoutePool server/routes_test.go:1966 (message sent and received across pool)
|
||||
[Fact]
|
||||
public async Task Messages_flow_across_route_with_pool_size()
|
||||
{
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
var optsA = new NatsOptions
|
||||
{
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
Cluster = new ClusterOptions
|
||||
{
|
||||
Name = cluster,
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
PoolSize = 2,
|
||||
},
|
||||
};
|
||||
var a = await StartServerAsync(optsA);
|
||||
|
||||
var optsB = new NatsOptions
|
||||
{
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
Cluster = new ClusterOptions
|
||||
{
|
||||
Name = cluster,
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
PoolSize = 2,
|
||||
Routes = [a.Server.ClusterListen!],
|
||||
},
|
||||
};
|
||||
var b = await StartServerAsync(optsB);
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
await using var subscriber = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{b.Server.Port}",
|
||||
});
|
||||
await subscriber.ConnectAsync();
|
||||
await using var sub = await subscriber.SubscribeCoreAsync<string>("pool.forward");
|
||||
await subscriber.PingAsync();
|
||||
await WaitForCondition(() => a.Server.HasRemoteInterest("pool.forward"));
|
||||
|
||||
await using var publisher = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{a.Server.Port}",
|
||||
});
|
||||
await publisher.ConnectAsync();
|
||||
|
||||
const int messageCount = 10;
|
||||
for (var i = 0; i < messageCount; i++)
|
||||
await publisher.PublishAsync("pool.forward", $"msg-{i}");
|
||||
|
||||
// With PoolSize=2, each message may be forwarded on multiple route connections.
|
||||
// Collect all received messages and verify each expected one arrived at least once.
|
||||
var received = new HashSet<string>();
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
||||
while (received.Count < messageCount)
|
||||
{
|
||||
var msg = await sub.Msgs.ReadAsync(timeout.Token);
|
||||
msg.Data.ShouldNotBeNull();
|
||||
received.Add(msg.Data!);
|
||||
}
|
||||
|
||||
for (var i = 0; i < messageCount; i++)
|
||||
received.ShouldContain($"msg-{i}");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: TestRoutePerAccount server/routes_test.go:2539 (account-scoped delivery)
|
||||
[Fact]
|
||||
public async Task Account_scoped_RMSG_delivers_to_correct_account()
|
||||
{
|
||||
var users = new User[]
|
||||
{
|
||||
new() { Username = "ua", Password = "p", Account = "A" },
|
||||
new() { Username = "ub", Password = "p", Account = "B" },
|
||||
};
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
|
||||
var optsA = new NatsOptions
|
||||
{
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
Users = users,
|
||||
Cluster = new ClusterOptions
|
||||
{
|
||||
Name = cluster,
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
},
|
||||
};
|
||||
var a = await StartServerAsync(optsA);
|
||||
|
||||
var optsB = new NatsOptions
|
||||
{
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
Users = users,
|
||||
Cluster = new ClusterOptions
|
||||
{
|
||||
Name = cluster,
|
||||
Host = "127.0.0.1",
|
||||
Port = 0,
|
||||
Routes = [a.Server.ClusterListen!],
|
||||
},
|
||||
};
|
||||
var b = await StartServerAsync(optsB);
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
// Account A subscriber on server B
|
||||
await using var subConn = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://ua:p@127.0.0.1:{b.Server.Port}",
|
||||
});
|
||||
await subConn.ConnectAsync();
|
||||
await using var sub = await subConn.SubscribeCoreAsync<string>("acct.fwd");
|
||||
await subConn.PingAsync();
|
||||
|
||||
await WaitForCondition(() => a.Server.HasRemoteInterest("A", "acct.fwd"));
|
||||
|
||||
// Publish from account A on server A
|
||||
await using var pubConn = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://ua:p@127.0.0.1:{a.Server.Port}",
|
||||
});
|
||||
await pubConn.ConnectAsync();
|
||||
await pubConn.PublishAsync("acct.fwd", "from-a");
|
||||
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
var msg = await sub.Msgs.ReadAsync(timeout.Token);
|
||||
msg.Data.ShouldBe("from-a");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: bidirectional forwarding
|
||||
[Fact]
|
||||
public async Task Bidirectional_message_forwarding_across_route()
|
||||
{
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
var a = await StartServerAsync(MakeClusterOpts(cluster));
|
||||
var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
await using var ncA = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{a.Server.Port}",
|
||||
});
|
||||
await ncA.ConnectAsync();
|
||||
|
||||
await using var ncB = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{b.Server.Port}",
|
||||
});
|
||||
await ncB.ConnectAsync();
|
||||
|
||||
// Sub on A, pub from B
|
||||
await using var subOnA = await ncA.SubscribeCoreAsync<string>("bidir.a");
|
||||
// Sub on B, pub from A
|
||||
await using var subOnB = await ncB.SubscribeCoreAsync<string>("bidir.b");
|
||||
await ncA.PingAsync();
|
||||
await ncB.PingAsync();
|
||||
|
||||
await WaitForCondition(() =>
|
||||
b.Server.HasRemoteInterest("bidir.a") && a.Server.HasRemoteInterest("bidir.b"));
|
||||
|
||||
await ncB.PublishAsync("bidir.a", "from-b");
|
||||
await ncA.PublishAsync("bidir.b", "from-a");
|
||||
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
var msgA = await subOnA.Msgs.ReadAsync(timeout.Token);
|
||||
var msgB = await subOnB.Msgs.ReadAsync(timeout.Token);
|
||||
msgA.Data.ShouldBe("from-b");
|
||||
msgB.Data.ShouldBe("from-a");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: Route forwarding with reply (non-request-reply, just reply subject)
|
||||
[Fact]
|
||||
public async Task Message_with_reply_subject_forwarded_across_route()
|
||||
{
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
var a = await StartServerAsync(MakeClusterOpts(cluster));
|
||||
var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
await using var subscriber = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{a.Server.Port}",
|
||||
});
|
||||
await subscriber.ConnectAsync();
|
||||
await using var sub = await subscriber.SubscribeCoreAsync<string>("reply.subject.test");
|
||||
await subscriber.PingAsync();
|
||||
await WaitForCondition(() => b.Server.HasRemoteInterest("reply.subject.test"));
|
||||
|
||||
// Use raw socket to publish with reply-to
|
||||
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await sock.ConnectAsync(IPAddress.Loopback, b.Server.Port);
|
||||
var buf = new byte[4096];
|
||||
_ = await sock.ReceiveAsync(buf); // INFO
|
||||
await sock.SendAsync(Encoding.ASCII.GetBytes(
|
||||
"CONNECT {}\r\nPUB reply.subject.test _INBOX.reply123 5\r\nHello\r\nPING\r\n"));
|
||||
await ReadUntilAsync(sock, "PONG");
|
||||
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
var msg = await sub.Msgs.ReadAsync(timeout.Token);
|
||||
msg.Data.ShouldBe("Hello");
|
||||
msg.ReplyTo.ShouldBe("_INBOX.reply123");
|
||||
sock.Dispose();
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: Multiple messages with varying payloads
|
||||
[Fact]
|
||||
public async Task Multiple_different_subjects_forwarded_simultaneously()
|
||||
{
|
||||
var cluster = Guid.NewGuid().ToString("N");
|
||||
var a = await StartServerAsync(MakeClusterOpts(cluster));
|
||||
var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));
|
||||
|
||||
try
|
||||
{
|
||||
await WaitForRouteFormation(a.Server, b.Server);
|
||||
|
||||
await using var ncA = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{a.Server.Port}",
|
||||
});
|
||||
await ncA.ConnectAsync();
|
||||
|
||||
await using var sub1 = await ncA.SubscribeCoreAsync<string>("multi.a");
|
||||
await using var sub2 = await ncA.SubscribeCoreAsync<string>("multi.b");
|
||||
await using var sub3 = await ncA.SubscribeCoreAsync<string>("multi.c");
|
||||
await ncA.PingAsync();
|
||||
|
||||
await WaitForCondition(() =>
|
||||
b.Server.HasRemoteInterest("multi.a") &&
|
||||
b.Server.HasRemoteInterest("multi.b") &&
|
||||
b.Server.HasRemoteInterest("multi.c"));
|
||||
|
||||
await using var pub = new NatsConnection(new NatsOpts
|
||||
{
|
||||
Url = $"nats://127.0.0.1:{b.Server.Port}",
|
||||
});
|
||||
await pub.ConnectAsync();
|
||||
await pub.PublishAsync("multi.a", "alpha");
|
||||
await pub.PublishAsync("multi.b", "beta");
|
||||
await pub.PublishAsync("multi.c", "gamma");
|
||||
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
var msgA = await sub1.Msgs.ReadAsync(timeout.Token);
|
||||
var msgB = await sub2.Msgs.ReadAsync(timeout.Token);
|
||||
var msgC = await sub3.Msgs.ReadAsync(timeout.Token);
|
||||
|
||||
msgA.Data.ShouldBe("alpha");
|
||||
msgB.Data.ShouldBe("beta");
|
||||
msgC.Data.ShouldBe("gamma");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await DisposeServers(a, b);
|
||||
}
|
||||
}
|
||||
|
||||
// Go: SendRmsgAsync (send RMSG on RouteConnection)
|
||||
[Fact]
|
||||
public async Task RouteConnection_SendRmsgAsync_sends_valid_wire_frame()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
var payload = Encoding.UTF8.GetBytes("test-payload");
|
||||
await route.SendRmsgAsync("$G", "subject.test", "_INBOX.reply", payload, timeout.Token);
|
||||
|
||||
// Read the RMSG frame from the remote side, waiting until expected content arrives
|
||||
var data = await ReadUntilAsync(remote, "test-payload");
|
||||
data.ShouldContain("RMSG $G subject.test _INBOX.reply 12");
|
||||
data.ShouldContain("test-payload");
|
||||
}
|
||||
|
||||
// Go: SendRsPlusAsync (send RS+ on RouteConnection)
|
||||
[Fact]
|
||||
public async Task RouteConnection_SendRsPlusAsync_sends_valid_wire_frame()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
await route.SendRsPlusAsync("$G", "foo.bar", null, timeout.Token);
|
||||
var data = await ReadAllAvailableAsync(remote, timeout.Token);
|
||||
data.ShouldContain("RS+ $G foo.bar");
|
||||
}
|
||||
|
||||
// Go: SendRsMinusAsync (send RS- on RouteConnection)
|
||||
[Fact]
|
||||
public async Task RouteConnection_SendRsMinusAsync_sends_valid_wire_frame()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
await route.SendRsMinusAsync("$G", "foo.bar", null, timeout.Token);
|
||||
var data = await ReadAllAvailableAsync(remote, timeout.Token);
|
||||
data.ShouldContain("RS- $G foo.bar");
|
||||
}
|
||||
|
||||
// Go: SendRsPlusAsync with queue
|
||||
[Fact]
|
||||
public async Task RouteConnection_SendRsPlusAsync_with_queue_sends_valid_frame()
|
||||
{
|
||||
using var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
|
||||
using var remote = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
|
||||
await remote.ConnectAsync(IPAddress.Loopback, port);
|
||||
using var routeSock = await listener.AcceptSocketAsync();
|
||||
await using var route = new RouteConnection(routeSock);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL", timeout.Token);
|
||||
_ = await ReadLineAsync(remote, timeout.Token);
|
||||
await WriteLineAsync(remote, "ROUTE REMOTE", timeout.Token);
|
||||
await handshakeTask;
|
||||
|
||||
await route.SendRsPlusAsync("ACCT_A", "foo.bar", "myqueue", timeout.Token);
|
||||
var data = await ReadAllAvailableAsync(remote, timeout.Token);
|
||||
data.ShouldContain("RS+ ACCT_A foo.bar myqueue");
|
||||
}
|
||||
|
||||
// -- Wire-level 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();
|
||||
|
||||
private static async Task<string> ReadAllAvailableAsync(Socket socket, CancellationToken ct)
|
||||
{
|
||||
var sb = new StringBuilder();
|
||||
var buf = new byte[4096];
|
||||
|
||||
// First read blocks until at least some data arrives
|
||||
var n = await socket.ReceiveAsync(buf, SocketFlags.None, ct);
|
||||
if (n > 0)
|
||||
sb.Append(Encoding.ASCII.GetString(buf, 0, n));
|
||||
|
||||
// Drain any additional data that's already buffered
|
||||
while (n == buf.Length && socket.Available > 0)
|
||||
{
|
||||
n = await socket.ReceiveAsync(buf, SocketFlags.None, ct);
|
||||
if (n == 0) break;
|
||||
sb.Append(Encoding.ASCII.GetString(buf, 0, n));
|
||||
}
|
||||
|
||||
return sb.ToString();
|
||||
}
|
||||
|
||||
private static async Task<string> ReadUntilAsync(Socket sock, string expected)
|
||||
{
|
||||
var sb = new StringBuilder();
|
||||
var buf = new byte[4096];
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
while (!sb.ToString().Contains(expected, StringComparison.Ordinal))
|
||||
{
|
||||
var n = await sock.ReceiveAsync(buf, SocketFlags.None, cts.Token);
|
||||
if (n == 0) break;
|
||||
sb.Append(Encoding.ASCII.GetString(buf, 0, n));
|
||||
}
|
||||
|
||||
return sb.ToString();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user