Streaming deep-dive

Chapter 8 built WatchShipment as a four-event, millisecond-long demonstration and closed by naming two questions it left open: what happens when a server produces events faster than a client can read them, and what keeps a stream alive across the minutes or days a real voyage’s tracking feed might sit quiet. Both questions only bite once a stream outlives the toy case — a fixed, four-item list a server can build and send before the reader has even started paying attention. This chapter puts a real answer on each one, on the same RPC, by changing how WatchShipment produces its events rather than changing the contract: a demand-driven feed on both stacks, a slow reader to pace against, and a keepalive configuration that survives a quiet stretch.

The code is in examples/17-streaming-at-scale/. The verify.sh script there builds and runs both stacks, drives a bundled slow-reading client against each, and asserts the timing claims this chapter makes; its README.md covers the full mode table and a real run’s numbers.

What a consumer that stops reading actually does

HTTP/2 answers Chapter 8’s first open question at the transport layer, below anything a service’s code decides: every stream carries its own flow-control window, a byte budget the receiver advertises for how much unacknowledged DATA it is willing to buffer for that one stream. A sender that has filled a receiver’s advertised window stops sending DATA frames on that stream — not an error, not a dropped frame, just silence — until a WINDOW_UPDATE frame from the receiver says more room has opened up. A second window operates one level up, scoped to the whole connection rather than one stream, so a single slow stream can’t starve every other call multiplexed over the same connection; Chapter 6 covered that multiplexing, and this is the mechanism that keeps one misbehaving stream from blocking the rest. Both windows default to a modest size (64 KiB is the HTTP/2 spec’s own starting point, which grpc-java and grpc.aio both begin from and can grow dynamically from there), small enough that a consumer which stops reading mid-stream can exhaust it inside a few hundred kilobytes of unacknowledged payload.

Two moments on the same HTTP/2 connection. At t0, a server sends DATA frames into a 64 KiB stream window while a client reads normally and returns WINDOW_UPDATE frames that replenish the credit. At t1, the same client has stopped draining its receive buffer; the window shrinks toward zero, WINDOW_UPDATE frames stop arriving, and the server's sender blocks with no credit left to write, shown as a note under the stalled connection. — Figure 17.1 — the same connection, two moments apart: a reading consumer keeps the window replenished, a stalled one lets it run out.

This is backpressure at its most literal: a slow reader doesn’t cause errors or dropped messages, it causes the sender to stop sending, full stop, until the reader catches up or the call’s deadline (Chapter 10) expires first. The number of bytes a server can get away with producing before that happens is small enough that any RPC meant to run for more than a handful of messages has to decide how its own code behaves once the transport applies this pressure — which is exactly the decision WatchShipment’s implementation makes in this chapter, on both stacks, by never producing an event the application hasn’t been asked for yet.

Reactive backpressure: Mutiny demand vs. async-generator pull

A gRPC server library has to turn “the transport won’t accept more bytes right now” into something the application’s code can react to, because the alternative — buffering an unbounded queue of already-built responses in server memory while the transport blocks — just moves the problem from the network into the heap. Both stacks solve this the same way in spirit, with different vocabulary.

On Quarkus, Multi<T> is smallrye-mutiny’s type for “zero or more items, then completion or failure,” and it is a full Reactive Streams Publisher: a Multi only produces as many items as its subscriber has requested, never ahead of that number. Mutiny’s own guide says as much — “emissions are constrained by the request emitted from downstream subscribers” — and the generated gRPC server bridge (io.quarkus.grpc.stubs.ServerCalls) is wired to exactly that contract: it wraps the outbound ServerCallStreamObserver with setOnReadyHandler, so the moment the HTTP/2 stream becomes ready to accept more bytes, the bridge asks the Multi for more items; the moment it isn’t ready (Figure 17.1’s stalled case), it simply doesn’t ask, and nothing downstream of that ask is forced to run. A Multi built from Multi.createFrom().items(aList) — the shape Chapter 8 used for its four fixed events — satisfies that contract by handing over items from an already-built list on request; nothing stops the list itself from having been built eagerly, all at once, long before the first item was ever requested. This chapter’s bkg_eager mode exists to make that contrast visible: a Multi can be backpressure-safe on the way out while still having done all of its production work up front, with no connection to when the consumer actually wanted it.

