FluxReactive

Reusable push streams with subscription-owned cancellation, bounded buffering, and optional ecosystem adapters.

Version

v2.2.0

Size

3.0 KB gzip

Dependencies

Zero dependencies

Runtime support

BrowserNode ≥22SSRDeno

Key exports

streampipeoffromfromEvent View all 25 exports

Works well with

ripple herald pulse courier

Why Flux?

Use Flux when an API pushes many values over time and consumers need independent cancellation. Streams describe reusable work; subscriptions own cleanup. Explicit queue capacity keeps async iteration from silently growing memory.

// Before
const controller = new AbortController();
const render = (value: string) => console.log(value);
const handler = (event: Event) => render((event.target as HTMLInputElement).value);
input.addEventListener('input', handler);
setTimeout(() => controller.abort(), 5_000);

// After
import { fromEvent, map, pipe, takeUntil } from '@vielzeug/flux';

const updates = pipe(
  fromEvent<InputEvent>(input, 'input'),
  map((event) => (event.target as HTMLInputElement).value),
  takeUntil(controller.signal),
);

updates.subscribe({ error: console.error, next: render });

Feature	Flux	RxJS	TC39 Observable
Bundle size	3.0 KB	Varies by imported operators	Native proposal / polyfill
Runtime dependencies
Subscription-owned cancellation
Explicit async queue policy		Operator-dependent	No standard policy
Vielzeug adapters	Ripple, Courier, Herald, Pulse	Manual adapters	Manual adapters

Use Flux when you need a small TypeScript stream primitive, explicit cancellation, and first-party Vielzeug adapters.

Consider RxJS when you need its larger operator catalog or third-party Observable integrations.

Installation

pnpm

pnpm add @vielzeug/flux

npm

npm install @vielzeug/flux

yarn

yarn add @vielzeug/flux

Quick Start

import { toArray, interval, map, pipe, take } from '@vielzeug/flux';

const firstThree = pipe(
  interval({ every: 100 }),
  map((value) => value * 2),
  take(3),
);

try {
  console.log(await toArray(firstThree, { maxItems: 3 })); // [0, 2, 4]
} catch (reason) {
  console.error('Stream failed', reason);
}

Features

• stream() — define cold reusable work with one teardown function
• pipe() — compose any number of typed operators
• Subscription — own cancellation through unsubscribe() or AbortSignal
• createChannel() — mutable multicast state with bounded replay
• toAsyncIterable() — explicit capacity and overflow policy for pull consumers
• retry() — retry failures with optional backoff
• fromSignal() / toSignal() — bridge Ripple signals
• fromQuery() — adapt Courier query state

Documentation

• Usage Guide
• API Reference
• Examples
• Migration Guide

See Also

• Ripple — adapt reactive signal state through @vielzeug/flux/ripple.
• Courier — adapt query snapshots and SSE events through @vielzeug/flux/courier.
• Herald — adapt typed bus events through @vielzeug/flux/herald.
• Pulse — adapt connection and presence events through @vielzeug/flux/pulse.