# Examples of Using OmniRoute's Streaming Engine: Implementation Guide and Code Samples

> Find examples of using OmniRoute's streaming engine. This guide covers SSE implementation, back-pressure, error handling, and graceful termination with code samples.

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

---

**OmniRoute's streaming engine implements a full-stack Server-Sent Events (SSE) pipeline that wraps upstream LLM responses into disconnect-aware streams, handling back-pressure, error conversion, and graceful termination through utilities like `createDisconnectAwareStream` and `finalizeReadableStream`.**

The `diegosouzapw/OmniRoute` repository provides a Next.js-based routing layer for LLM APIs that supports real-time streaming via Server-Sent Events. This guide provides concrete examples of using OmniRoute's streaming engine, from low-level stream utilities to CLI commands, based on the actual source code implementation in the project.

## Core Components of the Streaming Pipeline

OmniRoute's streaming architecture processes requests through a coordinated pipeline of handlers and utilities that transform provider responses into compliant SSE streams.

### API Entry Points and Route Handlers

Streaming requests enter through Next.js API routes. In [`src/app/api/v1/relay/chat/completions/route.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/app/api/v1/relay/chat/completions/route.ts), requests are received and passed to the stream finalization layer. The route handler acts as the public interface, accepting client connections and preparing the SSE response headers.

### The Stream Finalization Layer

The [`src/app/api/v1/relay/chat/completions/streamFinalizer.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/app/api/v1/relay/chat/completions/streamFinalizer.ts) module wraps upstream responses in a `ReadableStream` and attaches a finalizer that emits the `[DONE]` marker. This component handles abort signals and disconnect logic, ensuring that every stream terminates cleanly regardless of upstream errors.

### Core Chat Handler and Routing

In [`open-sse/handlers/chatCore.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chatCore.ts), the pipeline forwards requests to the combo router, which dispatches to provider-specific executors. This handler coordinates between the incoming request format and the appropriate backend streaming implementation.

## Streaming Utilities and Helper Functions

The `open-sse/utils/` directory contains the low-level engine components that manage stream lifecycle events.

### Disconnect-Aware Stream Wrappers

The `createDisconnectAwareStream` function in [`open-sse/utils/streamHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamHandler.ts) converts upstream errors or client disconnects into SSE-compatible error chunks. It guarantees that a `[DONE]` termination marker is always sent, even when the upstream connection fails.

The companion `createStreamController` utility (same file) supplies callback hooks including `onError` and `onDisconnect` that enable telemetry collection and resource cleanup.

### Stream Readiness and Payload Collection

[`open-sse/utils/streamReadinessPolicy.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamReadinessPolicy.ts) implements `shouldStream` logic that determines whether a request should be streamed based on provider capabilities and client flags. Meanwhile, [`open-sse/utils/streamPayloadCollector.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamPayloadCollector.ts) buffers partial chunks, updates usage statistics, and forwards aggregated payloads downstream.

## Code Examples: Implementing OmniRoute Streaming

These examples demonstrate how to use OmniRoute's streaming engine in different contexts, derived from the test suite and source implementations.

### Example 1: Wrapping an Upstream ReadableStream for SSE Output

