Streaming

rezo.stream() returns a RezoStreamResponse immediately and emits data chunks as they arrive. This is ideal for large responses, server-sent events, or any scenario where you want to process data incrementally without buffering the entire body in memory.

Quick Start

import { Rezo } from 'rezo';

const rezo = new Rezo();

const stream = rezo.stream('https://example.com/large-file.json');

stream.on('data', (chunk) => {
  console.log('Received:', chunk.length, 'bytes');
});

stream.on('finish', (info) => {
  console.log('Complete:', info.status, info.contentLength, 'bytes');
});

stream.on('error', (err) => {
  console.error('Stream error:', err.message);
});

How It Works

rezo.stream() is non-blocking. It initiates the HTTP request in the background and returns a StreamResponse event emitter immediately. As the response arrives, events fire in sequence:

initiated — request has been created
start — request is being sent
headers — response headers received (first byte)
status — HTTP status code received
cookies — cookies from Set-Cookie headers
data — body chunks (emitted multiple times)
progress — transfer progress updates
redirect — if a redirect was followed (before headers of final response)
finish / done — response fully received
error — if an error occurs at any point

Events Reference

initiated

Emitted when the request object has been created, before any network I/O:

stream.on('initiated', () => {
  console.log('Request created');
});

start

Emitted when the request is being sent to the server:

stream.on('start', (info) => {
  console.log('Requesting:', info.method, info.url);
  console.log('Request headers:', info.headers);
  console.log('Timeout:', info.timeout);
});

The info object includes url, method, headers, timestamp, timeout, and maxRedirects.

headers

Emitted when response headers are received (time to first byte):

stream.on('headers', (info) => {
  console.log('Status:', info.status, info.statusText);
  console.log('Content-Type:', info.contentType);
  console.log('Content-Length:', info.contentLength);
  console.log('TTFB:', info.timing.firstByte, 'ms');
});

The info object includes status, statusText, headers, contentType, contentLength, cookies, and timing with firstByte and total in milliseconds.

status

Emitted with the HTTP status code and status text:

stream.on('status', (status, statusText) => {
  console.log(`HTTP ${status} ${statusText}`);
});

cookies

Emitted with an array of Cookie objects from the response:

stream.on('cookies', (cookies) => {
  cookies.forEach(cookie => {
    console.log(`Cookie: ${cookie.key}=${cookie.value}`);
  });
});

data

Emitted for each chunk of the response body. Chunks are Uint8Array or string depending on encoding:

const chunks = [];

stream.on('data', (chunk) => {
  chunks.push(chunk);
});

stream.on('finish', () => {
  const body = Buffer.concat(chunks).toString('utf-8');
  console.log('Full body:', body);
});

progress

Emitted periodically during data transfer:

stream.on('progress', (progress) => {
  console.log(`${progress.percentage.toFixed(1)}% complete`);
  console.log(`${progress.loaded} / ${progress.total} bytes`);
  console.log(`Speed: ${(progress.speed / 1024).toFixed(1)} KB/s`);
  console.log(`ETA: ${(progress.estimatedTime / 1000).toFixed(1)}s`);
});

The progress object includes loaded, total, percentage (0—100), speed (bytes/sec), averageSpeed, estimatedTime (ms remaining), and timestamp.

redirect

Emitted when a redirect is followed:

stream.on('redirect', (info) => {
  console.log(`Redirect #${info.redirectCount}: ${info.sourceUrl} -> ${info.destinationUrl}`);
  console.log(`Status: ${info.sourceStatus} ${info.sourceStatusText}`);
  console.log(`Duration: ${info.duration}ms`);
});

The info object includes sourceUrl, sourceStatus, sourceStatusText, destinationUrl, redirectCount, maxRedirects, headers, cookies, method, timestamp, and duration.

finish / done

Both events fire when the response is fully received. They carry the same payload:

stream.on('finish', (info) => {
  console.log('Final URL:', info.finalUrl);
  console.log('Status:', info.status, info.statusText);
  console.log('Content-Length:', info.contentLength, 'bytes');
  console.log('Total time:', info.timing.total, 'ms');
  console.log('TTFB:', info.timing.firstByte, 'ms');
  console.log('Download time:', info.timing.download, 'ms');
  console.log('Cookies:', info.cookies.string);
  console.log('All URLs traversed:', info.urls);
});

done is an alias for finish — use whichever reads better in your code.

error

Emitted if the request fails at any point:

stream.on('error', (err) => {
  console.error('Error code:', err.code);
  console.error('Message:', err.message);
});

The error is a RezoError with code, message, and config properties.

close

Emitted when the stream is fully closed:

stream.on('close', () => {
  console.log('Stream closed');
});

Consumption Patterns

Collecting the Full Body

const stream = rezo.stream('https://api.example.com/large-dataset');

const chunks: Uint8Array[] = [];

stream.on('data', (chunk) => {
  if (chunk instanceof Uint8Array) {
    chunks.push(chunk);
  } else {
    chunks.push(Buffer.from(chunk));
  }
});

stream.on('finish', () => {
  const body = Buffer.concat(chunks).toString('utf-8');
  const data = JSON.parse(body);
  console.log('Records:', data.length);
});

Line-by-Line Processing

const stream = rezo.stream('https://api.example.com/ndjson-feed');

let buffer = '';

stream.on('data', (chunk) => {
  buffer += typeof chunk === 'string' ? chunk : new TextDecoder().decode(chunk);
  const lines = buffer.split('
');
  buffer = lines.pop() || ''; // Keep incomplete line in buffer

  for (const line of lines) {
    if (line.trim()) {
      const record = JSON.parse(line);
      processRecord(record);
    }
  }
});

Progress Bar

const stream = rezo.stream('https://releases.example.com/app-v2.tar.gz');

stream.on('headers', (info) => {
  console.log(`Downloading: ${(info.contentLength / 1024 / 1024).toFixed(1)} MB`);
});

stream.on('progress', (p) => {
  const bar = '='.repeat(Math.floor(p.percentage / 2)).padEnd(50);
  const speed = (p.speed / 1024 / 1024).toFixed(1);
  process.stdout.write(`
[${bar}] ${p.percentage.toFixed(0)}% ${speed} MB/s`);
});

stream.on('finish', () => {
  console.log('
Done!');
});

With Request Options

const stream = rezo.stream('https://api.example.com/events', {
  headers: {
    'Accept': 'text/event-stream',
    'Authorization': 'Bearer token123'
  },
  timeout: 0 // No timeout for long-lived streams
});

Encoding

Set the encoding for string-mode data chunks:

const stream = rezo.stream('https://example.com/text');
stream.setEncoding('utf-8');

stream.on('data', (chunk) => {
  // chunk is a string when encoding is set
  console.log(chunk);
});

Checking State

stream.isFinished(); // true after finish/done event