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

test: add E2E gateway tests (cross-gateway messaging, interest-only)

Browse Source 
- GatewayFixture: polls /gatewayz monitoring endpoint until both servers
  report num_gateways >= 1, replacing Task.Delay with proper synchronization
- GatewayTests: two tests covering cross-gateway delivery and interest-only
  no-delivery behaviour, using double PingAsync() instead of Task.Delay
- LeafNodeTests: replace Task.Delay(500) with sub.PingAsync()+pub.PingAsync()
  to properly fence subscription propagation without timing dependencies
- Fix GatewayManager.StartAsync to read remotes from RemoteGateways (config-
  parsed) in addition to the legacy Remotes list, enabling config-file-driven
  outbound gateway connections
This commit is contained in:
Joseph Doherty
2026-03-12 19:45:54 -04:00
parent 5d9d1bebd5
commit 338f44b07b
4 changed files with 243 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/NATS.Server/Gateways/GatewayManager.cs
View File

@@ -210,7 +210,17 @@ public sealed class GatewayManager : IAsyncDisposable
_options.Port = ((IPEndPoint)_listener.LocalEndPoint!).Port;
_acceptLoopTask = Task.Run(() => AcceptLoopAsync(_cts.Token));
foreach (var remote in _options.Remotes.Distinct(StringComparer.OrdinalIgnoreCase))
// Collect outbound endpoints from both the legacy Remotes list and the
// config-parsed RemoteGateways list (populated by ConfigProcessor).
var endpoints = new HashSet<string>(_options.Remotes, StringComparer.OrdinalIgnoreCase);
foreach (var rgw in _options.RemoteGateways)
{
foreach (var uri in rgw.GetUrls())
endpoints.Add($"{uri.Host}:{uri.Port}");
}
foreach (var remote in endpoints)
_ = Task.Run(() => ConnectWithRetryAsync(remote, _cts.Token));
_logger.LogDebug("Gateway manager started (name={Name}, listen={Host}:{Port})",

50
tests/NATS.E2E.Tests/GatewayTests.cs Normal file
View File

@@ -0,0 +1,50 @@
using NATS.Client.Core;
using NATS.E2E.Tests.Infrastructure;
namespace NATS.E2E.Tests;
[Collection("E2E-Gateway")]
public class GatewayTests(GatewayFixture fixture)
{
[Fact]
public async Task Gateway_MessageCrossesGateway()
{
await using var pub = fixture.CreateClientA();
await using var sub = fixture.CreateClientB();
await pub.ConnectAsync();
await sub.ConnectAsync();
await using var subscription = await sub.SubscribeCoreAsync<string>("e2e.gw.cross");
// Ping both sides: subscriber's ping flushes the SUB to server B,
// publisher's ping ensures server A has received the propagated interest.
await sub.PingAsync();
await pub.PingAsync();
await pub.PublishAsync("e2e.gw.cross", "gateway-msg");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var msg = await subscription.Msgs.ReadAsync(cts.Token);
msg.Data.ShouldBe("gateway-msg");
}
[Fact]
public async Task Gateway_NoInterest_NoDelivery()
{
await using var pub = fixture.CreateClientA();
await using var sub = fixture.CreateClientB();
await pub.ConnectAsync();
await sub.ConnectAsync();
await using var subscription = await sub.SubscribeCoreAsync<string>("e2e.gw.listen");
// Ping both sides to ensure subscription interest has propagated before publishing.
await sub.PingAsync();
await pub.PingAsync();
await pub.PublishAsync("e2e.gw.nolisten", "should-not-arrive");
await pub.PublishAsync("e2e.gw.listen", "should-arrive");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var msg = await subscription.Msgs.ReadAsync(cts.Token);
msg.Data.ShouldBe("should-arrive");
}
}

111
tests/NATS.E2E.Tests/Infrastructure/GatewayFixture.cs Normal file
View File

@@ -0,0 +1,111 @@
using System.Net.Http.Json;
using NATS.Client.Core;
namespace NATS.E2E.Tests.Infrastructure;
public sealed class GatewayFixture : IAsyncLifetime
{
private NatsServerProcess _serverA = null!;
private NatsServerProcess _serverB = null!;
public int PortA => _serverA.Port;
public int PortB => _serverB.Port;
public async Task InitializeAsync()
{
var gwPortA = NatsServerProcess.AllocateFreePort();
var gwPortB = NatsServerProcess.AllocateFreePort();
var configA = $$"""
server_name: gw-a
gateway {
name: cluster-a
listen: 127.0.0.1:{{gwPortA}}
gateways: [
{ name: cluster-b, url: nats://127.0.0.1:{{gwPortB}} }
]
}
""";
var configB = $$"""
server_name: gw-b
gateway {
name: cluster-b
listen: 127.0.0.1:{{gwPortB}}
gateways: [
{ name: cluster-a, url: nats://127.0.0.1:{{gwPortA}} }
]
}
""";
_serverA = NatsServerProcess.WithConfig(configA, enableMonitoring: true);
_serverB = NatsServerProcess.WithConfig(configB, enableMonitoring: true);
await Task.WhenAll(_serverA.StartAsync(), _serverB.StartAsync());
// Poll until both gateways report a connected outbound gateway
await WaitForGatewayConnectionAsync();
}
private async Task WaitForGatewayConnectionAsync()
{
using var http = new HttpClient();
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(200));
var monitorPorts = new[] { _serverA.MonitorPort!.Value, _serverB.MonitorPort!.Value };
while (await timer.WaitForNextTickAsync(timeout.Token).ConfigureAwait(false))
{
var allConnected = true;
foreach (var monitorPort in monitorPorts)
{
try
{
var gatewayz = await http.GetFromJsonAsync<Gatewayz>(
$"http://127.0.0.1:{monitorPort}/gatewayz",
timeout.Token);
if (gatewayz?.NumGateways < 1)
{
allConnected = false;
break;
}
}
catch (HttpRequestException)
{
// monitor not yet ready — retry on next tick
allConnected = false;
break;
}
}
if (allConnected)
return;
}
throw new TimeoutException("Gateways did not connect to each other within 30s.");
}
public async Task DisposeAsync()
{
await Task.WhenAll(
_serverA.DisposeAsync().AsTask(),
_serverB.DisposeAsync().AsTask());
}
public NatsConnection CreateClientA()
=> new(new NatsOpts { Url = $"nats://127.0.0.1:{PortA}" });
public NatsConnection CreateClientB()
=> new(new NatsOpts { Url = $"nats://127.0.0.1:{PortB}" });
private sealed class Gatewayz
{
[System.Text.Json.Serialization.JsonPropertyName("num_gateways")]
public int NumGateways { get; init; }
}
}
[CollectionDefinition("E2E-Gateway")]
public class GatewayCollection : ICollectionFixture<GatewayFixture>;

