|
| 1 | +# Event Streaming |
| 2 | + |
| 3 | +A function that produces its result progressively — an LLM relay emitting tokens, a |
| 4 | +long-running job reporting progress — should not make its caller wait for the whole |
| 5 | +answer. This chapter is the wrapper's half of the platform-wide streaming contract: |
| 6 | +the same paradigm on all four runtimes (Java, Rust, Python, Node.js): |
| 7 | + |
| 8 | +> **The caller provides a reply address; the callee streams events to it until a |
| 9 | +> terminal signal.** |
| 10 | +
|
| 11 | +Each segment is one event to the caller's `reply_to`, marked with the reserved |
| 12 | +envelope header `x-event-stream: data | eof | exception`. A calling engine renders |
| 13 | +the segments out its HTTP edge, hands them to a flow, or relays them onward — your |
| 14 | +Python function neither knows nor cares. |
| 15 | + |
| 16 | +## Write a streaming function |
| 17 | + |
| 18 | +A streaming producer is an **interceptor**: it receives the raw `EventEnvelope` |
| 19 | +(so the caller's reply address travels the engines' way) and replies through |
| 20 | +`EventStreamWriter` instead of a return value: |
| 21 | + |
| 22 | +```python |
| 23 | +from mercury_composable import EventEnvelope, EventStreamWriter, preload |
| 24 | + |
| 25 | +@preload(route="hello.tokens", instances=10, interceptor=True) |
| 26 | +async def stream_tokens(headers: dict[str, str], event: EventEnvelope): |
| 27 | + out = EventStreamWriter.from_request(event) |
| 28 | + out.first(200, "text/event-stream") # head control rides the first event |
| 29 | + out.write("The answer is") # data segment |
| 30 | + out.write_named("tokens", {"n": 2}) # named (typed) SSE event |
| 31 | + out.close({"usage": {"tokens": 2}}) # end of transmission + trailing metadata |
| 32 | + # or out.fail(e) # in-band failure |
| 33 | +``` |
| 34 | + |
| 35 | +The writer is the engines' exact API. `first(status, content_type, ttl_seconds=None)` |
| 36 | +declares the response head and, optionally, the idle allowance between segments; |
| 37 | +`fail(e)` carries the standard error key-values |
| 38 | +`'{"type": "error", "status": n, "message": text}'`; writes after `close()`/`fail()` |
| 39 | +are dropped. Plain-`def` handlers can stream too — the writer bridges from the |
| 40 | +executor thread back to the host loop. |
| 41 | + |
| 42 | +An interceptor's return value is never auto-replied. To answer single-shot from an |
| 43 | +interceptor (a relay that sometimes buffers, for example), send a plain envelope to |
| 44 | +`event.reply_to` yourself. An uncaught exception becomes the standard error envelope |
| 45 | +to the caller — single-shot before the stream starts, in-band after. |
| 46 | + |
| 47 | +## How it crosses the wire |
| 48 | + |
| 49 | +When a calling engine (or `curl`) invokes your streaming function through |
| 50 | +`POST /api/event` with `Accept: text/event-stream`, the host answers the same call |
| 51 | +with a Server-Sent Events response in the platform's hybrid dialect: |
| 52 | + |
| 53 | +- **envelope frames** — the reserved SSE event name `envelope`, one base64-encoded |
| 54 | + serialized envelope per frame — carry everything with envelope semantics: the |
| 55 | + first event (head control), the `eof`/`exception` terminals, and any segment that |
| 56 | + cannot round-trip as plain text (a dict or bytes body, text containing a carriage |
| 57 | + return, an event name colliding with the reserved word); |
| 58 | +- **raw SSE frames** carry plain text segments, so token relays stay near-zero |
| 59 | + overhead. |
| 60 | + |
| 61 | +Everything degrades explicitly: a caller that did not opt in receives |
| 62 | +`406 Streaming function requires a caller that accepts text/event-stream` instead of |
| 63 | +a truncated reply; a non-streaming (single-shot) answer over the capable path is |
| 64 | +byte-identical to a normal RPC reply; idle expiry fails the stream in-band with the |
| 65 | +standard 408 error body. The `x-ttl` request header (ms) is the idle allowance |
| 66 | +between segments — your `first(..., ttl_seconds=...)` can extend it for the whole |
| 67 | +stream. While the producer is quiet, the host emits `: ping` keep-alive comments |
| 68 | +(`event.stream.keep.alive`, the engines' config key — default 30s, `0` disables). |
| 69 | + |
| 70 | +## Consume a stream |
| 71 | + |
| 72 | +`PostOffice.stream()` is the consumer surface — an async iterator yielding the same |
| 73 | +decoded envelopes an engine reply route receives: `data` segments, then the terminal. |
| 74 | +It works against a remote peer's `/api/event` (an engine or another function host) |
| 75 | +and against local functions alike, and opting in is always safe — a non-streaming |
| 76 | +target simply yields its one classic reply: |
| 77 | + |
| 78 | +```python |
| 79 | +from mercury_composable import PostOffice |
| 80 | + |
| 81 | +async with PostOffice() as po: |
| 82 | + async for segment in po.stream("hello.tokens", None, |
| 83 | + endpoint="http://127.0.0.1:8100/api/event", |
| 84 | + timeout_ms=30000): |
| 85 | + marker = segment.headers.get("x-event-stream") |
| 86 | + if marker == "data": |
| 87 | + print(segment.body) |
| 88 | + elif marker == "exception": |
| 89 | + raise RuntimeError(segment.body["message"]) |
| 90 | + # eof: segment.body carries the trailing metadata, if any |
| 91 | +``` |
| 92 | + |
| 93 | +`timeout_ms` is the idle allowance between segments. The consumer guards the dialect |
| 94 | +for you: a malformed frame, a stream that ends without a terminal, or idle expiry |
| 95 | +each yield the standard in-band exception envelope, then the iterator ends. |
| 96 | + |
| 97 | +## Compose a relay |
| 98 | + |
| 99 | +The pattern the whole streaming program is built on: forward **your own caller's** |
| 100 | +reply address into a call against a remote streaming function, and the segments flow |
| 101 | +`engine → your function → remote peer → back to the original caller` with no |
| 102 | +buffering anywhere: |
| 103 | + |
| 104 | +```python |
| 105 | +@preload(route="llm.relay", instances=10, interceptor=True) |
| 106 | +async def relay(headers: dict[str, str], event: EventEnvelope): |
| 107 | + async with PostOffice() as po: |
| 108 | + await po.stream_to("remote.tokens", None, |
| 109 | + reply_to=event.reply_to or "", |
| 110 | + endpoint="http://peer:8085/api/event", |
| 111 | + cid=event.cid, timeout_ms=30000) |
| 112 | +``` |
| 113 | + |
| 114 | +`stream_to()` forwards every decoded envelope verbatim to the named LOCAL route |
| 115 | +(here, the reply sink the host opened for your caller) and returns the terminal. |
| 116 | +Combined with a calling engine's `stream: true` endpoint, this streams a remote |
| 117 | +peer's tokens progressively out that engine's HTTP edge — with zero imperative |
| 118 | +streaming code in between. |
| 119 | + |
| 120 | +## See also |
| 121 | + |
| 122 | +- The engines' HTTP Response Streaming guides (the same contract at the HTTP edge): |
| 123 | + [Java](https://accenture.github.io/mercury-composable/guides/http-streaming/) · |
| 124 | + [Rust](https://accenture.github.io/mercury/guides/http-streaming/) |
| 125 | +- [Interop Test Report — Progressive Rendering](../test-reports/progressive-rendering-interop.md) — |
| 126 | + the live four-runtime validation of this contract |
| 127 | +- [HTTP Surface Reference](http-surface-reference.md) — the `/api/event` contract |
| 128 | +- [Function Writing Patterns](function-patterns.md) |
0 commit comments