On Python, grpc.aio expresses the identical idea with a language feature instead of a library type: a server-streaming handler written as async def ... yield ... is a native async generator, and an async generator is pull-based by construction — nothing in its body past the current yield executes until something calls __anext__() on it again. grpc.aio’s internal plumbing drives that call only when the stream has room to send another message, the same signal Quarkus’s bridge reacts to via setOnReadyHandler. No separate demand API is needed on the Python side; the generator protocol is the pull contract, which is why Chapter 8 could already describe Python’s shape as “naturally pull-based” without this chapter’s machinery.

Two producers side by side. On the left, a demand-driven Multi or async generator for bkg_demo produces three events spread across 0, 260, and 520 milliseconds, one in flight at a time, pulled by onRequest or a yield; a consumer box pulls each one as it is ready. On the right, an eager Multi.createFrom().items(list) or a pre-built Python list for bkg_eager shows three events all stamped at approximately zero milliseconds, already built before the first byte leaves the server; a consumer box reads a buffer the producer never waited on. — Figure 17.2 — demand-paced production spreads event timestamps across real time; eager pre-materialization clusters them at the instant the call started.

The two shapes are not a Quarkus-versus-Python difference — they’re the same reactive-pull idea wearing two different syntaxes, which is the whole reason Chapter 8 could already preview both without this chapter’s code: Mutiny’s onRequest callback and Python’s generator-resume are the identical signal, “the consumer is ready for one more,” arriving through whichever mechanism each ecosystem already had.

How the code works

examples/17-streaming-at-scale/quarkus/.../TrackingEvents.java and python/app/events.py each define two waypoint tables and one builder function, kept separate from how the events get scheduled, because scheduling is this chapter’s subject and the event content isn’t. DEMO_WAYPOINTS is six (Location, BookingState) pairs extending Chapter 8’s PENDING → CONFIRMED → IN_TRANSIT → DELIVERED progression with two extra IN_TRANSIT waypoints — a second port call at Rotterdam, then a transshipment leg through Hamburg before the Singapore delivery — so there’s enough length in the feed to show a measurable production cadence; QUIET_WAYPOINTS is two pairs, PENDING then DELIVERED with no intermediate state, built only to carry a multi-second gap between them. buildEvent(bookingId, index, waypoints) (Java) and build_event(booking_id, index, waypoints) (Python) do the one thing both modes need identically: stamp event.at with the wall-clock instant this call runs, and set event.note to a bare "seq=<index>" marker rather than Chapter 8’s narrative prose, so verify.sh can read production order and production time straight off the wire instead of parsing sentences.

TrackingServiceImpl.watchShipment (Java) and TrackingServiceServicer.WatchShipment (Python) dispatch on request.booking_id — a demo-only mode selector carrying no domain meaning beyond picking a feed shape, the same trick Chapter 15’s origin.name flag used to switch FindOptimalRoute’s execution model without touching the frozen contract. "bkg_eager" is the simplest branch: build the whole six-event list in one tight loop — stamping every at field within microseconds of the others, since nothing yields control between iterations — then hand that finished list to Multi.createFrom().items(...) (Java) or a plain for event in events: yield ... loop (Python). Everything else falls through to the demand-driven path, which both stacks give a delaysMs/delays_s array: [0, 260, 260, 260, 260, 260] milliseconds for bkg_demo’s six events, [0, 2000] for bkg_quiet’s two — the first entry is always zero so the opening event is immediate, matching Chapter 8’s feel of an instant first response.

The Python side’s demand-driven path is the shorter of the two precisely because the language already does the work: _demand_driven_sequence is an async def generator that, for each waypoint, awaits asyncio.sleep(delays_s[i]) (never a blocking time.sleep, which would stall the whole event loop, not just this call) and then yields the freshly built event. Nothing about this loop checks for permission to proceed beyond the await itself — the generator protocol already guarantees nothing past a yield runs until the consumer resumes it, which is what makes “pull-based by construction” a precise description rather than a slogan.

