Streaming Prices over gRPC

Redis pub/sub moves FeedR's price ticks between services, but a browser or desktop client can't subscribe to Redis — nor should it ever be allowed to. The last hop, from service to client, needs its own transport, and the open-source DevMentors sample this series dissects makes an interesting choice for it: not SignalR, not WebSockets, but gRPC server streaming. One subscription request, one long-lived HTTP/2 response that yields a message per tick for as long as the client stays connected.

This part reads the contract, the server, and the client — and then stops at the bridge between the price generator's event and the gRPC response stream, because that bridge contains the single piece of code in this repository I most want to rewrite. Studying why it's wrong is worth more than a dozen examples that are merely correct.

The contract: four messages and a stream

gRPC starts with a .proto file, and FeedR's is refreshingly small (pricing.proto, shared verbatim between server and client projects):

service PricingFeed {
  rpc GetSymbols(GetSymbolsRequest) returns (GetSymbolsResponse);
  rpc SubscribePricing(PricingRequest) returns (stream PricingResponse);
}

message PricingResponse {
  string symbol = 1;
  int32 value = 2;
  int64 timestamp = 3;
}

That one keyword — stream on the return of SubscribePricing — is the entire push architecture at the client edge. The client sends one PricingRequest (optionally naming a symbol to filter on) and then reads for as long as it likes; the server writes whenever the generator ticks.

The detail I always point at: int32 value. The C# domain model carries prices as decimal, but Protobuf has no decimal type — floats would introduce exactly the rounding behavior you never want near money. FeedR does the classic thing: fixed-point integers. The server multiplies by 100 on the way out; the client divides on the way in (current.Value / 100M). It works, and it's how a lot of real market-data protocols behave — but notice that the scale factor 100 lives as an unwritten agreement in two codebases. Two decimal places is also suspiciously coarse for FX, where prices are conventionally quoted to 4–5 decimals (“pips”); a real feed would either carry units+nanos (the Google money.proto approach) or put the scale in the message so it can vary by instrument. Convention-based contracts again, one layer down.

One port for REST, one for gRPC

gRPC requires HTTP/2 end-to-end, and plain-text HTTP/2 (h2c) can't be negotiated on the same unencrypted endpoint as HTTP/1.1 — there's no TLS/ALPN step to pick a protocol in. FeedR's quotes service answers with dual Kestrel endpoints, straight from appsettings.json:

"Kestrel": {
  "Endpoints": {
    "Http":  { "Url": "http://localhost:5040", "Protocols": "Http1AndHttp2" },
    "gRPC":  { "Url": "http://localhost:5041", "Protocols": "Http2" }
  }
}

Port 5040 serves the minimal API from part 3; port 5041 is HTTP/2-only for gRPC. It's a pragmatic local-dev answer to a question that disappears in production behind TLS (ALPN negotiates the protocol per-connection on one port). Worth noticing: the gateway from part 2 proxies only the HTTP port — gRPC clients dial the quotes service directly, so the sample's “one front door” quietly has a side entrance.

The server: bridging an event to a stream

Here's where it gets instructive. The price generator exposes a plain .NET event, PricingUpdated. The gRPC call needs to turn that event into writes on a response stream. FeedR's bridge, from PricingGrpcService, quoted at length because the flaw is the lesson:

public override async Task SubscribePricing(PricingRequest request,
    IServerStreamWriter<PricingResponse> responseStream, ServerCallContext context)
{
    _pricingGenerator.PricingUpdated += OnPricingUpdated;

    while (!context.CancellationToken.IsCancellationRequested)
    {
        if (!_currencyPairs.TryTake(out var currencyPair))
        {
            continue;
        }

        if (!string.IsNullOrWhiteSpace(request.Symbol) && request.Symbol != currencyPair.Symbol)
        {
            continue;
        }

        await responseStream.WriteAsync(new PricingResponse { /* ... */ });
    }

    _pricingGenerator.PricingUpdated -= OnPricingUpdated;

    void OnPricingUpdated(object? sender, CurrencyPair currencyPair)
        => _currencyPairs.TryAdd(currencyPair);
}

The structure is genuinely nice: subscribe to the event on entry, unsubscribe on exit, keep the method alive until the client's CancellationToken fires, and use a local function as the handler so it can close over the call's own BlockingCollection. Server-side filtering (only stream the symbol the client asked for) is exactly where filtering belongs — don't ship messages a client will discard.

