mediatr strambehavior

MediatrStreamBehavior.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MediatR;
using MediatR.Pipeline;
using Microsoft.Extensions.DependencyInjection;
using Xunit;
using Xunit.Abstractions;

namespace MediatR.UnitTests;

public class StreamTests
{
    private readonly ITestOutputHelper _helper;

    public record GetDataCommand(long Id): IStreamRequest<string>;

    public class GetDataValidator : IRequestPreProcessor<GetDataCommand>
    {
        public Task Process(GetDataCommand request, CancellationToken cancellationToken)
            => throw new NotImplementedException("PreProcessor");
    }

    public class GetDataBehavior : IStreamPipelineBehavior<GetDataCommand, string>
    {
        public IAsyncEnumerable<string> Handle(GetDataCommand request, StreamHandlerDelegate<string> next, CancellationToken cancellationToken) 
            => throw new NotImplementedException("Behavior");
    }

    public class GetDataHandler : IStreamRequestHandler<GetDataCommand, string>
    {
        public async IAsyncEnumerable<string> Handle(GetDataCommand request, CancellationToken cancellationToken)
        {
            foreach (var data in Enumerable.Range(0, 10))
            {
                yield return data.ToString();
            }
        }
    }

    public StreamTests(ITestOutputHelper helper) => _helper = helper;
    
    [Fact]
    public async Task StreamPreProcessor()
    {
        var services = new ServiceCollection();
        services.AddMediatR(c => c
            .RegisterServicesFromAssembly(GetType().Assembly)
            .AddBehavior<IStreamPipelineBehavior<GetDataCommand, string>, GetDataBehavior>());
        services.AddTransient<IRequestPreProcessor<GetDataCommand>, GetDataValidator>();
        var provider = services.BuildServiceProvider();
        var mediator = provider.GetRequiredService<IMediator>();
        
        _helper.WriteLine("before CreateStream");
        var stream = mediator.CreateStream(new GetDataCommand(1));
        _helper.WriteLine("after CreateStream");
        await foreach (var buffer in stream) // here the behavior is called
        {
            _helper.WriteLine("iteration");
            _helper.WriteLine(buffer.Length.ToString());
        }
    }
}