natsdotnet/tests/NATS.Server.Tests/LeafNodes/LeafBasicTests.cs

using Microsoft.Extensions.Logging.Abstractions;
using NATS.Client.Core;
using NATS.Server.Configuration;

namespace NATS.Server.Tests.LeafNodes;

/// <summary>
/// Basic leaf node hub-spoke connectivity tests.
/// Reference: golang/nats-server/server/leafnode_test.go — TestLeafNodeRemoteIsHub
/// Verifies that subscriptions propagate between hub and leaf (spoke) servers
/// and that messages are forwarded in both directions.
/// </summary>
public class LeafBasicTests
{
    [Fact]
    public async Task Leaf_node_forwards_subscriptions_to_hub()
    {
        // Arrange: start hub with a leaf node listener, then start a spoke that connects to hub
        await using var fixture = await LeafBasicFixture.StartAsync();

        await using var leafConn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Spoke.Port}",
        });
        await leafConn.ConnectAsync();

        await using var hubConn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Hub.Port}",
        });
        await hubConn.ConnectAsync();

        // Subscribe on the leaf (spoke) side
        await using var sub = await leafConn.SubscribeCoreAsync<string>("leaf.test");
        await leafConn.PingAsync();

        // Wait for the subscription interest to propagate to the hub
        await fixture.WaitForRemoteInterestOnHubAsync("leaf.test");

        // Publish on the hub side
        await hubConn.PublishAsync("leaf.test", "from-hub");

        // Assert: message arrives on the leaf
        using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var msg = await sub.Msgs.ReadAsync(receiveTimeout.Token);
        msg.Data.ShouldBe("from-hub");
    }

    [Fact]
    public async Task Hub_forwards_subscriptions_to_leaf()
    {
        // Arrange: start hub with a leaf node listener, then start a spoke that connects to hub
        await using var fixture = await LeafBasicFixture.StartAsync();

        await using var hubConn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Hub.Port}",
        });
        await hubConn.ConnectAsync();

        await using var leafConn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Spoke.Port}",
        });
        await leafConn.ConnectAsync();

        // Subscribe on the hub side
        await using var sub = await hubConn.SubscribeCoreAsync<string>("hub.test");
        await hubConn.PingAsync();

        // Wait for the subscription interest to propagate to the spoke
        await fixture.WaitForRemoteInterestOnSpokeAsync("hub.test");

        // Publish on the leaf (spoke) side
        await leafConn.PublishAsync("hub.test", "from-leaf");

        // Assert: message arrives on the hub
        using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var msg = await sub.Msgs.ReadAsync(receiveTimeout.Token);
        msg.Data.ShouldBe("from-leaf");
    }
}

internal sealed class LeafBasicFixture : IAsyncDisposable
{
    private readonly CancellationTokenSource _hubCts;
    private readonly CancellationTokenSource _spokeCts;

    private LeafBasicFixture(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<LeafBasicFixture> 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();

        // Wait for the leaf node connection to be established on both sides
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested && (hub.Stats.Leafs == 0 || spoke.Stats.Leafs == 0))
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        return new LeafBasicFixture(hub, spoke, hubCts, spokeCts);
    }

    public async Task WaitForRemoteInterestOnHubAsync(string subject)
    {
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested)
        {
            if (Hub.HasRemoteInterest(subject))
                return;

            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);
        }

        throw new TimeoutException($"Timed out waiting for remote interest on hub for '{subject}'.");
    }

    public async Task WaitForRemoteInterestOnSpokeAsync(string subject)
    {
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested)
        {
            if (Spoke.HasRemoteInterest(subject))
                return;

            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);
        }

        throw new TimeoutException($"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();
    }
}