同时使用异步枚举的编写和阅读。

Question

我有一个客户服务器方案，其中包含GRPC双工流完全实现的管道。管道由客户端启动。在每个步骤（双工呼叫）中，客户端从iAsyncenumerable中读取项目，并在频道上写下它们。这些项目是由服务器处理的，其中一些是异步发送在频道上的。然后，客户以产量返回读取项目。我有4种这样链接的方法。

我使用的模式是：

public async IAsyncEnumerable<Result> Results(IAsyncEnumerable<Input> inputs)
{
    var duplex = Server.GetResults();

    await foreach (var input in inputs)
    {
        await duplex.RequestStream.WriteAsync(input);
    }

    await duplex.RequestStream.CompleteAsync();

    await foreach (var result in duplex.ResponseStream.ToAsyncEnumerable())
    {
        yield return result;
    }
}

此模式的问题是服务器在第二个foreach之前（从频道读取）启动数据。如果有很多数据，则频道缓冲区在阅读之前就会充满（大概是） GRPC抛出了启动和取消操作的例外。

如何重构，而不会破坏iAsyncencenumerables和yourt的使用情况？如果这是不可能的，那么我还能如何获得“首先写，阅读下一个”方案？

只是为了澄清，我无法使用以下模式：

while(await stream.MoveNext()) 
{
   // send
   // yield
}

当我调用等待Duplex.ResponseStream.movenext（）客户端等待服务器等待服务器返回数据时。但这是无法预测的，服务器正在返回哪些数据。这将导致僵局。

我需要某种isDataavailableOnChannel，但是ryncduplexstreamingcall上没有这种方法

Answer 1

您可以将数据发送到另一个任务，然后一旦完成收益率 Ing，就等待它。

鉴于我假设您有一个普通的ThreadPool调度程序，因此应交错每个功能的等待。

public async IAsyncEnumerable<Result> Results(IAsyncEnumerable<Input> inputs)
{
    var duplex = Server.GetResults();

    var sendingTask = Task.Run(async () =>
    {
        await foreach (var input in inputs)
        {
            await duplex.RequestStream.WriteAsync(input);
        }

        await duplex.RequestStream.CompleteAsync();
    });

    await foreach (var result in duplex.ResponseStream.ToAsyncEnumerable())
    {
        yield return result;
    }

    await sendingTask;
}

如果将匿名lambda放置在适当的功能中，可能会更容易管理。