The Quarkus side has to build that same guarantee explicitly, because Multi.createFrom().emitter(...) hands you a raw MultiEmitter with no scheduling of its own — demandDrivenMulti and scheduleStep are the code that turns “an item every delaysMs[i] milliseconds” into “an item every delaysMs[i] milliseconds, but only while a subscriber has asked for one.” scheduleStep reads the current index, and if there are waypoints left, schedules a produce block — via Vertx.setTimer, never Thread.sleep, so the pacing costs no thread on an event-loop-based server — that builds and emits one event, advances the index, and then checks emitter.requested(): if the subscriber still has outstanding demand, it calls itself again immediately for the next waypoint; if not, it stops and flips a producing flag back to false, so the chain sits idle until something restarts it. emitter.isCancelled() is checked on every produce invocation too, so a caller that drops the connection mid-feed (Chapter 10’s cancellation story) stops this loop from doing any more work rather than continuing to schedule timers for a reader that is no longer there.

What restarts the chain is MultiEmitter.onRequest(LongConsumer), registered once per call — but the demo build’s first run against the actual quarkus-grpc bridge surfaced a real timing subtlety: the generated bridge’s subscriber calls Subscription.request(256) (its fixed prefetch batch) as part of onSubscribe, which happens before the Multi.createFrom().emitter(...) consumer lambda — the block where onRequest gets registered — ever runs. That first request(256) still lands in emitter.requested(), because the emitter’s own counter is updated regardless of whether a callback is attached to observe it, but no callback fires for it, since none was registered yet at the moment it happened. A demandDrivenMulti that only started production from inside onRequest would stay silent forever on a six-event feed, waiting for a demand signal that already came and went. The fix is the maybeStart check both call sites in demandDrivenMulti share: check emitter.requested() > 0 and try to start producing immediately, in addition to on every later onRequest call — the first catches the prefetch demand that already landed before registration, the second catches any demand granted afterward (the replenish-after-exhaustion case a longer feed would actually exercise). This is the kind of detail that only shows up by running the real generated bridge rather than reasoning about the Multi contract in isolation, and it’s exactly the “write the code yourself” level this chapter is aiming for: the Reactive Streams contract (“you’ll be asked before you may produce”) doesn’t specify when the first ask happens relative to when your code starts listening for it, and a correct implementation has to handle both orders.

The bundled python/app/slow_consumer.py is this chapter’s slow reader: it opens a WatchShipment call, reads one event, prints it as a JSON line (including the server-stamped event.at), awaits asyncio.sleep(0.3), and reads the next — a consumer that takes roughly three times as long to process an event as bkg_demo’s producer takes to make one. It ships inside the Python image, reusing the same generated stubs the server uses, and runs as a one-off container on the compose network: docker compose -p ch17-streaming run --rm python python -m app.slow_consumer quarkus:9000 bkg_demo reaches the Quarkus service by its compose service name and container port, not the host-mapped 9170, which is what lets the identical command target either stack by swapping one word.

Keepalive on a long-lived stream

Chapter 8’s second open question was about time, not bytes: a stream that sits quiet for seconds, minutes, or — for a real voyage — days has to survive network infrastructure between client and server (load balancers, NAT devices, stateful firewalls) that often closes a TCP connection it hasn’t seen traffic on recently, silently, with no error delivered to either endpoint until the next write fails. gRPC’s answer is a keepalive PING frame: a side channel on the same connection, invisible to application code, that either endpoint can send on a schedule to prove the connection is still alive, answered by the peer with an ACK. A connection with keepalive configured generates its own traffic across an idle stretch precisely so a middlebox never sees it go quiet long enough to reap; a connection without it depends entirely on how long the quietest link in the path tolerates silence, which no application can observe or control from either end.

