# Where to Find the Streaming Engine Code in OmniRoute: A Complete Guide

> Find the streaming engine code in OmniRoute at open-sse/utils/stream.ts. This guide shows you where to locate the createSSEStream factory function for Server-Sent Events processing.

- Repository: [Diego Rodrigues de Sa e Souza/OmniRoute](https://github.com/diegosouzapw/OmniRoute)
- Tags: how-to-guide
- Published: 2026-07-02

---

**The streaming engine code in OmniRoute is located in [`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts), which exports the `createSSEStream` factory function that builds a TransformStream for processing Server-Sent Events (SSE).**

OmniRoute unifies LLM provider responses through a sophisticated streaming pipeline that normalizes, translates, and enriches SSE data. The engine handles everything from OpenAI to Claude format conversion while managing idle timeouts and usage tracking. This guide maps the exact file locations and architecture patterns used in the `diegosouzapw/OmniRoute` repository.

## Core Streaming Engine Implementation

The heart of OmniRoute's streaming functionality lives in a single factory function that constructs a standards-compliant **TransformStream**.

### Primary Entry Point: `createSSEStream`

According to the OmniRoute source code, the definitive entry point for SSE handling is:

- **[[`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts)](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts)** – Contains the `createSSEStream` factory

This function returns a `TransformStream` instance that intercepts `Uint8Array` chunks from upstream LLM providers, processes them through normalization and translation layers, and emits standardized SSE lines to the client. The implementation supports two distinct operational modes selected via the `mode` parameter:

- **`STREAM_MODE.TRANSLATE`** – Full format conversion (e.g., OpenAI → Claude)
- **`STREAM_MODE.PASSTHROUGH`** – Forward upstream payload unchanged while normalizing events and extracting usage metrics

## Helper Modules and Architecture

The streaming engine delegates specific responsibilities to six specialized utility modules within the `open-sse/utils/` directory:

| Component | Purpose | Source File |
|-----------|---------|-------------|
| **SSE Parsing** | Parses raw `data:` lines, normalizes multi-line events, formats SSE output, and injects synthetic Claude events | [[`open-sse/utils/streamHelpers.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamHelpers.ts)](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamHelpers.ts) |
| **Payload Collection** | Gathers structured payloads for provider-side and client-side logging | [[`open-sse/utils/streamPayloadCollector.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamPayloadCollector.ts)](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamPayloadCollector.ts) |
| **Readiness Policies** | Defines idle-timeout and body-read timeout policies that abort hung upstream streams | [[`open-sse/utils/streamReadinessPolicy.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamReadinessPolicy.ts)](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamReadinessPolicy.ts) |
| **Policy Enforcement** | Runs the policy checks during active stream processing | [[`open-sse/utils/streamReadiness.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamReadiness.ts)](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamReadiness.ts) |
| **Usage Tracking** | Extracts token usage from SSE chunks and estimates usage when providers omit it | [[`open-sse/utils/usageTracking.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/usageTracking.ts)](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/usageTracking.ts) |
| **Response Sanitization** | Removes unsafe content from OpenAI-style chunks before forwarding to the client | [[`open-sse/handlers/responseSanitizer.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responseSanitizer.ts)](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responseSanitizer.ts) |

These modules are imported and orchestrated by the main `createSSEStream` implementation to handle the complete lifecycle of an SSE connection.

## How the Streaming Engine Works

The engine processes upstream LLM responses through a six-stage pipeline implemented within the `TransformStream` constructor.

### Stage 1: Stream Creation

