How to implement a code first gRPC server stream without loop and call delay in .Net5

Viewed 351

In the gRPC for WCF developers repo there is a great example how to implement a gRPC server stream using the contract first approach. The great thing on in is that it does not need a loop ar a queue and triggers the gRPC call without delay once the event is raised.

The code I am referencing is:


public override async Task Subscribe(IAsyncStreamReader requestStream, IServerStreamWriter responseStream, ServerCallContext context)
{
    using var subscriber = _subscriberFactory.GetSubscriber();

    subscriber.Update += async (sender, args) =>
        await WriteUpdateAsync(responseStream, args.Symbol, args.Price);

    var actionsTask = HandleActions(requestStream, subscriber, context.CancellationToken);
    
    _logger.LogInformation("Subscription started.");
    await AwaitCancellation(context.CancellationToken);

    try { await actionsTask; } catch { /* Ignored */ }

    _logger.LogInformation("Subscription finished.");
}

Now I am trying (but struggle) to achive the same behaviour (no loop, immediate call without delay) but using the code first approach with protobuf-net. The protobuf-net getting started sample (somewhere at the beginning of the last third) shows how to create it with a loop.


public class MyTimeService : ITimeService
{
    public IAsyncEnumerable SubscribeAsync(CallContext context = default)
        => SubscribeAsync(context.CancellationToken);

    private async IAsyncEnumerable SubscribeAsync([EnumeratorCancellation] CancellationToken cancel)
    {
        while (!cancel.IsCancellationRequested)
        {
            await Task.Delay(TimeSpan.FromSeconds(10), cancel);
            yield return new TimeResult { Time = DateTime.UtcNow };
        }
    }
}

Has anyone ever managed to implement a code first stream without loop and immediate calls?


All my current approaches failed because they would need something in the interface method signature (e.g. the ServerResponseStream) which I could not provide in the client...


1 Answers

I think what you're looking for is this:

var buffer = Channel.CreateUnbounded<StockTickerUpdate>();

from the link for showing how to do the same implementation from your link, but with code-first gRPC.

Related