JSON Lines Streaming

Overview

Vovk.ts supports the JSON Lines  format, which answers one request with many responses. It is an output type with its own validation field, iteration. The response has the application/jsonl content type when the client sends an Accept: application/jsonl header. Otherwise it is text/plain, so a browser shows it when you open the endpoint URL.

Use JSON Lines for:

• A type-safe alternative to Server-Sent Events (SSE) for streaming data to clients.
• Long-running operations that produce results over time, such as LLM completions or database polling.
• Progressive data loading, which sends partial results as they become ready.

Creating a JSON Lines Generator Procedure

To stream JSON Lines, write the handler as a generator or an async generator. Each yielded value is serialized to JSON and sent as its own line, so the client gets each item after JSON: a Date arrives as a string. .fn() yields the values as they are.

import { z } from 'zod';
import { procedure, prefix, post, type VovkIteration } from 'vovk';
 
@prefix('stream')
export default class StreamController {
  @post('completions')
  static getJSONLines = procedure({
    // ...
    iteration: z.object({
      message: z.string(),
    }),
  }).handle(async function*() {
    const tokens: VovkIteration<typeof StreamController.getJSONLines>[] = [
      { message: 'Hello,' },
      { message: ' World' },
      { message: ' from' },
      { message: ' Stream' },
      { message: '!' },
    ];
 
    for (const token of tokens) {
      await new Promise((resolve) => setTimeout(resolve, 300));
      yield token;
    }
  });
}

To stream from a service, delegate to it with yield*:

import { z } from 'zod';
import { procedure, prefix, post } from 'vovk';
import StreamService from './stream-service';
 
@prefix('stream')
export default class StreamController {
  @post('completions')
  static getJSONLines = procedure({
    // ...
    iteration: z.object({
      message: z.string(),
    }),
  }).handle(async function* () {
    yield* StreamService.getJSONLines();
  });
}

import type { VovkIteration } from 'vovk';
import type StreamController from './stream-controller';
 
export default class StreamService {
  static async *getJSONLines() {
    const tokens: VovkIteration<typeof StreamController.getJSONLines>[] = [
      { message: 'Hello,' },
      { message: ' World' },
      { message: ' from' },
      { message: ' Stream' },
      { message: '!' },
    ];
 
    for (const token of tokens) {
      await new Promise((resolve) => setTimeout(resolve, 300));
      yield token;
    }
  }
}

On the client, read each line as it arrives with disposable  async iterators :

import { StreamRPC } from '@/client';
 
using stream = await StreamRPC.getJSONLines();
 
for await (const { message } of stream) {
  console.log('Received message:', message);
}

Besides Symbol.asyncIterator, Symbol.dispose and Symbol.asyncDispose, the iterable (stream above) has:

• status: the HTTP response status, such as 200 for OK or 404 for Not Found.
• asPromise: returns a promise that resolves with an array of all items when the stream ends. Every call gets the same items.
• onIterate: registers a callback for each item.
• abortController: an AbortController that aborts the stream. abortController.abort() makes the stream reader throw an AbortError, which the client finds in the error’s cause property.
• abortSilently: aborts the stream without an error on the stream reader, to stop reading it quietly.

With using, the stream is aborted with stream.abortSilently('Stream disposed') when it goes out of scope.

The client reads the response only as fast as the iteration takes items, so a slow consumer slows the stream down instead of buffering it. An item is kept until every running iteration has passed it. Iterations that run at the same time each get all items. An iteration that starts later begins at the oldest kept item, and iterating a consumed stream again yields nothing. After a break, the items already read stay for a later iteration or asPromise(). A line that isn’t JSON, such as a last line the connection cut short, ends the iteration with an error.

console.log('Response status:', stream.status);
stream.onIterate((item) => {
  console.log('Iterated item:', item);
});
if (someCondition) {
  stream.abortSilently();
}
console.log('All messages:', await stream.asPromise());

OpenAI Chat Example

This procedure streams OpenAI’s chat completions by delegating to them:

import { post, prefix, operation, type VovkRequest } from 'vovk';
import OpenAI from 'openai';
 
