OpenAPI Code Generation with Corvus: Streaming
At endjin, we maintain Corvus.JsonSchema, and in the previous post we wired a generated client to a generated server over real HTTP.
Now let's look at a pattern that's become increasingly common since the rise of LLM-backed APIs: streaming responses.
Why streaming matters now
A traditional REST call returns a single JSON document. The client waits, the server responds, done. But LLM completions, real-time feeds, and event-driven UIs don't work that way. They produce a sequence of items over time, and the client needs to start processing before the full response is available.
HTTP has two established patterns for this: Server-Sent Events (SSE), where each item is framed as data: {...}\n\n, and Newline-Delimited JSON (NDJSON), where each item is a single JSON line terminated by \n. Both are widely supported and work through proxies, load balancers, and CDNs.
The challenge for code generators is that these aren't regular request/response cycles. The server needs a way to push items incrementally, and the client needs to consume them as they arrive. Ideally, they do so as typed, validated documents rather than raw strings. That's what the Corvus generator handles for both sides.
Declaring a streaming response in OpenAPI
In your OpenAPI spec, a streaming endpoint looks like a regular response with text/event-stream or application/x-ndjson as the media type, plus an itemSchema that describes the shape of each individual item:
/chat:
post:
operationId: startVetChat
responses:
"200":
content:
text/event-stream:
schema:
type: array
items:
$ref: "#/components/schemas/ChatChunk"
The generator reads this and produces typed infrastructure on both sides.
Server: the writer callback
On the server, the generated result factory doesn't return a body directly. Instead, it accepts a writer callback. This is an async delegate that appends typed items to a stream. The generated infrastructure handles serialization and framing:
public ValueTask<StartVetChatResult> HandleStartVetChatAsync(
StartVetChatParams parameters,
JsonWorkspace workspace,
CancellationToken cancellationToken = default)
{
return new(StartVetChatResult.Ok(static async (stream, cancellationToken) =>
{
ChatChunk greeting = ChatChunk.ParseValue(
"""{"delta":"Hello! ","done":false}"""u8);
await stream.AppendChatChunk(greeting, cancellationToken);
ChatChunk answer = ChatChunk.ParseValue(
"""{"delta":"How can I help?","done":true}"""u8);
await stream.AppendChatChunk(answer, cancellationToken);
}));
}
Each call to AppendChatChunk serializes the item as compact JSON and writes the SSE frame (data: {"delta":"Hello! ","done":false}\n\n) to the response stream. The response stays open until the callback returns. There's no explicit "end stream" method. When you're done appending, you just return, and the generated endpoint flushes and closes the HTTP response.
If the client disconnects mid-stream, the cancellation token fires. Your callback can check it between items, or simply let the next Append call throw OperationCanceledException. Either way, you don't leak connections.
Client: IAsyncEnumerable of typed documents
On the client side, the generated response type exposes EnumerateOkItems(). It is an IAsyncEnumerable that yields pooled, typed documents as they arrive:
await using StartVetChatResponse chatResponse = await chatClient.StartVetChatAsync(
body: new ChatRequest.Source((ref ChatRequest.Builder b) =>
{
b.Create(question: "My cat won't eat. What should I do?"u8);
}));
await foreach (ParsedJsonDocument<ChatChunk> chunk in chatResponse.EnumerateOkItems())
{
using (chunk)
{
Console.Write(chunk.RootElement.Delta);
}
}
Each chunk is a pooled document. The using inside the loop returns the pooled memory after you've read the item, so you never accumulate the full stream in memory. This matters for long-running streams where thousands of items might flow through.
The generated code handles the SSE frame parsing transparently. It strips the data: prefix, handles multi-line data fields, ignores comment lines, and recognises the double-newline boundary. You just iterate typed objects.
SSE metadata: event IDs and types
Sometimes you need more than just the payload. SSE supports id: and event: fields that carry stream position and event discrimination. If you need those, use EnumerateOkSseItems() instead:
await foreach (SseItem<ChatChunk> item in chatResponse.EnumerateOkSseItems())
{
using (item.Document)
{
Console.WriteLine($"Event ID: {item.Id}, Type: {item.EventType}");
Console.Write(item.Document.RootElement.Delta);
}
}
This is particularly useful for reconnection scenarios where you need to send a Last-Event-ID header to resume from where you left off.
NDJSON: the simpler framing
For endpoints that use application/x-ndjson, the pattern is identical from your perspective. On the server, the same writer callback appends typed items:
public ValueTask<StreamPetActivityResult> HandleStreamPetActivityAsync(
StreamPetActivityParams parameters,
JsonWorkspace workspace,
CancellationToken cancellationToken = default)
{
return new(StreamPetActivityResult.Ok(static async (stream, cancellationToken) =>
{
ActivityEvent checkIn = ActivityEvent.ParseValue(
"""{"eventId":"evt-1","timestamp":"2026-05-30T18:00:00Z","type":"check-in","description":"Bella checked in"}"""u8);
await stream.AppendActivityEvent(checkIn, cancellationToken);
}));
}
The difference is purely in the wire format: NDJSON writes {...}\n (one line per item, no data: prefix). The client-side EnumerateOkItems() works the same way, yielding typed documents one per line.
NDJSON is a better fit when you don't need SSE's reconnection semantics or event typing. It's simpler, slightly more compact, and easier to process with command-line tools like jq.
Cancellation and backpressure
The writer callback model gives you natural backpressure. Each Append call is awaitable. If the transport buffer is full because the client isn't reading fast enough, Append won't complete until there's space. You don't need to implement flow control yourself.
For long-running streams (think a real-time telemetry feed), combine the cancellation token with your data source:
return new(StreamTelemetryResult.Ok(async (stream, cancellationToken) =>
{
await foreach (SensorReading reading in sensorService.GetReadingsAsync(cancellationToken))
{
await stream.AppendSensorReading(reading, cancellationToken);
}
}));
When the client disconnects, the token cancels, the await foreach exits, the callback returns, and the response closes. Clean.
What's next
In the [ref slug=openapi-code-generation-with-corvus-authentication text=next post], we'll look at authentication - generated OAuth2 scope constants, Entra ID integration, API keys, and cookie-based auth, all using standard .NET middleware patterns.
