GRPC异步响应流C#

Question

如何从处理程序外部为RPC生成流响应值？（特别是，从I观察者）我目前正在执行以下操作，但这会产生跨线程问题，因为AnRx观察者在RPC处理程序之间共享...

public override Task GetTicker(RequestProto request, ServerCallContext context)
{
    var subscription = AnRxObservable.Subscribe(value =>
    {
        responseStream.WriteAsync(new ResponseProto
        {
            Value = value
        });
    });

    // Wait for the RPC to be canceled (my extension method
    // that returns a task that completes when the CancellationToken
    // is cancelled)
    await context.CancellationToken.WhenCancelled();

    // Dispose of the buffered stream
    bufferedStream.Dispose();

    // Dispose subscriber (tells rx that we aren't subscribed anymore)
    subscription.Dispose();

    return Task.FromResult(1);
}

这段代码感觉不对劲...但是我看不到从RPC处理程序之外创建的共享源流式传输RPC响应的任何其他方式。

Answer 1

一般来说，当您试图从推模型（IObservable）转换为拉模型（枚举要写入和写入的响应）时，您需要为消息提供一个中间缓冲区，例如blockingQueue。然后，处理程序主体可以是一个异步循环，尝试获取队列的下一条消息（最好以异步方式），并将其写入responseStream。

此外，请注意，gRPC API只允许您在任何给定时间有1个飞行中响应，而您的代码片段并不尊重这一点。因此，在开始另一次写入之前，需要等待WriteAsync（）（这是需要中间队列的另一个原因）。

此链接可能有助于解释推与拉的范例：何时使用IEnumerable与IObservable？