@nexussdk/stream

Resilient AI/LLM streaming pipeline shield for the Nexus SDK ecosystem.
Handles SSE parsing, OpenAI-compatible format, disconnect recovery, Markdown buffering, and heartbeat detection.
Zero dependencies · ~1.0 KB gzipped · Works in any browser or Node.js 18+.

The Problem

Standard fetch streaming breaks in production:

| Problem | Without @nexussdk/stream | With @nexussdk/stream | |---------|---------------------------|-------------------------| | Network drop mid-stream | Blank screen, lost content | Preserved text, graceful error | | Frozen LLM provider | Hangs forever | Heartbeat timeout fires after N seconds | | Broken Markdown | Half-rendered code fences | Buffered to safe boundary | | SSE boilerplate | 50+ lines per implementation | 5 lines |

Installation

npm install @nexussdk/stream
# or
pnpm add @nexussdk/stream

Quick Start

import { NexusStreamClient } from '@nexussdk/stream';

const client = new NexusStreamClient('https://api.myapp.com/ai/chat', {
  preserveOnDisconnect: true,  // keep text if network drops
  safeMarkdownBuffer: true,    // batch to Markdown boundaries
  heartbeatTimeoutMs: 30000,   // fire error if no data for 30s
});

await client.start(
  // Called for each decoded chunk:
  (chunk, accumulated) => {
    document.getElementById('output').textContent = accumulated;
  },
  // Called on completion or error:
  (state) => {
    if (state.isDone)  console.log('Complete:', state.content);
    if (state.error)   console.error('Error:', state.error.message);
    // state.content is always preserved on disconnect
  },
  // Standard fetch options:
  {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ messages: [{ role: 'user', content: 'Hello!' }] }),
  },
);

// Abort at any time (e.g. user clicks "Stop generating"):
client.abort();

Compatible With

Works out-of-the-box with any SSE streaming endpoint:

| Provider | Format | Supported | |----------|--------|-----------| | OpenAI | choices[0].delta.content | ✅ | | Anthropic | raw SSE text | ✅ | | Google Gemini | raw SSE text | ✅ | | Local Ollama | /api/generate | ✅ | | Custom Go/Node.js | data: text\n\n | ✅ |

Framework Examples

import { useCallback, useRef, useState } from 'react';
import { NexusStreamClient } from '@nexussdk/stream';

export function useAIStream(url: string) {
  const [content, setContent] = useState('');
  const [isStreaming, setIsStreaming] = useState(false);
  const clientRef = useRef<NexusStreamClient | null>(null);

  const startStream = useCallback(async (prompt: string) => {
    clientRef.current?.abort();
    clientRef.current = new NexusStreamClient(url);
    setContent('');
    setIsStreaming(true);

    await clientRef.current.start(
      (_, accumulated) => setContent(accumulated),
      (state) => setIsStreaming(false),
      { method: 'POST', body: JSON.stringify({ prompt }) },
    );
  }, [url]);

  const stopStream = () => { clientRef.current?.abort(); setIsStreaming(false); };

  return { content, isStreaming, startStream, stopStream };
}

import { ref } from 'vue';
import { NexusStreamClient } from '@nexussdk/stream';

export function useAIStream(url: string) {
  const content = ref('');
  const isStreaming = ref(false);
  let client: NexusStreamClient | null = null;

  async function startStream(prompt: string) {
    client?.abort();
    client = new NexusStreamClient(url);
    content.value = '';
    isStreaming.value = true;

    await client.start(
      (_, acc) => { content.value = acc; },
      () => { isStreaming.value = false; },
      { method: 'POST', body: JSON.stringify({ prompt }) },
    );
  }

  function stopStream() { client?.abort(); isStreaming.value = false; }

  return { content, isStreaming, startStream, stopStream };
}

import { Injectable, signal } from '@angular/core';
import { NexusStreamClient } from '@nexussdk/stream';

@Injectable({ providedIn: 'root' })
export class AIStreamService {
  content = signal('');
  isStreaming = signal(false);
  private client: NexusStreamClient | null = null;

  async start(url: string, prompt: string) {
    this.client?.abort();
    this.client = new NexusStreamClient(url);
    this.content.set('');
    this.isStreaming.set(true);

    await this.client.start(
      (_, acc) => this.content.set(acc),
      () => this.isStreaming.set(false),
      { method: 'POST', body: JSON.stringify({ prompt }) },
    );
  }

  stop() { this.client?.abort(); this.isStreaming.set(false); }
}

import { NexusStreamClient } from '@nexussdk/stream';

const output = document.getElementById('output');
const client = new NexusStreamClient('/api/ai/chat');

document.getElementById('send').addEventListener('click', async () => {
  await client.start(
    (_, accumulated) => { output.textContent = accumulated; },
    (state) => { if (state.error) output.textContent = `Error: ${state.error.message}`; },
    { method: 'POST', body: JSON.stringify({ prompt: document.getElementById('input').value }) },
  );
});

document.getElementById('stop').addEventListener('click', () => client.abort());

Configuration

interface StreamOptions {
  preserveOnDisconnect?: boolean;  // default: true  — keep text on network drop
  safeMarkdownBuffer?: boolean;    // default: true  — batch to Markdown boundaries
  heartbeatTimeoutMs?: number;     // default: 30000 — timeout if no data received
}

License

MIT © Hồ Huỳnh Dũng