natsdotnet/tests/NATS.Server.TestUtilities/LeafFixture.cs

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

namespace NATS.Server.TestUtilities;

/// <summary>
/// Shared fixture for leaf node tests that creates a hub and a spoke server
/// connected via leaf node protocol.
/// </summary>
public 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,
            timeoutMs: 5000,
            intervalMs: 50);

        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}'.",
            timeoutMs: 5000,
            intervalMs: 50);
    }

    public async Task WaitForRemoteInterestOnSpokeAsync(string subject)
    {
        await PollHelper.WaitOrThrowAsync(
            () => Spoke.HasRemoteInterest(subject),
            $"Timed out waiting for remote interest on spoke for '{subject}'.",
            timeoutMs: 5000,
            intervalMs: 50);
    }

    public async ValueTask DisposeAsync()
    {
        await _spokeCts.CancelAsync();
        await _hubCts.CancelAsync();
        Spoke.Dispose();
        Hub.Dispose();
        _spokeCts.Dispose();
        _hubCts.Dispose();
    }
}