DXJS

Typed async event streams with explicit buffering and automatic cleanup. Zero runtime dependencies. Consume callbacks and events with for await...of, compose synchronous or asynchronous transforms, and keep standard AsyncIterable<T> at your API boundary.

DXJS is a small event adapter and pipeline library. Use it for application notifications, task progress, worker messages, and callback-based SDKs. For byte streams use Web Streams; for large reactive operator graphs consider RxJS.

Install

npm install dxjs

The API below describes v0.1.0. If npm still lists an earlier version, install the package tarball attached to the v0.1.0 GitHub release. From a checkout, run npm ci and npm test.

ES modules, Node.js 20+ and modern browsers with async iteration and AbortController are supported. The library requires no Node.js APIs. TypeScript projects should include the DOM library because the public API includes typed EventTarget adapters.

Quick start

import { Iterative, filter, map } from 'dxjs';

const progress = new Iterative<{ task: string; percent: number }>();

// Start the consumer before emitting: Iterative is hot and does not replay.
const rendering = (async () => {
  for await (const label of progress.pipe(
    filter((event) => event.task === 'build'),
    map((event) => `Build: ${event.percent}%`),
  )) {
    console.log(label);
  }
})();

progress.next({ task: 'build', percent: 25 });
progress.next({ task: 'build', percent: 100 });
progress.complete();
await rendering; // Both updates arrive, even in the same synchronous burst.

Run npm run example for a task-log producer with a slower async consumer. Its five matching updates arrive in order, including those queued at completion. Ctrl+C cancels both the stream and the simulated rendering operation.

Callback sources

createStream(subscribe, options?) creates a fresh subscription for each iterator. Return a synchronous cleanup function to unsubscribe or stop work. The sink exposes next(value), complete(), error(reason), and closed. next returns whether this consumer accepted the value.

import { createStream } from 'dxjs';

const ticks = createStream<number>((sink) => {
  let count = 0;
  const timer = setInterval(() => {
    sink.next(++count);
    if (count === 10) sink.complete();
  }, 100);
  return () => clearInterval(timer);
});

for await (const count of ticks) {
  console.log(count);
  if (count === 3) break; // Stops the interval.
}

Subscription starts when [Symbol.asyncIterator]() is called (normally at the start of for await), not when the Pipe is created. Synchronous emissions during subscription are buffered. An already aborted signal prevents subscription. If you create an iterator manually, finish it with return() when done.

DOM, WebSocket and Worker events

import { fromEvent, map } from 'dxjs';

const controller = new AbortController();
const clicks = fromEvent(document.body, 'click', {
  signal: controller.signal,
  capacity: 1,
  overflow: 'drop-oldest',
});

try {
  for await (const x of clicks.pipe(map((event) => event.clientX))) {
    console.log(x);
  }
} catch (error) {
  if (!controller.signal.aborted) throw error;
}

// Call controller.abort() from your component/view teardown.

fromEvent(target, type, options?) supports EventTarget, including WebSockets and Workers. Known HTML, document and window event names infer their event types. For other targets, specify a type: fromEvent<MessageEvent<string>>(socket, 'message'). capture and passive are supported. Breaking, aborting, or overflowing removes the listener. This adapter owns the listener only: it does not close a socket, terminate a worker, or infer completion/errors from other event names. Use createStream to connect those lifecycle events when needed.

Buffering and lifecycle

| Option | Default | Behavior | | ------------------------- | --------- | --------------------------------------------------------- | | capacity | 64 | Positive safe integer; maximum queued values per consumer | | overflow | 'error' | Fail that consumer with BufferOverflowError when full | | overflow: 'drop-oldest' | Opt-in | Discard the oldest queued value to retain the new value | | overflow: 'drop-newest' | Opt-in | Reject the arriving value and retain the existing queue | | signal | None | Abort the subscription with signal.reason |

Capacity counts queued values, excluding a value already handed to a reader. For latest-only updates use { capacity: 1, overflow: 'drop-oldest' }. For loss-sensitive events keep the error policy and size the queue for expected bursts. DXJS does not pause an external producer: bounded buffering is not producer backpressure. Use a pull-based async iterable or Web Streams when the producer must wait for the consumer.

• Iterative broadcasts to existing consumers, each with its own queue. Values emitted before subscription are not retained. One slow or cancelled consumer does not affect the others. The constructor's signal applies to all consumers.
• complete() stops the source and lets each consumer drain its queue. Future iterators finish immediately. Repeated completion and later emissions do nothing.
• error(reason) discards queued values and rejects pending/future reads. Future Iterative subscribers also receive the error. The first terminal source action wins.
• Abort discards the consumer's queue and rejects pending reads. It detaches the producer immediately. Once a source completes, its abort listener is removed and queued values drain normally.
• break / iterator return() discards that consumer's queue, resolves pending reads as done, and runs cleanup once. Returning a pipeline also releases its upstream subscription while an async transform is pending.
• Async transforms are sequential. Returning ignores unfinished transform results; it cannot stop arbitrary user promises. Pass the same signal to cancellable work such as fetch. Aborting the source while a transform is running is observed by the pipeline when that transform finishes and it next reads.
• Cleanup is synchronous. Cleanup failures reach the consumer; an existing source or transform error takes precedence. Manual throw(reason) closes a consumer and rejects with that reason.

Operators and interoperability

map(fn) transforms a value and awaits the result before passing it onward. filter(predicate) awaits predicates and supports synchronous TypeScript type guards. Thrown errors and rejected promises propagate through the consumer and close its upstream subscription.

import { from, map, filter } from 'dxjs';

async function* values() {
  yield 1;
  yield 2;
}

const labels = from(values()).pipe(
  map(async (value) => String(value)),
  filter((value) => value !== '1'),
);
// labels is Pipe<string>, and is usable anywhere AsyncIterable<string> is expected.
for await (const label of labels) console.log(label); // '2'

from(asyncIterable) preserves the source's replay, buffering, cancellation and single-use behavior. A generator object remains single-use. DXJS forwards return() but cannot forcibly interrupt a foreign iterator that does not cooperate with cancellation.

A custom PipeFn<Input, Output> receives (value, skip). Return a value or promise; call skip() to omit it. skip() exits by throwing an internal sentinel, so do not catch it inside your operator. Pipelines infer up to five operators per call; chain .pipe(...) calls for longer pipelines. Empty .pipe() is supported.

Development and release

npm ci
npm test        # Build, behavior regressions, and consumer type checks
npm run example # Slow-consumer task-log example
npm pack        # Runs checks and packages dist, README and license

CI covers Node.js 20, 22 and 24. The package is ESM-only and includes declarations and source maps. Build tools are development dependencies only.

Before publishing, choose the next version, update the changelog, run npm pack, and review the resulting package. Publish from the repository root. No automatic publishing workflow is installed.

Migration from the 0.0.2 experiment

The Iterative, .next(value), .complete(), .pipe(), map, and filter entry points remain. Delivery is now buffered per subscriber, with explicit bounded overflow instead of accidental event loss. Add a type argument to new Iterative<T>(); the safe default is now unknown. .next() is a producer method; use [Symbol.asyncIterator]().next() to read. Iterative implements AsyncIterable, not AsyncIterator. Custom operators should return skip() to omit a value. Package entry points now resolve from the repository-root package.

License

MIT