How to Test the OmniRoute Streaming Engine: A Complete Guide

Test the OmniRoute streaming engine by executing the unit tests in tests/unit/web-cookie-providers-new.test.ts, v1-ws-bridge.test.ts, and zenmux-free-provider.test.ts, which validate SSE chunk forwarding, WebSocket bridging, and error sanitization using the Node.js native test runner.

The streaming engine in the diegosouzapw/OmniRoute repository powers real-time data flow from upstream LLM providers to clients via Server-Sent Events (SSE) and WebSocket connections. Understanding how to test the OmniRoute streaming engine ensures that your proxy infrastructure correctly handles back-pressure, heartbeat intervals, and protocol transformations. This guide walks through the architecture, existing test coverage, and practical patterns for writing your own streaming tests.

Streaming Engine Architecture

The streaming pipeline is modular and resides primarily in the open-sse/utils/ directory. Each component handles a specific concern in the data flow from upstream provider to client.

Core Components

• Request Entry: Next.js API routes in src/app/api/v1/chat/completions/route.ts delegate to executors that return streaming responses.
• Stream Utilities: The stream.ts utility reads from upstream ReadableStream instances and manages back-pressure, while streamHandler.ts wraps responses in a TransformStream for standardized processing.
• Readiness Policies: streamReadinessPolicy.ts guarantees clients receive the first chunk within milliseconds, preventing timeout errors.
• Heartbeat Mechanism: sseHeartbeat.ts injects periodic keep-alive events to maintain connections during slow upstream responses.
• WebSocket Bridge: The /v1/ws endpoint pipes SSE streams into WebSocket connections, tested in v1-ws-bridge.test.ts.

Data Flow

The typical flow follows this path: upstream SSE → streamReadinessPolicy → streamHandler (TransformStream) → optional responsesTransformer → client SSE or WebSocket output.

Existing Test Coverage

The repository includes three critical test files that exercise the streaming engine:

1. web-cookie-providers-new.test.ts: Validates that providers returning Content-Type: text/event-stream are streamed verbatim to the client, preserving chunk order and the stream: true flag.
2. v1-ws-bridge.test.ts: Tests the WebSocket bridge that forwards SSE chunks to WS clients, verifying protocol translation and error recovery.
3. zenmux-free-provider.test.ts: Ensures upstream error responses (e.g., HTTP 401) are sanitized before reaching the client, preventing raw HTML or stack trace leaks.

Running the Test Suite

Execute the full streaming test suite using the Node.js native test runner or the project's npm scripts.

Standard execution:

npm run test:all

Targeted execution for streaming components only:

node --import tsx/esm --test tests/unit/web-cookie-providers-new.test.ts
node --import tsx/esm --test tests/unit/v1-ws-bridge.test.ts
node --import tsx/esm --test tests/unit/zenmux-free-provider.test.ts

Writing Custom Stream Tests

When extending functionality, replicate the patterns found in the existing test suite. Below are minimal, runnable examples that mock upstream providers and verify the streamHandler and WebSocket bridge behaviors.

Testing SSE Stream Handling

This example mocks an upstream SSE response and verifies that streamHandler in open-sse/utils/streamHandler.ts correctly forwards chunks without mutation.

import { assert } from "node:assert/strict";

function mockSseStream(chunks: string[]): ReadableStream {
  const encoder = new TextEncoder();
  return new ReadableStream({
    async start(controller) {
      for (const c of chunks) {
        controller.enqueue(encoder.encode(`data: ${c}\n\n`));
        await new Promise((r) => setTimeout(r, 10));
      }
      controller.close();
    },
  });
}

await test("streamHandler forwards upstream SSE chunks", async () => {
  const upstream = new Response(
    mockSseStream([
      JSON.stringify({ type: "stream", token: "Hello " }),
      JSON.stringify({ type: "stream", token: "world" }),
    ]),
    { status: 200, headers: { "Content-Type": "text/event-stream" } }
  );

  const { streamHandler } = await import("../open-sse/utils/streamHandler.ts");
  const transformed = await streamHandler(upstream);
  
  const reader = transformed.body!.getReader();
  const decoder = new TextDecoder();
  const received: string[] = [];
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    received.push(decoder.decode(value));
  }

  assert.equal(received.length, 2);
  assert.match(received[0], /"token":"Hello "/);
  assert.match(received[1], /"token":"world"/);
});

