C# .NET 8 — 使用 MediatR 流式传输请求和管道

概述：MediatR 是一个库，它自称为“.NET 中的简单调解器实现”。在本文中，我们将探讨 MediatR 中请求的异常处理过程，特别关注 MediatR 实体只有一个接收方的场景。有关MediatR的进一步见解，请随时浏览我就此主题撰写的其他文章。流请求普通请求可用于命令和查询操作。流请求专门设计用于使用 MediatR 检索一组流数据。为了使用 Mediator 处理流请求，需要使用特定的接口。IStreamRequest与标准请求非常相似，它使用单个接收器运行，并且来自 MediatR 的返回对象是一个集合。IAsyncEnumerable我不会在这里深入研究如何在 C# 中处理流;我将

MediatR 是一个库，它自称为“.NET 中的简单调解器实现”。

在本文中，我们将探讨 MediatR 中请求的异常处理过程，特别关注 MediatR 实体只有一个接收方的场景。

有关MediatR的进一步见解，请随时浏览我就此主题撰写的其他文章。

流请求

普通请求可用于命令和查询操作。流请求专门设计用于使用 MediatR 检索一组流数据。

为了使用 Mediator 处理流请求，需要使用特定的接口。IStreamRequest

与标准请求非常相似，它使用单个接收器运行，并且来自 MediatR 的返回对象是一个集合。IAsyncEnumerable

我不会在这里深入研究如何在 C# 中处理流;我将来可能会单独写一篇关于它的文章。

为了简短起见，提供流需要使用语句。AsyncEnumerable()yield

在 C# 中，关键字有助于对集合进行迭代，一次返回一个元素。此方法允许在接收数据时返回数据，使调用方能够立即处理数据。yield

示例实现

这是我准备的一个项目的链接：文末提供下载。

启动应用程序，将出现 Swagger 页面：

您可以找到 4 个不同的终结点：

SampleCommand：使用此终结点测试命令。此命令通过所有 MediatR 管道：日志记录 -> 验证 -> 授权。
SampleRequest：使用此终结点跳过所有管道。这之所以有效，是因为它使用普通的 IRequest 而不是自定义的 ICommand。
SampleEntity：此终结点可用于测试添加 Sample 实体终结点的结果。
AddSampleEntity：此终结点使用 ITransactionCommand 接口，是工作单元管道行为的示例。在此项目中，它还实现了 IRepository 模式的示例，该模式使用 EF 和 InMemory 数据库。
SampleStreamEntity：利用纯 IStreamRequest，此终结点从数据库中检索数据。
SampleStreamEntityWithPipeFilter：与上一个终结点类似，此终结点也使用普通 IStreamRequest 从数据库中提取数据。但是，它包含应用于请求的管道行为。

出于本文的目的，我们将深入研究最后两个终结点的复杂性：SampleStreamEntity 和 SampleStreamEntityWithPipeFilter。

要探索其他端点的功能，请参阅我的另一篇专注于 MediatR 的文章。

代码解释

在本文的以下部分中，我们将了解如何制作 2 个终结点 SampleStreamEntity 和 SampleStreamEntityWithPipeFilter。

检索流

首先，让我们看看如何在 InMemory 配置的上下文中使用 EF Core 检索流。

public async IAsyncEnumerable<T> GetStream(CancellationToken cancellationToken = default)  
    {  
        var datas = _dbSet.AsAsyncEnumerable<T>();  
        await foreach (var data in datas)  
        {  
            if (cancellationToken.IsCancellationRequested)  
                yield break;  
  
            yield return data;  
        }  
    }

示例流实体

现在，让我们深入研究 SampleStreamEntity 终结点的实现，该终结点利用 MediatR 处理请求。

app.MapGet(  
        "/SampleStreamEntity",  
        async (IMediator mediator, CancellationToken cancellationToken) =>  
        {  
            return mediator.CreateStream(new SampleStreamEntityQuery(), cancellationToken);  
        }  
    )  
    .WithName("SampleStreamEntity")  
    .Produces<IAsyncEnumerable<SampleStreamEntityQueryResult>>()  
    .WithOpenApi();