This pattern from [`tests/unit/stream-handler.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/tests/unit/stream-handler.test.ts) shows how to wrap a provider response using the core streaming utilities:

```typescript
import { createDisconnectAwareStream, createStreamController } from
  '../../open-sse/utils/streamHandler.ts';
import { finalizeReadableStream } from
  '../../src/app/api/v1/relay/chat/completions/streamFinalizer.ts';

// Mock upstream transform stream (e.g., the result of a provider executor)
const upstream = {
  readable: new ReadableStream({
    start(ctrl) {
      // Simulate a normal streaming response
      ctrl.enqueue(new TextEncoder().encode('data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n'));
      // End of stream
      ctrl.enqueue(new TextEncoder().encode('data: [DONE]\n\n'));
      ctrl.close();
    },
  }),
  writable: { getWriter: () => ({ abort() {} }) },
};

// Attach disconnect-aware wrapper
const sseStream = createDisconnectAwareStream(
  upstream,
  createStreamController({
    // Optional hooks for logging/metrics
    onError: (err) => console.error('SSE error:', err),
    onDisconnect: () => console.warn('Client disconnected')
  })
);

// The finalizer adds the finishing event and makes the stream compatible with Next.js
export const response = new Response(finalizeReadableStream(sseStream), {
  headers: { 'Content-Type': 'text/event-stream; charset=utf-8' },
});

```

### Example 2: Streaming from the Command Line

OmniRoute exposes streaming functionality through its CLI. As documented in [`skills/cli-chat/SKILL.md`](https://github.com/diegosouzapw/OmniRoute/blob/main/skills/cli-chat/SKILL.md), you can invoke the streaming engine directly:

```bash

# Run the built-in CLI command that streams a prompt to the default provider

omniroute stream "Explain quantum entanglement"

```

Programmatically, you can spawn this process to consume the SSE stream:

```typescript
import { spawn } from 'node:child_process';

const proc = spawn('omniroute', ['stream', 'Explain quantum entanglement']);
proc.stdout.on('data', (chunk) => {
  // Each chunk is an SSE line; strip the SSE prefix for pretty printing
  const line = chunk.toString().replace(/^data: /, '');
  console.log('🧩', line);
});
proc.stderr.on('data', (err) => console.error('❗', err.toString()));

```

### Example 3: Server-Side Stream Readiness Checks

Before initiating a stream, check if the configuration supports it using the readiness policy:

```typescript
import { shouldStream } from '../../open-sse/utils/streamReadinessPolicy.ts';
import { FORMATS } from '../../open-sse/translator/formats.ts';

function handleChat(req) {
  const format = req.headers['accept'] ?? FORMATS.OPENAI_CHAT; // default
  const canStream = shouldStream({ 
    format, 
    clientWantsStream: req.query.stream === 'true' 
  });
  
  if (canStream) {
    // Route through streaming pipeline using createDisconnectAwareStream
  } else {
    // Fall back to non-streaming JSON response
  }
}

```

## Error Handling and Disconnection Management

The [`tests/unit/stream-handler.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/tests/unit/stream-handler.test.ts) file demonstrates how the streaming engine handles edge cases:

- **Basic disconnect handling**: When a mock upstream stream errors, `createDisconnectAwareStream` ensures the client receives an SSE error object containing `finish_reason:"error"` followed by a `[DONE]` marker.
- **Post-completion error suppression**: Any error occurring after the `[DONE]` token is ignored, ensuring clients see a clean termination.
- **Provider-specific error conversion**: The test suite shows conversion of HTTP 503 errors (such as Gemini high-demand errors) into SSE error chunks while preserving the original error message in the payload.

These behaviors ensure that client applications receive predictable stream termination even when upstream providers fail.

## Summary

- **OmniRoute's streaming engine** constructs SSE pipelines through [`src/app/api/v1/relay/chat/completions/streamFinalizer.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/app/api/v1/relay/chat/completions/streamFinalizer.ts) and [`open-sse/handlers/chatCore.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chatCore.ts).
- **Stream utilities** in [`open-sse/utils/streamHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamHandler.ts) provide `createDisconnectAwareStream` for error conversion and `createStreamController` for lifecycle hooks.
- **Client scenarios** include wrapping upstream ReadableStreams, invoking streams via the `omniroute stream` CLI command, and checking stream readiness with `shouldStream`.
- **Error resilience** is guaranteed through automatic `[DONE]` marker injection and graceful handling of upstream disconnections.

## Frequently Asked Questions

### How does OmniRoute handle client disconnections during streaming?

When a client disconnects, the `createDisconnectAwareStream` utility in [`open-sse/utils/streamHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/streamHandler.ts) detects the broken connection through the abort signal. It invokes the `onDisconnect` callback for cleanup and ensures the stream terminates with a `[DONE]` marker, preventing resource leaks.

### Can I use OmniRoute's streaming engine without the Next.js API routes?

Yes. While the repository provides Next.js route handlers in [`src/app/api/v1/relay/chat/completions/route.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/app/api/v1/relay/chat/completions/route.ts), the core streaming utilities are framework-agnostic. You can import `createDisconnectAwareStream` and `finalizeReadableStream` into any Node.js or edge runtime environment to build custom SSE endpoints.

### What error format does OmniRoute use for streaming failures?

The engine converts upstream HTTP errors into SSE-compatible JSON chunks containing `finish_reason:"error"` and preserves the original message. For example, a 503 error from a provider becomes a data chunk with error details, followed by the standard `[DONE]` termination marker, as demonstrated in [`tests/unit/stream-handler.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/tests/unit/stream-handler.test.ts).

### Where can I find additional runnable examples of the streaming utilities?

The [`tests/unit/stream-handler.test.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/tests/unit/stream-handler.test.ts) file contains executable test cases that demonstrate disconnect handling, error conversion, and graceful completion. Additionally, [`src/lib/a2a/streaming.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/streaming.ts) implements a JSON-RPC over SSE endpoint that shows production usage of the streaming pipeline for A2A (Agent-to-Agent) communication.