Testing the WebSocket Bridge

Validate that the WebSocket bridge in src/app/api/v1/ws/route.ts correctly translates SSE streams into WebSocket messages.

import { createServer } from "http";
import WebSocket from "ws";
import { assert } from "node:assert/strict";

await test("WebSocket bridge forwards SSE chunks", async () => {
  const upstream = createServer((req, res) => {
    res.writeHead(200, { "Content-Type": "text/event-stream" });
    res.write(`data: {"type":"stream","token":"ping"}\n\n`);
    setTimeout(() => res.end(), 10);
  });
  
  const { port } = await new Promise((r) => 
    upstream.listen(0, () => r(upstream.address()))
  );

  const { handler } = await import("../src/app/api/v1/ws/route.ts");
  const wsServer = new WebSocket.Server({ noServer: true });
  
  wsServer.on("connection", (ws) => {
    handler(ws, `http://127.0.0.1:${port}`);
  });

  const client = new WebSocket(`ws://localhost:${port}`);
  const messages: string[] = [];
  
  client.on("message", (msg) => messages.push(msg.toString()));
  await new Promise((r) => client.on("open", r));
  
  await new Promise((r) => setTimeout(r, 50));
  
  assert.equal(messages.length, 1);
  assert.match(messages[0], /"token":"ping"/);
});

Key Source Files Reference

Monitor these files when modifying streaming behavior according to the diegosouzapw/OmniRoute source code:

• open-sse/utils/stream.ts - Core stream reading and back-pressure logic
• open-sse/utils/streamHandler.ts - TransformStream wrapper for responses
• open-sse/utils/streamReadiness.ts - First-chunk timeout guarantees
• open-sse/utils/streamReadinessPolicy.ts - Readiness enforcement policies
• open-sse/utils/sseHeartbeat.ts - Keep-alive event injection
• src/app/api/v1/ws/route.ts - WebSocket bridge implementation

Summary

• Test the OmniRoute streaming engine using the Node.js native test runner against tests/unit/web-cookie-providers-new.test.ts, v1-ws-bridge.test.ts, and zenmux-free-provider.test.ts.
• Key components reside in open-sse/utils/ and handle stream transformation, readiness policies, and heartbeat management.
• Validation criteria include correct SSE chunk ordering, proper WebSocket forwarding, sanitized error responses, and adherence to readiness timeouts.
• Run commands using npm run test:all or targeted node --import tsx/esm --test invocations for rapid iteration.

Frequently Asked Questions

How do I run only the streaming-related tests?

Execute the specific test files using the Node.js test runner with the --import tsx/esm flag. For example: node --import tsx/esm --test tests/unit/web-cookie-providers-new.test.ts. This avoids running the entire suite while you iterate on streaming logic changes.

What is the purpose of streamReadinessPolicy.ts?

The streamReadinessPolicy.ts file implements the readiness layer that guarantees clients receive the first data chunk within a configured timeout window. It prevents client-side timeouts by monitoring the upstream stream and injecting early responses or heartbeats if the provider is slow to respond.

How does the WebSocket bridge handle SSE streams?

The bridge, implemented in src/app/api/v1/ws/route.ts, consumes the SSE stream from the upstream provider and forwards each chunk as a WebSocket message. It maintains message ordering and handles protocol errors by closing the WebSocket connection gracefully, as validated in v1-ws-bridge.test.ts.

Why are my streaming tests failing with timeout errors?

Timeout failures typically indicate that the streamReadiness.ts heartbeat is not firing or the mock upstream stream is not closing properly. Ensure your mock ReadableStream calls controller.close() after emitting all chunks, and verify that the Content-Type: text/event-stream header is set in the mock Response.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}