• Basic Usage
• Compose Streams
• Consume Values
• Channels
• Async Iteration and Bounds
• Testing
• Framework Integration
• Working with Other Vielzeug Libraries
• Best Practices

Basic Usage

Define one cold stream. Return teardown work from producer. Every subscription runs producer independently.

import { stream } from '@vielzeug/flux';

const clock = stream<number>((sink) => {
  let value = 0;
  const id = setInterval(() => sink.next(value++), 1_000);

  return () => clearInterval(id);
});

const subscription = clock.subscribe({
  error: console.error,
  next: console.log,
});

subscription.unsubscribe();

Pass AbortSignal when another owner controls lifetime.

const controller = new AbortController();
clock.subscribe(console.log, { signal: controller.signal });
controller.abort();

Compose Streams

Pass source first to pipe(). Operators retain inferred value types across chains.

import { filter, fromEvent, map, pipe, take } from '@vielzeug/flux';

const clicks = pipe(
  fromEvent<MouseEvent>(document, 'click'),
  filter((event) => event.button === 0),
  map((event) => ({ x: event.clientX, y: event.clientY })),
  take(10),
);

clicks.subscribe({
  complete: () => console.log('done'),
  error: console.error,
  next: console.log,
});

Use switchMap() for latest-only work, mergeMap() for concurrent work with bounded concurrency and queue capacity, and concatMap() for ordered work with bounded queue capacity.

import { from, pipe, retry, switchMap } from '@vielzeug/flux';

const results = pipe(
  queries,
  switchMap((query) => from(fetch(`/api/search?q=${encodeURIComponent(query)}`).then((response) => response.json()))),
  retry({ times: 2, delay: (attempt) => 250 * (attempt + 1) }),
);

Consume Values

Use bounded array conversion for finite streams. toArray() rejects once source exceeds maxItems.

import { toArray, of } from '@vielzeug/flux';

try {
  const values = await toArray(of(1, 2, 3), { maxItems: 3 });
  console.log(values);
} catch (reason) {
  console.error('Collection failed', reason);
}

Use first() for first emission and last() for last value before completion. Both reject with FluxEmptyError when the source completes without emitting; pass { defaultValue } to resolve instead. Pass { signal } to cancel waiting; cancellation rejects with signal.reason (the default reason is AbortError).

Channels

Use channels only at imperative boundaries. Expose channel.stream to consumers; keep send() near event producer.

import { createChannel } from '@vielzeug/flux';

const status = createChannel({ initial: 'starting', replay: 1 });
status.stream.subscribe({ error: console.error, next: console.log });
status.send('ready');
status.dispose();

Disposal completes active and future subscribers. Replay retains only configured latest values.

Async Iteration and Bounds

Convert push stream only when pull syntax is required. Capacity and overflow policy are mandatory.

import { interval, toAsyncIterable } from '@vielzeug/flux';

const values = toAsyncIterable(interval(100), {
  capacity: 32,
  overflow: 'error',
});

for await (const value of values) {
  console.log(value);
  if (value === 2) break;
}

return() from a loop permanently completes its iterator; external signal cancellation rejects with signal.reason. Use drop-oldest or drop-newest only when loss is acceptable.

Testing

Use fake timers for time operators. Test producer cleanup through returned subscription.

import { expect, it, vi } from 'vitest';
import { first, pipe, stream, timeout } from '@vielzeug/flux';

it('fails after inactivity', async () => {
  vi.useFakeTimers();
  const result = first(pipe(stream(() => {}), timeout(500)));
  const expectation = expect(result).rejects.toThrow('Timeout after 500ms');

  await vi.advanceTimersByTimeAsync(500);
  await expectation;
  vi.useRealTimers();
});

Framework Integration

React

import { useEffect, useState } from 'react';
import type { Stream } from '@vielzeug/flux';

export function useStream<T>(source: Stream<T>, initial: T): T {
  const [value, setValue] = useState(initial);

  useEffect(() => {
    const subscription = source.subscribe({ error: console.error, next: setValue });
    return () => subscription.unsubscribe();
  }, [source]);

  return value;
}

Vue 3

import { onUnmounted, ref } from 'vue';
import type { Stream } from '@vielzeug/flux';

export function useStream<T>(source: Stream<T>, initial: T) {
  const value = ref(initial);
  const subscription = source.subscribe({ error: console.error, next: (next) => (value.value = next) });

  onUnmounted(() => subscription.unsubscribe());

  return value;
}

Svelte

import type { Stream } from '@vielzeug/flux';

export function streamStore<T>(source: Stream<T>, initial: T) {
  return {
    subscribe(run: (value: T) => void) {
      run(initial);
      const subscription = source.subscribe({ error: console.error, next: run });
      return () => subscription.unsubscribe();
    },
  };
}

Working with Other Vielzeug Libraries

Use fromStore() for snapshot state and fromSubscribe() for callback-delivered event values. Write stream output into the owning state or event layer explicitly.

import { fromStore, fromSubscribe } from '@vielzeug/flux';

const state = fromStore({
  getSnapshot: () => store.value,
  subscribe: (listener) => store.subscribe(listener),
});

const events = fromSubscribe<Message>((listener) => bus.on('message', listener));

Best Practices

• Return one idempotent producer teardown function.
• Pass { signal } from component, request, or task owner.
• Provide error when subscription can recover locally.
• Use pipe(source, ...); never mutate stream definitions.
• Bound mergeMap() concurrency and queue capacity.
• Bound concatMap() queue capacity.
• Bound toArray() with realistic maxItems.
• Choose async iterator overflow policy deliberately.
• Keep timer-backed durations between 0 and 2,147,483,647 milliseconds.
• Keep Channel.send() at integration boundaries.