But look at the loop. TryTake is non-blocking: when no tick is available — which is 99.99% of the time, since ticks arrive every second and this loop runs millions of times a second — it returns false instantly and the loop continues. This is a hot spin. One connected client pins a thread-pool thread at full CPU doing nothing, for the lifetime of the subscription. Ten clients, ten cores of busy-waiting. It will pass every functional test you write and fall over in the first load test — my least favorite category of bug, because the code reads like it works.

There's a second, subtler issue: the filter discards non-matching ticks after taking them from the collection. With multiple subscribers each holding their own BlockingCollection that's fine, but each SubscribePricing call also attaches another handler to the same generator event — fan-out by event multicast, unbounded queue per client, no backpressure anywhere. A slow client's collection just grows.

The rewrite it's asking for

The modern .NET answer is the same primitive that part 3 celebrated: System.Threading.Channels, which gives you an awaitable take. The minimal fix preserves the whole design and deletes the spin:

var channel = Channel.CreateBounded<CurrencyPair>(
    new BoundedChannelOptions(64) { FullMode = BoundedChannelFullMode.DropOldest });

void OnPricingUpdated(object? s, CurrencyPair pair) => channel.Writer.TryWrite(pair);

_pricingGenerator.PricingUpdated += OnPricingUpdated;
try
{
    await foreach (var pair in channel.Reader.ReadAllAsync(context.CancellationToken))
    {
        if (!string.IsNullOrWhiteSpace(request.Symbol) && request.Symbol != pair.Symbol)
            continue;

        await responseStream.WriteAsync(ToResponse(pair));
    }
}
finally
{
    _pricingGenerator.PricingUpdated -= OnPricingUpdated;
}

ReadAllAsync parks the call — zero CPU — until a tick arrives or the client disconnects. The bounded channel with DropOldest encodes the correct policy for price data explicitly: a slow client gets the freshest ticks, not an ever-growing backlog of stale ones. And the try/finally fixes a third quiet bug in the original — if WriteAsync throws because the client vanished mid-write, the original code never reaches its unsubscribe line, leaking a handler on a singleton generator for the process lifetime. Event handlers on long-lived objects are where .NET memory leaks go to raise families.

Same shape, same contract, no spin, bounded memory, leak-free. This is my favorite code review in the whole repository precisely because the distance between “demo that works” and “code you can ship” is four changes you can name.

The client: await foreach in a trench coat

The console client (FeedR.Clients.Console) compiles the same proto with GrpcServices="Client" and gets a typed stub for free:

using var channel = GrpcChannel.ForAddress("http://localhost:5041");
var client = new PricingFeed.PricingFeedClient(channel);

var stream = client.SubscribePricing(new PricingRequest { Symbol = providedSymbol });
while (await stream.ResponseStream.MoveNext(CancellationToken.None))
{
    var current = stream.ResponseStream.Current;
    Console.WriteLine($"{current.Symbol} = {current.Value / 100M:F}");
}

A unary call first (GetSymbols) to list what's subscribable, then the stream. The consuming loop is just MoveNext until the server ends the stream — push semantics with pull-shaped code. Passing CancellationToken.None means the only way out is killing the process, which is fine for a demo and a habit to unlearn immediately afterward.

Would I choose gRPC here?

Honest trade-off time. For service-to-service or desktop/mobile-client streaming, gRPC server streaming is excellent: contract-first, strongly typed, HTTP/2 multiplexed, and dramatically less ceremony than hand-rolled WebSockets. For browsers, it's the wrong default — browsers can't speak native gRPC (no control over HTTP/2 frames), so you'd need gRPC-Web plus a proxy, at which point SignalR wins on practicality: transparent fallbacks, groups, and a JavaScript client that just works. My rule of thumb: SignalR when the subscriber is a browser, gRPC streaming when it's code, Redis pub/sub when it's another service you own. FeedR's console client sits squarely in gRPC's sweet spot, so the choice fits — just know it was made for that client, and a web dashboard would reopen the question.

Next, part 6 walks the weather feed — the one service that talks to the outside world — where an external REST API gets wrapped in a typed HttpClient, guarded by a Polly policy with one clause I still can't fully defend, and re-published as a push stream.