Reactive flows built on async generators.

streamix is a reactive flows library built on async generators. It gives you synchronous state reads, composable derived values, lifecycle-aware scopes, and flows that work naturally with modern TypeScript.

npm install @epikodelabs/streamix

Core concepts

Atoms

Atoms are reactive values: readable, writable, composable with derived, and consumable as async iterables when you need pipelines.

import { atom, derived, iterate, map, pipe, take } from '@epikodelabs/streamix';

const count = atom(0);         // always has a value
const label = atom<string>();  // value arrives later

const summary = derived(() => `count is ${count.value}`);

count.next(5);
console.log(summary.value); // "count is 5"

const doubled = pipe(iterate(count), map(n => n * 2), take(3));
for await (const value of doubled) console.log(value);

Operators

Operators transform flows. Sync and async callbacks are both supported.

pipe(
  iterate(count),
  map(async x => await enrich(x)),
  filter(x => x.valid),
  debounce(200),
  take(10)
)

Full catalog: audit, buffer, bufferCount, bufferUntil, bufferWhile, catchError, concatMap, debounce, defaultIfEmpty, delay, delayUntil, distinctUntilChanged, distinctUntilKeyChanged, endWith, exhaustMap, expand, filter, finalize, first, fork, groupBy, ignoreElements, last, map, mergeMap, observeOn, partition, reduce, sample, scan, select, shareReplay, skip, skipUntil, skipWhile, slidingPair, startWith, switchMap, take, takeUntil, takeWhile, tap, throttle, throwError, toArray, withLatestFrom.

Flow Factories

Factory	Description
combineLatest(...sources)	Latest value from each source, combined
concat(...sources)	Sources run sequentially
defer(factory)	Fresh flow per subscriber
EMPTY()	Completes immediately
forkJoin(...sources)	Emits once when all complete
from(source)	Arrays, iterables, generators, promises
fromEvent(target, event)	DOM / Node events
fromPromise(p)	Promise as a single-emission flow
interval(ms)	Counter every ms milliseconds
merge(...sources)	Interleaved concurrent emissions
of(...values)	Fixed sequence, then complete
race(...sources)	First source to emit wins
range(start, count)	Sequential integers
retry(source, n)	Retry on error, up to n times
timer(delay, period?)	Delayed, optionally repeating
zip(...sources)	Pair emissions by index

Custom operators

import { createOperator, DONE, NEXT } from '@epikodelabs/streamix';

const onlyPrime = () =>
  createOperator<number, number>('onlyPrime', source => ({
    async next() {
      while (true) {
        const result = await source.next();
        if (result.done) return DONE;
        if (isPrime(result.value)) return NEXT(result.value);
      }
    }
  }));

query() - promise from a flow

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

Resolves to the first emitted value and unsubscribes automatically.

---

Coroutines

Offload heavy work to Web Workers without losing composability.

import { actor, compose, compute, coroutine, main } from '@epikodelabs/streamix/coroutines';

// Run a function in a worker pool
const square = coroutine(function square(value: number) {
  return value * value;
});
const result = await square.processTask(7); // 49
await square.finalize();

// Long-lived stateful worker
const counter = actor('counter', (msg: { action: 'inc' | 'get' }, state: number) => {
  if (msg.action === 'inc') return state + 1;
  return state;
}, 0);

const one = await main.outbox.request(counter, 'update', { action: 'inc' }); // 1
const two = await main.outbox.request(counter, 'update', { action: 'inc' }); // 2
await main.outbox.stop(counter);

Worker functions are serialized and run in isolation, so they must be self-contained. streamix APIs are not available inside Web Workers.

---

HTTP client

import { createHttpClient, readJson, useBase, useTimeout } from '@epikodelabs/streamix/networking';

const api = createHttpClient().withDefaults(
  useBase('https://api.example.com'),
  useTimeout(5000)
);

for await (const data of api.get('/items', readJson)) {
  console.log(data);
}

---

Why pull-based?

Most reactive libraries push values eagerly. streamix pulls: the consumer asks for the next value, and only then is it computed.

async function* primes() {
  let n = 2;
  while (true) {
    if (isPrime(n)) yield n;
    n++;
  }
}

// Only 5 primes are ever computed
for await (const p of pipe(primes, take(5))) {
  console.log(p);
}

This gives you on-demand computation, bounded memory, and consumer-driven backpressure without manual coordination.

---

streamix vs RxJS

	streamix	RxJS
Execution model	Pull-based (lazy)	Push-based (eager)
Backpressure	Consumer-driven	Manual patterns required
Async/await	Native	Limited
Bundle size	Small	Larger
Reactive state	Atoms + derived	Manual stores

---

Resources

• Documentation
• npm
• GitHub
• Give feedback

---

License

GNU AGPL v3 or later

API Reference

Check the detailed API Reference here.