我们如何使用“yield return”从 MediatR 请求处理程序返回 IAsyncEnumerable？

Question

我有一个实现 Mediatr.IRequest 的查询类，如下所示：

public class ExportDataQuery : IRequest<IAsyncEnumerable<byte[]>> {}

查询处理程序的实现如下：

public class ExportDataQueryHandler : IRequestHandler<ExportDataQuery, IAsyncEnumerable<byte[]>>
{
  public async IAsyncEnumerable<byte[]> Handle(ExportDataQuery query, CancellationToken cancellationToken)
  {
     for (int page = 1; page <= pageCount; ++page)
     {
        // Get paginated data asynchronously.
        var data = await _dbUtils.GetDataAsync(page, pageSize);
        yield return data;
     }
  }
}

但是我在编译上述代码时遇到以下构建错误：

Error CS0738 'ExportDataQueryHandler' does not implement interface member 'IRequestHandler<ExportDataQuery, IAsyncEnumerable<byte[]>>.Handle(ExportDataQuery, CancellationToken)'. 'ExportDataQueryHandler.Handle(ExportDataQuery, CancellationToken)' cannot implement 'IRequestHandler<ExportDataQuery, IAsyncEnumerable<byte[]>>.Handle(ExportDataQuery, CancellationToken)' because it does not have the matching return type of 'Task<IAsyncEnumerable<byte[]>>'.

当我将 Handle 的返回类型更改为

Task<IAsyncEnumerable<byte[]>>

时，出现以下错误：

Error CS1624 The body of 'ExportDataQueryHandler.Handle(ExportDataQuery, CancellationToken)' cannot be an iterator block because 'Task<IAsyncEnumerable<byte[]>>' is not an iterator interface type.

有没有办法在上面的请求处理程序中使用

yield return

一次返回一个页面数据？

Answer 1

你能做到吗？修改您的调用以返回集合 AsAsyncEnumerable，然后通过继承

RequestHandler<T>

使用同步请求类型，这样它就不会将结果包装在

Task<T>

中

public class ExportDataQueryHandler : RequestHandler<ExportDataQuery, IAsyncEnumerable<byte[]>>
{
    protected override IAsyncEnumerable<byte[]> Handle(ExportDataQuery query)
    {            
        return _dbUtils.GetDataAsync(page, pageSize).AsAsyncEnumerable();           
    }
}

Answer 2

您可以使用

IStreamRequest

和

IStreamRequestHandler

public class ExportDataQuery : IStreamRequest<byte[]> {}

public class ExportDataQueryHandler : IStreamRequestHandler<ExportDataQuery, byte[]>
{
  public async IAsyncEnumerable<byte[]> Handle(ExportDataQuery query, CancellationToken cancellationToken)
  {
     for (int page = 1; page <= pageCount; ++page)
     {
        // Get paginated data asynchronously.
        var data = await _dbUtils.GetDataAsync(page, pageSize);
        yield return data;
     }
  }
}