Two outcomes for the same quiet gap on a WatchShipment stream. On top, with server keepalive configured, the server sends a PING roughly every thirty seconds and the client answers with an ACK, holding the connection open across the gap until the next real event arrives. On the bottom, without keepalive, the connection sits idle after the last DATA frame with no PING traffic, and a middlebox reaps it before the final event can be delivered. — Figure 17.3 — the same quiet gap survives with a keepalive PING/ACK cycle and is at risk without one; this figure is illustrative, since the actual reaping behavior depends on middlebox configuration this chapter's example never exercises.

Quarkus’s gRPC server runs, by default, on grpc-java’s own Netty server rather than the shared Vert.x HTTP server (quarkus.grpc.server.use-separate-server defaults to true), and that server’s keepalive knobs sit one level deeper in the configuration tree than a first guess might expect: not quarkus.grpc.server.keep-alive-time directly, but quarkus.grpc.server.netty.keep-alive-time, alongside netty.permit-keep-alive-time and netty.permit-keep-alive-without-calls — three properties that map straight onto NettyServerBuilder’s keepAliveTime, permitKeepAliveTime, and permitKeepAliveWithoutCalls calls. keep-alive-time is how often this server pings an otherwise-silent client; permit-keep-alive-time is the shortest ping interval this server tolerates from a client’s own keepalive pings before treating them as abusive and tearing down the connection with a GOAWAY; permit-keep-alive-without-calls, set to true here, lets a client keep a connection warm even between calls, with no RPC actively in flight. This example sets all three — 30s / 10s / true — in application.properties.

Python’s grpc.aio.server takes the equivalent settings as channel-argument tuples rather than a config file, documented in grpc’s own keepalive guide: grpc.keepalive_time_ms, grpc.keepalive_timeout_ms, grpc.keepalive_permit_without_calls, and grpc.http2.min_ping_interval_without_data_ms — the same four concerns (how often this server pings, how long it waits for an ACK, whether it pings with no call active, and the floor on how often it tolerates a client’s own pings) expressed as the options list app/main.py passes to grpc.aio.server(options=...).

Neither the HTTP/2 window sizes from Figure 17.1 nor keepalive’s survival against a real middlebox’s reaping policy is something verify.sh can assert deterministically — a loopback container network never naturally exhausts a 64 KiB window on a handful of small messages, and there is no middlebox in this stack’s compose topology to reap anything. Both are illustrative, covered by this prose, the figures, and the “On the wire” box below, not by a numeric gate.

Decision

Reach for a demand-driven producer — Multi.createFrom().emitter(...) gated on real demand, or a plain async generator — the moment a stream's length isn't bounded to a handful of items known in advance, or its production work is expensive enough that doing it all up front would waste work a cancelled or short-lived call never needed. Multi.createFrom().items(list) (or building a Python list before the first yield) stays the right default for Chapter 8's shape: a short, fixed, cheap-to-build sequence where "pre-materialized" and "demand-driven" produce identical wire behavior and the simpler code wins. Configure keepalive on every stream a caller might hold open longer than the shortest idle timeout anywhere in its network path — which in practice means every long-lived stream, since that timeout is rarely visible to either endpoint. The three numbers worth setting explicitly rather than leaving at a library default are the server's own ping interval (how fast it notices or masks a silently dead peer), the floor it enforces on a client's pings (protection against a misconfigured or hostile client turning keepalive into a denial-of-service vector), and whether pings are permitted with no call in flight (needed for a client that holds a channel open between bursts of calls, unneeded and wasteful for one that reconnects per call).

Build, run, observe

cd examples/17-streaming-at-scale
./verify.sh

This brings both services up with docker compose, waits for each gRPC server to answer reflection’s list, then against each of localhost:9170 (Quarkus) and localhost:9171 (Python): reads bkg_demo’s full six-event stream and checks its count, order, and fields; times that same call and confirms the whole stream took at least six-tenths of a second, proof the events arrived spread over time rather than buffered and dumped; runs the bundled slow consumer against both bkg_demo and bkg_eager and compares the server-stamped event.at deltas between them; reads bkg_quiet and confirms both its events arrive, including the one after the two-second gap, with the call’s total elapsed time proving the gap was real; and confirms ReportPositions and Sync still answer UNIMPLEMENTED. A real run against both stacks measured bkg_demo’s consecutive deltas at roughly 260 milliseconds and bkg_eager’s at well under a millisecond — a three-order-of-magnitude gap between a producer that waits and one that doesn’t, from the same event content on the same contract.

