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 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 GetJszAsync() { var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/jsz"); response.StatusCode.ShouldBe(HttpStatusCode.OK); var jsz = await response.Content.ReadFromJsonAsync(); 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."); } }