GRPC非同期応答ストリームC＃

Question

ハンドラの外側からRPCのストリーミング応答値を生成するにはどうすればよいですか？ （具体的には、IObservableから）右に感じることはありません...

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); 
} 

このコードを私は現在、以下のことをやっているが、AnRxObservableはRPCハンドラ間で共有されているので、これは、クロススレッドの問題を作成しています... RPCハンドラの外部で作成された共有ソースからRPC応答をストリーミングする方法は他にありません。

Answer 1

一般に、プッシュモデル（IObservable）からプルモデル（プル型モデルへの変換を列挙する）に変換しようとすると、メッセージの中間バッファが必要です。 blockingQueue。ハンドラ本体は、キューの次のメッセージをフェッチしようとする非同期ループ（できれば非同期の方法）であり、それをresponseStreamに書き込むことができます。

また、gRPC APIでは、一度に1つの機内応答しか得られないことに注意してください。あなたのスニペットはこれを尊重しません。したがって、別の書き込みを開始する前にWriteAsync（）を待つ必要があります（これが中間キューを必要とするもう1つの理由です）。あなたは「クロススレッドの問題を引き起こして」とはどういう意味ですかWhen to use IEnumerable vs IObservable?