@prefix('openai')
export default class OpenAiController {
  @operation({
    summary: 'Create a chat completion',
  })
  @post('chat')
  static async *createChatCompletion(
    req: VovkRequest<{ messages: OpenAI.Chat.Completions.ChatCompletionMessageParam[] }>
  ) {
    const { messages } = await req.json();
    const openai = new OpenAI();
 
    yield* await openai.chat.completions.create({
      messages,
      model: 'gpt-5-nano',
      stream: true,
    });
  }
}

On the client, read the streamed completion:

// ...
using completion = await OpenAiRPC.createChatCompletion({
  body: { messages: [...messages, userMessage] },
});
 
for await (const part of completion) {
  // ...
}

View full example on examples.vovk.dev » 

JSONLinesResponder Class

JSONLinesResponder is the lower-level API behind generator handlers. It gives you more control: you send the lines yourself. It creates the ReadableStream that is the response body.

const responder = new JSONLinesResponder<IterationType>(req);

const responder = new JSONLinesResponder<IterationType>(req, ({ readableStream, headers }) => new Response(readableStream, { headers }));

The constructor takes two optional parameters:

• request?: Request | null: the incoming request. When given, the responder checks its Accept: application/jsonl header and sets Content-Type: application/jsonl in the headers of the response. Without the request or the header, the content type is text/plain. The RPC client streams only a JSON Lines response and returns any other Response as is, so pass the request when the client reads the stream. In a procedure with iteration, the request also lets the responder validate a line sent before the handler returns it.
• getResponse?: (responder: JSONLinesResponder<T>) => Response: builds a custom Response, for example with your own headers or other response options. Without it, the responder creates a Response with the default headers.

The responder has these members:

• send(item: T): Promise<void>: sends a JSON line to the client. The item is validated (when the procedure has iteration, only the first item, unless validateEachIteration: true is set), serialized to JSON and followed by a newline. Lines go out in call order, also when send() is not awaited. A send that fails, on iteration validation or with an item JSON can’t serialize, ends the stream as throw() does, and the segment’s onError gets the error.
• close(): Promise<void>: closes the response stream once the lines sent before it are out. A line sent after close() is dropped.
• throw(err: Error): Promise<void>: sends an error line after the lines sent before it, and closes the stream; a line sent after throw() is dropped. The error line is {"isError":true,"reason":"…"}, plus the statusCode of an HttpException, which is 500 for a status that is not a number from 200 to 599, as on a JSON response. The client rethrows it as an HttpException with that status. The error of notFound(), forbidden() or unauthorized() from next/navigation, which Next.js can no longer answer once the stream started, sends its status: 404, 403 or 401. In production, an error that is not an HttpException, or is one with status 0, which a client throws for a call that got no response, reads “Internal server error”.
• isClosed: boolean: whether the stream is closed, by close() or throw() (from the moment either is called), by a failed send, or by the client going away. send() waits while the client reads slower than the handler writes, and a generator handler is stopped, with its finally run, when the client disconnects or an item fails.
• response: Response: the Response that the Next.js route handler returns.
• headers: Record<string, string>: the content-type of the response.
• readableStream: ReadableStream<Uint8Array>: the stream used as the response body.

With JSONLinesResponder, the service method is a regular function, not a generator. It takes the responder and sends the lines through it.

import type { JSONLinesResponder, VovkIteration } from 'vovk';
import type StreamController from './stream-controller';
 
export type Token = VovkIteration<typeof StreamController.streamTokens>
 
export default class StreamService {
  static async streamTokens(responder: JSONLinesResponder<Token>) {
    const tokens: Token[] = [{ message: 'Hello,' }, { message: ' World' }, { message: '!' }];
 
    for (const token of tokens) {
      await new Promise((resolve) => setTimeout(resolve, 300));
      await responder.send(token);
    }
 
    responder.close();
  }
}

The handler returns the responder. The streaming runs in a promise that starts before return and isn’t awaited.

import { z } from 'zod';
import { prefix, get, procedure, JSONLinesResponder } from 'vovk';
import StreamService, { type Token } from './stream-service';
 
@prefix('stream')
export default class StreamController {
  @get('tokens')
  static streamTokens = procedure({
    iteration: z.object({ message: z.string() }),
  }).handle(async (req) => {
    const responder = new JSONLinesResponder<Token>(req);
 
    void StreamService.streamTokens(responder);
 
    return responder;
  });
}

The iteration schema gives Token its type, in the service and in the RPC client.

To end the stream with an error, call throw. The client rethrows the error:

await responder.throw(new Error('Stream error'));