@substrate-system/stream

Use the native browser streams API, but with a nicer wrapper.

Install
Examples
API
- from
- Stream
- through
- transform
- filter
- collect
- run
Comparison with native TransformStream API
- Native API
- @substrate-system/stream API
Advanced: Custom Transformer
Real-World Example
Backpressure Example
Modules

Install

npm i -S @substrate-system/stream

Examples

Simple Transform Chain

import { from, through, collect } from '@substrate-system/stream';

// Create a pipeline that transforms numbers
const pipeline = from([1, 2, 3, 4, 5])
  .pipe(through(x => x * 2))           // double each number
  .pipe(through(x => x + 1))           // add 1
  .pipe(through(x => `Number: ${x}`)); // convert to string

const result = await collect(pipeline);

console.log(result);
// ['Number: 3', 'Number: 5', 'Number: 7', 'Number: 9', 'Number: 11']

Text Processing

import { Stream, through } from '@substrate-system/stream';

// Fetch and process text line by line
const response = await fetch('data.txt');

const pipeline = Stream(response.body.pipeThrough(new TextDecoderStream()))
  .pipe(through(text => text.split('\n')))
  .pipe(through(lines => lines.map(line => line.trim())))
  .pipe(through(lines => lines.map(line => line.toUpperCase())));

await pipeline.pipeTo(new WritableStream({
  write(lines) {
    lines.forEach(line => console.log(line));
  }
}));

JSON Processing

import { from, through, transform, collect } from '@substrate-system/stream';

const users = [
  { name: 'Alice', age: 30 },
  { name: 'Bob', age: 25 },
  { name: 'Charlie', age: 35 }
];

// Filter adults and extract names
const filterAdults = transform<typeof users[0], string>({
  transform(user, controller) {
    if (user.age >= 30) {
      controller.enqueue(user.name);
    }
  }
});

const pipeline = from(users).pipe(filterAdults);

const adults = await collect(pipeline);
console.log(adults);
// ['Alice', 'Charlie']

File Processing with Flush

The through function takes a second callback, or "flush" function, that gets called once when the stream ends. The callback takes any remainig items, ie, any items at the end that do not fill a complete batch.

import { through, from, collect } from '@substrate-system/stream';

// Batch processing with flush
let batch:string[] = [];

const batcher = through<string, string[]>(
  (item) => {
    batch.push(item);
    if (batch.length >= 3) {
      const result = [...batch];
      batch = [];
      return result;
    }
    return []; // Don't emit yet
  },
  () => {
    // Flush remaining items
    if (batch.length > 0) {
      console.log('Flushing remaining:', batch);
    }
  }
);

const pipeline = from(['a', 'b', 'c', 'd', 'e']).pipe(batcher);
const batches = await collect(pipeline);

API

from

Create a readable stream from an array or iterable (sync or async).

function from<T> (iterable:Iterable<T>|AsyncIterable<T>):PipeableStream<T, never>

import { from, collect } from '@substrate-system/stream';

const pipeline = from([1, 2, 3, 4, 5]);
const result = await collect(pipeline);
// [1, 2, 3, 4, 5]

// Also works with async iterables
async function* generator() {
  yield 1;
  yield 2;
  yield 3;
}

const asyncPipeline = from(generator());
const asyncResult = await collect(asyncPipeline);
// [1, 2, 3]

Stream

Take a native ReadableStream and wrap it in our API.

Add a .pipe method

function Stream<R> (readable:ReadableStream<R>):PipeableStream<R, never>

import { Stream, through, collect } from '@substrate-system/stream';

const response = await fetch('data.txt');
const pipeline = Stream(response.body)
  .pipe(through(chunk => new TextDecoder().decode(chunk)));

const result = await collect(pipeline);

through

Create a simple transform that applies a function to each chunk.

function through<I, O> (
    transformFn:(chunk:I)=>O|Promise<O>,
    flushFn?:()=>void|Promise<void>
):PipeableStream<O, I>

import { from, through, collect } from '@substrate-system/stream';

const pipeline = from([1, 2, 3])
  .pipe(through(x => x * 2))
  .pipe(through(x => x + 1));

const result = await collect(pipeline);
// [3, 5, 7]

// Optional flush callback runs when stream closes
const withFlush = through(
  (x) => x * 2,
  () => console.log('Stream finished!')
);

transform

Create a custom transform with full control over the TransformStream API.

function transform<I, O> (transformer:Transformer<I, O>):PipeableStream<O, I>

import { from, transform, collect } from '@substrate-system/stream';

// One-to-many: emit multiple values per input
const splitter = transform<string, string>({
  transform(chunk, controller) {
    for (const char of chunk) {
      controller.enqueue(char);
    }
  }
});

const pipeline = from(['hello', 'world']).pipe(splitter);
const result = await collect(pipeline);
// ['h', 'e', 'l', 'l', 'o', 'w', 'o', 'r', 'l', 'd']

// With flush callback
const withFlush = transform<number, number>({
  transform(chunk, controller) {
    controller.enqueue(chunk * 2);
  },
  flush(controller) {
    controller.enqueue(999); // Emit final value
  }
});

filter

Filter stream values based on a predicate function.

function filter<T> (
  predicate:(item:T) => boolean|Promise<boolean>
):PipeableStream<T, T>

import { from, filter, collect } from '@substrate-system/stream';

const pipeline = from([1, 2, 3, 4, 5])
  .pipe(filter(x => x > 2));

const result = await collect(pipeline);
// [3, 4, 5]

// Async predicates work too
const asyncFilter = filter(async (x: number) => {
  const shouldKeep = await someAsyncCheck(x);
  return shouldKeep;
});

collect

Collect all values from a stream into an array.

function collect<T> (stream:PipeableStream<T, any>):Promise<T[]>

import { from, through, collect } from '@substrate-system/stream';

const pipeline = from([1, 2, 3])
  .pipe(through(x => x * 2));

const result = await collect(pipeline);
// [2, 4, 6]

run

Execute a stream pipeline without collecting results (for side effects only).

function run<T> (stream:PipeableStream<T, any>):Promise<void>

import { from, through, run } from '@substrate-system/stream';

const pipeline = from([1, 2, 3])
  .pipe(through(x => {
    console.log(x);
    return x;
  }));

await run(pipeline);
// Logs: 1, 2, 3
// Returns: void

Comparison with native `TransformStream` API

Native API

// Native TransformStream API
const response = await fetch('data.json');

const decoder = new TextDecoderStream();
const parseJson = new TransformStream({
  transform(chunk, controller) {
    try {
      const parsed = JSON.parse(chunk);
      controller.enqueue(parsed);
    } catch (e) {
      controller.error(e);
    }
  }
});

const filterTransform = new TransformStream({
  transform(chunk, controller) {
    if (chunk.active) {
      controller.enqueue(chunk);
    }
  }
});

// pipe
const stream1 = response.body.pipeThrough(decoder);
const stream2 = stream1.pipeThrough(parseJson);
const stream3 = stream2.pipeThrough(filterTransform);

const reader = stream3.getReader();
const results = [];
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  results.push(value);
}