GRPC async response stream C#

Viewed 6128

How would I generate streaming response values for an RPC from outside the handler? (specifically, from a IObservable) I'm currently doing the following, but this is creating cross-thread issues because AnRxObservable is shared between RPC handlers...

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

This code doesn't feel right... but I can't see any other way of streaming RPC responses from a shared source created outside the RPC handler.

1 Answers
Related