Pull-based reactive streams built on async generators.

What is streamix?

streamix is essentially a lightweight, modern alternative to RxJS that feels much closer to native JavaScript. It’s a great fit if you're building highly concurrent apps, web-based dashboards, or complex UI tools where you want reactive state but hate the debugging nightmares of push-based subscriber chains.

The core package focuses on:

• Pull-based streams for lazy asynchronous composition
• Subjects for hot multicast event sources
• Familiar operators such as map, filter, switchMap, debounce, and scan
• Async iterator first consumption with for await...of
• query() for awaiting the next stream value with automatic cleanup
• Optional companion packages for coroutines, DOM observers, networking, and aggregate operators

Installation

npm install @epikodelabs/streamix

yarn add @epikodelabs/streamix

pnpm add @epikodelabs/streamix

Quick start

import { range, map, filter, take } from '@epikodelabs/streamix';

const values = range(1, 20).pipe(
  filter(n => n % 2 === 0),
  map(n => n * 10),
  take(5)
);

for await (const value of values) {
  console.log(value); // 20, 40, 60, 80, 100
}

Streams

A stream is an async iterable sequence. You can iterate it directly, subscribe to it, or pass it through an operator pipeline.

import { createStream, debounce, filter, map } from '@epikodelabs/streamix';

async function* searchTerms() {
  yield 'str';
  yield 'stream';
  yield 'streamix';
}

const normalized = createStream('searchTerms', searchTerms).pipe(
  debounce(100),
  map(term => term.trim().toLowerCase()),
  filter(term => term.length >= 3)
);

for await (const term of normalized) {
  console.log(term);
}

Streams are pull-based by default. Work is performed only when values are consumed, which gives predictable cancellation and natural backpressure.

Subjects

Subjects are hot streams. They are useful when the producer is imperative or event-driven and multiple consumers need to observe the same emissions.

import {
  createSubject,
  createBehaviorSubject,
  createReplaySubject,
} from '@epikodelabs/streamix';

const events = createSubject<{ type: string }>();

events.subscribe(event => {
  console.log('event:', event.type);
});

events.next({ type: 'ready' });

Use the subject variants according to subscriber needs:

Factory	Use case
createSubject<T>()	Live event stream with no retained value
createBehaviorSubject<T>(initial)	Live stream with a current value for new subscribers
createReplaySubject<T>(capacity)	Live stream that replays recent emissions to late subscribers

Subjects also support async iteration:

for await (const event of events) {
  console.log(event);
}

Operators

Operators transform streams while preserving async-iterable semantics.

import { from, map, filter, scan, take } from '@epikodelabs/streamix';

const totals = from([1, 2, 3, 4, 5]).pipe(
  filter(n => n % 2 === 1),
  scan((sum, n) => sum + n, 0),
  take(3)
);

Common operators include map, filter, scan, tap, debounce, throttle, switchMap, mergeMap, concatMap, exhaustMap, take, skip, buffer, retry, catchError, share, shareReplay, combineLatest, zip, and forkJoin.

Stream factories

The package includes stream factories for common sources:

Factory	Description
of(...values)	Fixed sequence
from(source)	Arrays, iterables, async iterables, promises
range(start, count)	Sequential integers
interval(ms)	Repeating counter
timer(delay, period?)	Delayed, optionally repeating counter
defer(factory)	Fresh stream per subscriber
merge(...sources)	Interleaved concurrent emissions
concat(...sources)	Sources run sequentially
combineLatest(...sources)	Latest value from each source
zip(...sources)	Pair emissions by index
race(...sources)	First source to emit wins
forkJoin(...sources)	Emit once when all sources complete

Querying a stream

query() awaits the next emitted value and then cleans up the subscription.

import { interval, take } from '@epikodelabs/streamix';

const firstTick = await interval(1000).pipe(take(1)).query();

Coroutines and actors

Use @epikodelabs/streamix/coroutines for worker-backed execution.

import { compute } from '@epikodelabs/streamix/coroutines';

const primes = compute(async function* () {
  let n = 2;

  while (true) {
    while (!isPrime(n)) n++;
    yield n++;
  }
});

The coroutines package provides:

• compute() for worker-backed async generators
• compose() for worker-side pipeline fusion
• actor() for long-lived stateful workers

Companion packages

Package	Description
@epikodelabs/streamix	Core streams, subjects, operators, factories
@epikodelabs/streamix/aggregates	Aggregate operators such as average, min, max, sum
@epikodelabs/streamix/coroutines	Web Worker utilities, coroutines, actors
@epikodelabs/streamix/dom	DOM observation utilities
@epikodelabs/streamix/networking	HTTP client, WebSocket, JSONP

Monorepo structure

projects/libraries/streamix/
├── src/           # Core runtime: streams, subjects, operators, factories
├── aggregates/    # Aggregate operators
├── coroutines/    # Coroutines and actors
├── dom/           # DOM observation utilities
└── networking/    # HTTP client, WebSocket, JSONP

Documentation

• Full Documentation
• Medium: A Generator-Driven, Pull-Based Reactive Core
• Medium: streamix vs redux-saga

Community

• Join GitHub Discussions for questions and ideas.
• Share your feedback

License

GNU AGPL v3 or later

API Reference

Check the detailed API Reference here.