该方法向调用方返回的集合。CreateStreamIAsyncEnumerable<SampleStreamEntityQueryResult>

该接口是 MediatR 提供的标准接口，专门用于处理流式处理操作。它返回一个名为的对象。IStreamRequestSampleStreamEntityQueryResult

public class SampleStreamEntityQuery : IStreamRequest<SampleStreamEntityQueryResult> { }  
  
public record SampleStreamEntityQueryResult  
{  
    public Guid Id { get; set; }  
  
    public DateTime EventTime { get; set; }  
  
    public string Description { get; set; }  
}

QueryHandler 实现利用存储库以及从数据库获取流数据的方法。GetStream

public class SampleStreamQueryHandler  
    : IStreamRequestHandler<SampleStreamEntityQuery, SampleStreamEntityQueryResult>  
{  
    private readonly IRepository<SampleEntity> _repository;  
  
    public SampleStreamQueryHandler(IRepository<SampleEntity> repository)  
    {  
        _repository = repository;  
    }  
  
    public async IAsyncEnumerable<SampleStreamEntityQueryResult> Handle(  
 SampleStreamEntityQuery request,  
 CancellationToken cancellationToken  
 )  
    {  
        await foreach (var entity in _repository.GetStream(cancellationToken))  
        {  
            if (cancellationToken.IsCancellationRequested)  
                break;  
  
            yield return new SampleStreamEntityQueryResult()  
            {  
                Id = entity.Id,  
                Description = entity.Description,  
                EventTime = entity.RegistrationTime  
            };  
        }  
    }  
}

使用管道过滤器的示例流图元

第二个终结点 SampleStreamEntityWithPipeFilter 与前一个终结点类似，但包括用于筛选的管道行为。

以下代码片段表示终结点定义：

   app.MapGet(  
           "/SampleStreamEntityWithPipeFilter",  
           async (IMediator mediator, CancellationToken cancellationToken) =>  
           {  
               return mediator.CreateStream(  
                   new SampleStreamEntityWithPipeFilterQuery(),  
                   cancellationToken  
               );  
           }  
       )  
       .WithName("SampleStreamEntityWithPipeFilter")  
       .Produces<IAsyncEnumerable<SampleStreamEntityWithPipeFilterQueryResult>>()  
       .WithOpenApi();

下面是 SampleStreamEntityWithPipeFilterQuery 的定义。

public class SampleStreamEntityWithPipeFilterQuery  
    : IStreamRequest<SampleStreamEntityWithPipeFilterQueryResult> { }  
  
public record SampleStreamEntityWithPipeFilterQueryResult  
{  
    public Guid Id { get; set; }  
  
    public DateTime EventTime { get; set; }  
  
    public string Description { get; set; }  
}

这是它的实现：

public class SampleStreamQueryWithPipeFilterHandler  
    : IStreamRequestHandler<  
        SampleStreamEntityWithPipeFilterQuery,  
        SampleStreamEntityWithPipeFilterQueryResult  
    >  
{  
    private readonly IRepository<SampleEntity> _repository;  
  
    public SampleStreamQueryWithPipeFilterHandler(IRepository<SampleEntity> repository)  
    {  
        _repository = repository;  
    }  
  
    public async IAsyncEnumerable<SampleStreamEntityWithPipeFilterQueryResult> Handle(  
 SampleStreamEntityWithPipeFilterQuery request,  
 CancellationToken cancellationToken  
 )  
    {  
        await foreach (var entity in _repository.GetStream(cancellationToken))  
        {  
            if (cancellationToken.IsCancellationRequested)  
                break;  
  
            yield return new SampleStreamEntityWithPipeFilterQueryResult()  
            {  
                Id = entity.Id,  
                Description = entity.Description,  
                EventTime = entity.RegistrationTime  
            };  
        }  
    }  
}

流管道示例

与我之前的文章一样，我的目标是为流请求设计管道行为。

