16. Async Generators | Abhimanyu Aryan

Async generators produce values lazily over time. Used extensively for SSE parsing and WebSocket streams.

SSE Parser:

File: packages/ai/src/providers/openai-codex-responses.ts L469-510

async function* parseSSE(response: Response): AsyncGenerator<Record<string, unknown>> {
  if (!response.body) return;
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      buffer += decoder.decode(value, { stream: true });
      // Parse and yield JSON events from buffer...
      yield JSON.parse(data);
    }
  } finally {
    await reader.cancel();
    reader.releaseLock();
  }
}

Composing generators (piping output of one into another):

File: packages/ai/src/providers/openai-codex-responses.ts L431-457

async function* mapCodexEvents(events: AsyncIterable<Record<string, unknown>>): AsyncGenerator<ResponseStreamEvent> {
  for await (const event of events) {
    // Transform and yield...
    yield event as unknown as ResponseStreamEvent;
  }
}

Consuming with for await:

File: packages/ai/src/providers/openai-responses-shared.ts L286-298

export async function processResponsesStream<TApi extends Api>(
  openaiStream: AsyncIterable<ResponseStreamEvent>,
  output: AssistantMessage,
  stream: AssistantMessageEventStream,
  model: Model<TApi>
): Promise<void> {
  for await (const event of openaiStream) {
    // Process each event...
  }
}