What you learned

HTTP/2 flow control is the transport-level root of backpressure: a per-stream and a per-connection byte window, replenished by WINDOW_UPDATE frames a reader only sends once it has actually drained what it already received — a consumer that stops reading makes a sender stop sending, not error or drop data.
Multi is a full Reactive Streams Publisher: it only produces what its subscriber has requested, which is what lets Multi.createFrom().emitter(...) paired with MultiEmitter.onRequest build a server-streaming RPC that never gets ahead of what the HTTP/2 stream is ready to carry — and what makes Multi.createFrom().items(list) safe on the way out even when the list behind it was built eagerly.
Python’s async def ... yield server-streaming handler is pull-based by construction, with no separate demand API needed — the language’s own generator-resume protocol is the signal Mutiny’s onRequest callback exists to provide explicitly.
Keepalive PING/ACK traffic is what holds a long-lived, otherwise-quiet stream open against middleboxes that reap idle connections silently; Quarkus exposes it under quarkus.grpc.server.netty.*, one level deeper than the top-level quarkus.grpc.server.* properties this book has used elsewhere, and Python exposes the same four concerns as grpc.aio.server(options=...) channel arguments.

Chapter 18 turns from correctness under load to speed under load: channel and stub reuse (the single most common gRPC performance mistake), keepalive revisited from a connection-pooling angle, message-size and compression trade-offs, and a real ghz benchmark run against RoutingService.GetQuote and ListRoutes rather than the loose, relative timing checks this chapter and Chapter 15 have both leaned on.

On the wire

Reading a demand-driven stream over grpcurl looks identical to Chapter 8’s four-event read — grpcurl has no way to see that this server is pacing itself between messages rather than sending all six at once, because from the wire’s perspective both look like DATA frames arriving when they arrive:

grpcurl -plaintext -d '{"bookingId":"bkg_demo"}' \
  localhost:9170 shipping.tracking.v1.TrackingService/WatchShipment

{"event":{"bookingId":"bkg_demo","location":{"unLocode":"USNYC","name":"New York"},"state":"BOOKING_STATE_PENDING","at":"2026-01-02T08:00:00.123456Z","note":"seq=0"}}
{"event":{"bookingId":"bkg_demo","location":{"unLocode":"USNYC","name":"New York"},"state":"BOOKING_STATE_CONFIRMED","at":"2026-01-02T08:00:00.383512Z","note":"seq=1"}}

— each at roughly 260 milliseconds after the one before it, continuing through all six events. The flow-control mechanics Figure 17.1 describes are real but invisible at this layer; seeing the actual WINDOW_UPDATE traffic (or its absence once a reader stalls) needs a frame-level trace, for example GODEBUG=http2debug=2 against a Go HTTP/2 client, or grpcurl -v to see headers and trailers without full frame detail. This book doesn’t ship a frame-trace capture as part of verify.sh — the window and keepalive numbers stay illustrative, exactly as the “Keepalive on a long-lived stream” section says, confirmed by this chapter’s prose and figures rather than gated by an assertion.

Verification status: unverified. verify.sh ran green against both stacks on the authoring machine, including the count/order/fields check, the incremental-delivery lower bound, the slow-consumer backpressure relative gate, and the keepalive quiet-stream check — a real run measured bkg_demo’s consecutive deltas at roughly 260ms and bkg_eager’s at microseconds, comfortably clearing the assertions’ margins. The highest-risk things a reader’s own run should confirm: that the ~200ms and ~0.6s timing floors hold on meaningfully slower or more contended hardware than the authoring machine, and that the MultiEmitter.onRequest prefetch-timing detail this chapter documents (the generated bridge’s initial request(256) landing before onRequest is registered) holds across future quarkus-grpc releases rather than being an implementation detail of the pinned 3.40.1 version this chapter verified against.