natsdotnet/tests/NATS.Server.Monitoring.Tests/JszMonitorTests.cs

using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using System.Text;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Configuration;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;

namespace NATS.Server.Monitoring.Tests;

public class JszMonitorTests
{
    [Fact]
    public async Task Jsz_reports_live_stream_and_consumer_counts()
    {
        await using var fixture = await JetStreamMonitoringFixture.StartWithStreamAndConsumerAsync();

        var jsz = await fixture.GetJszAsync();
        jsz.Streams.ShouldBeGreaterThan(0);
        jsz.Consumers.ShouldBeGreaterThan(0);
    }
}

internal sealed class JetStreamMonitoringFixture : IAsyncDisposable
{
    private readonly NatsServer _server;
    private readonly int _monitorPort;
    private readonly CancellationTokenSource _cts = new();
    private readonly HttpClient _http = new();

    private JetStreamMonitoringFixture(NatsServer server, int monitorPort)
    {
        _server = server;
        _monitorPort = monitorPort;
    }

    public static async Task<JetStreamMonitoringFixture> StartWithStreamAndConsumerAsync()
    {
        var natsPort = TestPortAllocator.GetFreePort();
        var monitorPort = TestPortAllocator.GetFreePort();
        var options = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = natsPort,
            MonitorHost = "127.0.0.1",
            MonitorPort = monitorPort,
            JetStream = new JetStreamOptions
            {
                StoreDir = Path.Combine(Path.GetTempPath(), "natsdotnet-jsz"),
                MaxMemoryStore = 1_024 * 1_024,
                MaxFileStore = 10 * 1_024 * 1_024,
            },
        };

        var server = new NatsServer(options, NullLoggerFactory.Instance);
        var fixture = new JetStreamMonitoringFixture(server, monitorPort);

        _ = server.StartAsync(fixture._cts.Token);
        await server.WaitForReadyAsync();
        await fixture.WaitForHealthAsync();

        var router = server.JetStreamApiRouter ?? throw new InvalidOperationException("JetStream API router unavailable.");
        _ = router.Route("$JS.API.STREAM.CREATE.ORDERS", Encoding.UTF8.GetBytes("{\"name\":\"ORDERS\",\"subjects\":[\"orders.*\"]}"));
        _ = router.Route("$JS.API.CONSUMER.CREATE.ORDERS.DUR", Encoding.UTF8.GetBytes("{\"durable_name\":\"DUR\",\"filter_subject\":\"orders.*\"}"));

        return fixture;
    }

    public async Task<JszResponse> GetJszAsync()
    {
        var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/jsz");
        response.StatusCode.ShouldBe(HttpStatusCode.OK);

        var jsz = await response.Content.ReadFromJsonAsync<JszResponse>();
        return jsz ?? throw new InvalidOperationException("Failed to deserialize /jsz.");
    }

    public async ValueTask DisposeAsync()
    {
        _http.Dispose();
        await _cts.CancelAsync();
        _server.Dispose();
    }

    private async Task WaitForHealthAsync()
    {
        for (int i = 0; i < 50; i++)
        {
            try
            {
                var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
                if (response.IsSuccessStatusCode)
                    return;
            }
            catch (HttpRequestException)
            {
                // server not ready
            }

            await Task.Delay(50);
        }

        throw new TimeoutException("Monitoring endpoint did not become healthy.");
    }

}