71
tests/NATS.E2E.Tests/LeafNodeTests.cs Normal file
View File

@@ -0,0 +1,71 @@
using NATS.Client.Core;
using NATS.E2E.Tests.Infrastructure;
namespace NATS.E2E.Tests;
[Collection("E2E-LeafNode")]
public class LeafNodeTests(LeafNodeFixture fixture)
{
[Fact]
public async Task LeafNode_HubToLeaf_MessageDelivered()
{
await using var pub = fixture.CreateHubClient();
await using var sub = fixture.CreateLeafClient();
await pub.ConnectAsync();
await sub.ConnectAsync();
await using var subscription = await sub.SubscribeCoreAsync<string>("e2e.leaf.h2l");
// Ping both sides: subscriber's ping flushes the SUB to the leaf server,
// publisher's ping ensures the hub has received the propagated interest.
await sub.PingAsync();
await pub.PingAsync();
await pub.PublishAsync("e2e.leaf.h2l", "hub-to-leaf");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var msg = await subscription.Msgs.ReadAsync(cts.Token);
msg.Data.ShouldBe("hub-to-leaf");
}
[Fact]
public async Task LeafNode_LeafToHub_MessageDelivered()
{
await using var pub = fixture.CreateLeafClient();
await using var sub = fixture.CreateHubClient();
await pub.ConnectAsync();
await sub.ConnectAsync();
await using var subscription = await sub.SubscribeCoreAsync<string>("e2e.leaf.l2h");
// Ping both sides: subscriber's ping flushes the SUB to the hub server,
// publisher's ping ensures the leaf has received the propagated interest.
await sub.PingAsync();
await pub.PingAsync();
await pub.PublishAsync("e2e.leaf.l2h", "leaf-to-hub");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var msg = await subscription.Msgs.ReadAsync(cts.Token);
msg.Data.ShouldBe("leaf-to-hub");
}
[Fact]
public async Task LeafNode_OnlySubscribedSubjectsPropagate()
{
await using var pub = fixture.CreateHubClient();
await using var sub = fixture.CreateLeafClient();
await pub.ConnectAsync();
await sub.ConnectAsync();
await using var subscription = await sub.SubscribeCoreAsync<string>("e2e.leaf.specific");
// Ping both sides to ensure subscription interest has propagated before publishing.
await sub.PingAsync();
await pub.PingAsync();
await pub.PublishAsync("e2e.leaf.other", "wrong-subject");
await pub.PublishAsync("e2e.leaf.specific", "right-subject");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var msg = await subscription.Msgs.ReadAsync(cts.Token);
msg.Data.ShouldBe("right-subject");
}
}