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

using NATS.Server.JetStream;
using NATS.Server.JetStream.Cluster;
using NATS.Server.JetStream.Models;

namespace NATS.Server.Tests;

public class JetStreamMetaGroupTests
{
    [Fact]
    public async Task Stream_create_requires_meta_group_commit()
    {
        await using var fixture = await JetStreamClusterFixture.StartAsync(nodes: 3);

        var result = await fixture.CreateStreamAsync("ORDERS", replicas: 3);
        result.Error.ShouldBeNull();

        var meta = await fixture.GetMetaStateAsync();
        meta.Streams.ShouldContain("ORDERS");
    }
}

internal sealed class JetStreamClusterFixture : IAsyncDisposable
{
    private readonly JetStreamMetaGroup _metaGroup;
    private readonly StreamManager _streamManager;

    private JetStreamClusterFixture(JetStreamMetaGroup metaGroup, StreamManager streamManager)
    {
        _metaGroup = metaGroup;
        _streamManager = streamManager;
    }

    public static Task<JetStreamClusterFixture> StartAsync(int nodes)
    {
        var meta = new JetStreamMetaGroup(nodes);
        var streamManager = new StreamManager(meta);
        return Task.FromResult(new JetStreamClusterFixture(meta, streamManager));
    }

    public Task<NATS.Server.JetStream.Api.JetStreamApiResponse> CreateStreamAsync(string name, int replicas)
    {
        var response = _streamManager.CreateOrUpdate(new StreamConfig
        {
            Name = name,
            Subjects = [name.ToLowerInvariant() + ".*"],
            Replicas = replicas,
        });
        return Task.FromResult(response);
    }

    public Task<MetaGroupState> GetMetaStateAsync() => Task.FromResult(_metaGroup.GetState());

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