[Fact]
public async Task Test()
{
    using var fixture = new ClusterFixture();
    var client = fixture.SiloHost.Services.GetRequiredService<IClusterClient>();
    var logger = fixture.SiloHost.Services.GetRequiredService<ILoggerFactory>().CreateLogger("Logger");
    var streamProvider = client.GetStreamProvider("SimpleMemoryStreamProvider");
    var stream = streamProvider.GetStream<int>("TestStream", Guid.Empty);

    foreach (var i in Enumerable.Range(6, 6))
    {
        logger.LogInformation("Sending {Value}...", i);
        await Task.Delay(TimeSpan.FromSeconds(i));
        await stream.OnNextAsync(i);
    }
}

[ImplicitStreamSubscription("TestStream")]
public class TestGrain : IGrainBase, IGrainWithGuidKey
{
    private readonly ILogger<TestGrain> _logger;
    private readonly IGrainContext _grainContext;

    public TestGrain(ILogger<TestGrain> logger, IGrainContext grainContext)
    {
        _logger = logger;
        _grainContext = grainContext;
    }

    public IGrainContext GrainContext => _grainContext;

    public async Task OnActivateAsync(CancellationToken token)
    {
        var streamProvider = this.GetStreamProvider("SimpleMemoryStreamProvider");
        var stream = streamProvider.GetStream<int>("TestStream", this.GetPrimaryKey());

        await stream.SubscribeAsync(
            (value, token) =>
            {
                _logger.LogInformation("Received {Value}", value);
                return Task.CompletedTask;
            },
            ex =>
            {
                _logger.LogError(ex, "A stream error has occurred.");
                return Task.CompletedTask;
            });
    }
}

public class ClusterFixture : IDisposable
{
    public ClusterFixture()
    {
        var builder = new TestClusterBuilder(1);

        builder.AddSiloBuilderConfigurator<TestSiloConfig>();

        Cluster = builder.Build();
        Cluster.Deploy();
        SiloHost = ((InProcessSiloHandle)Cluster.Primary).SiloHost;
    }

    public TestCluster Cluster { get; private set; }

    public IHost SiloHost { get; private set; }

    public void Dispose()
    {
        Cluster.StopAllSilos();
        Cluster.Dispose();
        GC.SuppressFinalize(this);
    }

    private class TestSiloConfig : ISiloConfigurator
    {
        public void Configure(ISiloBuilder hostBuilder)
        {
            hostBuilder
                .ConfigureLogging(logger => logger.AddDebug())
                .ConfigureServices(services =>
                {
                    services
                        .Configure<StreamCacheEvictionOptions>("SimpleMemoryStreamProvider", options =>
                        {
                            options.DataMaxAgeInCache = TimeSpan.FromSeconds(6);
                            options.DataMinTimeInCache = TimeSpan.FromSeconds(1);
                        });
                });

            hostBuilder
                .AddMemoryGrainStorageAsDefault()
                .AddMemoryGrainStorage(ProviderConstants.DEFAULT_PUBSUB_PROVIDER_NAME)
                .AddMemoryStreams<DefaultMemoryMessageBodySerializer>("SimpleMemoryStreamProvider", configure =>
                {
                    configure.ConfigurePullingAgent(ob => ob.Configure(options =>
                    {
                        options.StreamInactivityPeriod = TimeSpan.FromSeconds(6);
                    }));

                    configure.ConfigureStreamPubSub(Orleans.Streams.StreamPubSubType.ImplicitOnly);
                });
        }
    }
}