natsdotnet/tests/NATS.Server.Tests/JetStreamApiFixture.cs

using System.Text;
using NATS.Server.JetStream;
using NATS.Server.JetStream.Api;
using NATS.Server.JetStream.Models;
using NATS.Server.JetStream.Publish;

namespace NATS.Server.Tests;

internal sealed class JetStreamApiFixture : IAsyncDisposable
{
    private static readonly StreamManager SharedStreamManager = new();
    private static readonly JetStreamApiRouter SharedRouter = new(SharedStreamManager);

    private readonly StreamManager _streamManager;
    private readonly JetStreamApiRouter _router;
    private readonly JetStreamPublisher _publisher;

    private JetStreamApiFixture()
    {
        _streamManager = new StreamManager();
        _router = new JetStreamApiRouter(_streamManager);
        _publisher = new JetStreamPublisher(_streamManager);
    }

    public static Task<JetStreamApiResponse> RequestAsync(string subject, string payload)
    {
        return Task.FromResult(SharedRouter.Route(subject, Encoding.UTF8.GetBytes(payload)));
    }

    public static async Task<JetStreamApiFixture> StartWithStreamAsync(string streamName, string subject, int maxMsgs = 0)
    {
        var fixture = new JetStreamApiFixture();
        var payload = $"{{\"name\":\"{streamName}\",\"subjects\":[\"{subject}\"],\"max_msgs\":{maxMsgs}}}";
        _ = await fixture.RequestLocalAsync($"$JS.API.STREAM.CREATE.{streamName}", payload);
        return fixture;
    }

    public Task<PubAck> PublishAndGetAckAsync(string subject, string payload, string? msgId = null, bool expectError = false)
    {
        if (_publisher.TryCapture(subject, Encoding.UTF8.GetBytes(payload), msgId, out var ack))
            return Task.FromResult(ack);

        if (expectError)
            return Task.FromResult(new PubAck { ErrorCode = 404 });

        throw new InvalidOperationException($"No stream matched subject '{subject}'.");
    }

    public Task<JetStreamApiResponse> RequestLocalAsync(string subject, string payload)
    {
        return Task.FromResult(_router.Route(subject, Encoding.UTF8.GetBytes(payload)));
    }

    public Task<StreamState> GetStreamStateAsync(string streamName)
    {
        return _streamManager.GetStateAsync(streamName, default).AsTask();
    }

    public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}