A handler (such as [`open-sse/handlers/chatCore.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chatCore.ts)) invokes `createSSEStream({ ... })` with configuration options including `mode`, `sourceFormat`, `targetFormat`, and callback functions.

### Stage 2: Mode Selection

The `mode` parameter determines the processing strategy:
- **TRANSLATE mode** triggers full format conversion using translators in `open-sse/translator/`
- **PASSTHROUGH mode** forwards the upstream payload while still normalizing keep-alive events

### Stage 3: Chunk Processing

Incoming `Uint8Array` chunks are decoded and split into lines. The `multilineSseDataLineNormalizer` from [`streamHelpers.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamHelpers.ts) handles edge cases like multi-line SSE events and malformed data lines.

### Stage 4: Event Handling

Depending on the mode, the engine either:
- **Translates** the payload using format-specific translators, performing tool-call detection, usage estimation, and synthetic Claude empty-response injection
- **Passthroughs** the payload, extracting tool calls from textual content and aggregating usage for logging

### Stage 5: Idle-Timeout Watchdog

A timer configured via `STREAM_IDLE_TIMEOUT_MS` monitors data flow. If no data arrives for the configured period, the engine aborts the stream and emits a `StreamIdleTimeoutError`.

### Stage 6: Finalization

Upon stream completion, the engine invokes the `onComplete` callback with final status, usage statistics, and cost calculations. On error, it triggers `onFailure` if supplied.

## Implementation Examples

### Creating a Translate-Mode SSE Stream

Use this pattern when converting between provider formats, such as OpenAI to Claude:

```typescript
import { createSSEStream } from '@/open-sse/utils/stream.ts';
import { FORMATS } from '@/open-sse/translator/formats.ts';

export async function POST(req: Request) {
  const body = await req.json();

  const upstreamResp = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: { Authorization: `Bearer ${process.env.OPENAI_API_KEY}` },
    body: JSON.stringify(body),
  });

  const transformed = upstreamResp.body!.pipeThrough(
    createSSEStream({
      mode: 'translate',
      sourceFormat: FORMATS.OPENAI,
      targetFormat: FORMATS.CLAUDE,
      provider: 'openai',
      model: body.model,
      connectionId: body.connectionId,
      body,
      onComplete: (payload) => {
        console.log('Stream finished', payload);
      },
      onFailure: (err) => {
        console.error('Stream error', err);
      },
    })
  );

  return new Response(transformed, {
    status: upstreamResp.status,
    headers: { 'Content-Type': 'text/event-stream' },
  });
}

```

Key implementation details:
- `mode: 'translate'` triggers full format conversion
- `sourceFormat` and `targetFormat` specify the translation direction
- `onComplete` receives final usage and cost data calculated internally by the engine

### Using Passthrough Mode for Native Provider Formats

Use this when the upstream provider already returns the desired format:

```typescript
import { createSSEStream } from '@/open-sse/utils/stream.ts';
import { FORMATS } from '@/open-sse/translator/formats.ts';

export async function POST(req: Request) {
  const body = await req.json();

  const upstreamResp = await fetch('https://api.anthropic.com/v1/messages', {
    method: 'POST',
    headers: { 'x-api-key': process.env.ANTHROPIC_API_KEY },
    body: JSON.stringify(body),
  });

  const transformed = upstreamResp.body!.pipeThrough(
    createSSEStream({
      mode: 'passthrough',
      sourceFormat: FORMATS.CLAUDE,
      clientResponseFormat: FORMATS.OPENAI_RESPONSES,
      provider: 'anthropic',
      model: body.model,
      connectionId: body.connectionId,
      body,
    })
  );

  return new Response(transformed, {
    status: upstreamResp.status,
    headers: { 'Content-Type': 'text/event-stream' },
  });
}

```

In passthrough mode, the engine still extracts usage metrics and normalizes keep-alive events while avoiding unnecessary format conversion overhead.

### Testing Idle Timeout Behavior

Verify the idle timeout protection using a mock stream:

```typescript
import { createSSEStream } from '@/open-sse/utils/stream.ts';
import { FORMATS } from '@/open-sse/translator/formats.ts';

test('idle timeout aborts the stream', async () => {
  const mockUpstream = new ReadableStream({
    start(controller) {
      controller.enqueue(new TextEncoder().encode('data: {"choices":[]}\n\n'));
      // No further data – idle timeout should fire
    },
  });

  const transformed = mockUpstream.pipeThrough(
    createSSEStream({
      mode: 'translate',
      sourceFormat: FORMATS.OPENAI,
      targetFormat: FORMATS.CLAUDE,
      provider: 'openai',
      model: 'gpt-4',
      onFailure: (err) => {
        expect(err.message).toMatch(/idle timeout/i);
      },
    })
  );

  await new Response(transformed).text();
});

```

## Summary

- **Primary location**: The streaming engine code resides in [`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts) as the `createSSEStream` factory function
- **Architecture**: Built on Web Streams API `TransformStream` with modular helpers for parsing, policy enforcement, and usage tracking
- **Processing modes**: Supports `TRANSLATE` for format conversion and `PASSTHROUGH` for lightweight forwarding
- **Key dependencies**: [`streamHelpers.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamHelpers.ts) for SSE normalization, [`streamReadiness.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamReadiness.ts) for timeout handling, and [`usageTracking.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/usageTracking.ts) for token accounting
- **Safety features**: Implements idle timeouts via `STREAM_IDLE_TIMEOUT_MS` and response sanitization through [`responseSanitizer.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/responseSanitizer.ts)

## Frequently Asked Questions

### Where exactly is the main streaming engine file in OmniRoute?

The main streaming engine file is **[`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts)**. This file exports the `createSSEStream` factory function that constructs a `TransformStream` for processing Server-Sent Events. All SSE handling across the application flows through this single entry point.

### What is the difference between TRANSLATE and PASSTHROUGH modes in the streaming engine?

**TRANSLATE mode** performs full format conversion between provider APIs (such as converting OpenAI's response format to Claude's format) using the translator modules in `open-sse/translator/`. **PASSTHROUGH mode** forwards the upstream SSE data unchanged while still normalizing event formatting, extracting usage metrics, and handling idle timeouts. Choose TRANSLATE when bridging different API formats and PASSTHROUGH when the provider already speaks the target format.

### How does OmniRoute handle idle timeouts in the streaming engine?

The engine implements an idle-timeout watchdog configured via the `STREAM_IDLE_TIMEOUT_MS` environment variable. If no data arrives from the upstream provider for the specified duration, the stream aborts and emits a `StreamIdleTimeoutError`. This logic is implemented in the [`streamReadiness.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamReadiness.ts) and [`streamReadinessPolicy.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamReadinessPolicy.ts) modules, which monitor the byte stream during the `transform` phase of the `TransformStream`.

### Which helper modules are required for the streaming engine to function?

The `createSSEStream` function depends on **[`streamHelpers.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamHelpers.ts)** for parsing SSE data lines and formatting output, **[`streamReadiness.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamReadiness.ts)** for timeout enforcement, **[`usageTracking.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/usageTracking.ts)** for token extraction and estimation, and **[`streamPayloadCollector.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/streamPayloadCollector.ts)** for logging aggregation. For OpenAI-specific providers, **[`responseSanitizer.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/responseSanitizer.ts)** in `open-sse/handlers/` provides additional content filtering before forwarding chunks to the client.