为了实现这一点，我将利用该接口。IStreamPipelineBehavior

此接口允许创建可以截获和修改流式处理请求的自定义行为，从而为应用程序的数据处理管道提供灵活性和可扩展性。

通用流管道示例

将为每个 .这可确保定义的行为一致地应用于所有流式处理请求，从而保持一致性并促进对数据处理管道的集中控制。IStreamRequest

public class GenericStreamLoggingBehavior<TRequest, TResponse>  
    : IStreamPipelineBehavior<TRequest, TResponse>  
{  
    private readonly ILogger<GenericStreamLoggingBehavior<TRequest, TResponse>> _logger;  
  
    public GenericStreamLoggingBehavior(  
 ILogger<GenericStreamLoggingBehavior<TRequest, TResponse>> logger  
 )  
    {  
        _logger = logger;  
    }  
  
    public async IAsyncEnumerable<TResponse> Handle(  
 TRequest request,  
 StreamHandlerDelegate<TResponse> next,  
 [EnumeratorCancellation] CancellationToken cancellationToken  
 )  
    {  
        _logger.LogInformation("Stream Request Start");  
        await foreach (  
            var response in next().WithCancellation(cancellationToken).ConfigureAwait(false)  
        )  
        {  
            _logger.LogInformation(  
                "Processing message {json}",  
                System.Text.Json.JsonSerializer.Serialize(response)  
            );  
  
            yield return response;  
        }  
        _logger.LogInformation("Stream Request End");  
    }  
}

特定流管道的示例

在上一篇文章中，我们创建了一个继承自默认原始请求接口的自定义接口。这使我们能够创建针对特定请求系列量身定制的管道。

但是，在本文中，我将采用不同的方法。在这里，我专门为单个 MediatR 流创建了一个管道。

为了实现这一点，我使用 where 子句直接在实现上指定类类型。IStreamPipelineBehavior

此方法提供了一种更集中、更简化的方法来实现管道，以应用程序体系结构中的特定流为目标。

public class SampleFilterStreamBehavior<TRequest, TResponse>  
    : IStreamPipelineBehavior<TRequest, TResponse>  
    where TRequest : SampleStreamEntityWithPipeFilterQuery  
    where TResponse : SampleStreamEntityWithPipeFilterQueryResult  
{  
    private readonly ILogger<SampleFilterStreamBehavior<TRequest, TResponse>> _logger;  
    private readonly IAuthService _authService;  
  
    public SampleFilterStreamBehavior(  
 IAuthService authService,  
 ILogger<SampleFilterStreamBehavior<TRequest, TResponse>> logger  
 )  
    {  
        _logger = logger;  
        _authService = authService;  
    }  
  
    public async IAsyncEnumerable<TResponse> Handle(  
 TRequest request,  
 StreamHandlerDelegate<TResponse> next,  
 [EnumeratorCancellation] CancellationToken cancellationToken  
 )  
    {  
        await foreach (  
            var response in next().WithCancellation(cancellationToken).ConfigureAwait(false)  
        )  
        {  
            var isAllowed = _authService.OperationAlowed();  
  
            if (isAllowed.IsSuccess)  
            {  
                yield return response;  
            }  
            else  
            {  
                _logger.LogWarning(  
                    "User is not allowed to get this data, entity {json} has not be returned.",  
                    System.Text.Json.JsonSerializer.Serialize(response)  
                );  
            }  
        }  
    }  
}

在域项目中，您将找到一个注册所有依赖项的扩展方法。

  services.AddTransient(  
      typeof(IStreamPipelineBehavior<,>),  
      typeof(GenericStreamLoggingBehavior<,>)  
  );  
  services.AddTransient(  
      typeof(IStreamPipelineBehavior<,>),  
      typeof(SampleFilterStreamBehavior<,>)  
  );

通过分享这些见解，我希望创建一个有价值的资源，帮助提高 .NET 开发的能力。

源代码获取：公众号回复消息【code：15528】

本文PDF版本下载