Function writeEventStream

src/aws-event-stream.ts:118–161

(
  res: http.ServerResponse,
  events: Array<{ eventType: string; payload: object }>,
  options?: {
    latency?: number;
    streamingProfile?: StreamingProfile;
    recordedTimings?: RecordedTimings;
    replaySpeed?: number;
    signal?: AbortSignal;
    onChunkSent?: () => void;
  },
)

 * the provided abort signal.
 */
export async function writeEventStream(
  res: http.ServerResponse,
  events: Array<{ eventType: string; payload: object }>,
  options?: {
    latency?: number;
    streamingProfile?: StreamingProfile;
    recordedTimings?: RecordedTimings;
    replaySpeed?: number;
    signal?: AbortSignal;
    onChunkSent?: () => void;
  },
): Promise<boolean> {
  const opts = options ?? {};
  const latency = opts.latency ?? 0;
  const profile = opts.streamingProfile;
  const { recordedTimings, replaySpeed } = opts;
  const signal = opts.signal;
  const onChunkSent = opts.onChunkSent;

  if (res.writableEnded) return true;
  res.setHeader("Content-Type", "application/vnd.amazon.eventstream");
  res.setHeader("Transfer-Encoding", "chunked");

  let chunkIndex = 0;
  for (const event of events) {
    const chunkDelay = calculateDelay(chunkIndex, profile, latency, recordedTimings, replaySpeed);
    if (chunkDelay > 0) {
      await delay(chunkDelay, signal);
    }
    if (signal?.aborted) return false;
    if (res.writableEnded) return true;

    const frame = encodeEventStreamMessage(event.eventType, event.payload);
    res.write(frame);
    onChunkSent?.();
    if (signal?.aborted) return false;
    chunkIndex++;
  }

  if (!res.writableEnded) {
    res.end();
  }
  return true;
}