saga-custom-events

TypeScript orchestration inspired by redux-saga, using native EventTarget and CustomEvent instead of a Redux store.

The library is ESM-only and works in modern browsers and Node.js 18+.

Install

npm install saga-custom-events

Quick start

import createSagaRuntime, {
  call,
  cancel,
  put,
  takeLatest,
  type SagaEvent,
  type SagaIterator,
} from "saga-custom-events";

type LoadUser = { id: string };

function* loadUser(event: SagaEvent<LoadUser>): SagaIterator<void> {
  const response: Response = yield call(fetch, `/api/users/${event.payload.id}`);
  const user: unknown = yield call([response, "json"]);
  yield put("USER_LOADED", user);
}

function* rootSaga(): SagaIterator<void> {
  yield takeLatest("LOAD_USER", loadUser);
}

const runtime = createSagaRuntime();
const rootTask = runtime.run(rootSaga);
const offUserLoaded = runtime.bus.on("USER_LOADED", (event) => {
  console.log(event.detail);
});

runtime.emit("LOAD_USER", { id: "42" });

// Later:
rootTask.cancel();
await rootTask.promise;
offUserLoaded();
runtime.close();

A saga receives an envelope:

interface SagaEvent<T> {
  type: string;
  payload: T;
  event: CustomEvent<T>;
}

Both convenience dispatch and native dispatch are supported:

runtime.emit("SAVE", { id: 1 });
runtime.bus.dispatchEvent(new CustomEvent("SAVE", {
  detail: { id: 2 },
}));

Native EventTarget listeners run synchronously. Delivery to sagas is queued in order so a watcher can re-arm between events in a synchronous burst. The first queued event is delivered in a microtask; later events are separated by task turns so the watcher has time to install its next take.

The saga queue is bounded by default:

const runtime = createSagaRuntime({
  busOptions: {
    capacity: 4096,
    overflow: "throw", // "drop-newest" | "drop-oldest"
  },
});

The defaults are exported as DEFAULT_EVENT_QUEUE_CAPACITY, DEFAULT_CHANNEL_CAPACITY and DEFAULT_ACTION_CHANNEL_CAPACITY. throw is the default and fails the dispatch before native listeners see an event that cannot be queued. The dropping policies keep dispatch non-throwing and expose the number of discarded events through diagnostics. If bus and busOptions are both supplied, the already-created bus takes precedence.

Effects

Import effects either from the package root or from saga-custom-events/effects.

• Events: take, takeMaybe, put, putResolve, actionChannel, flush
• Functions: call, apply, cps
• Tasks: fork, spawn, detach, join, cancel, cancelled
• Composition: all, race, delay
• State and context: select, getContext, setContext
• Watchers: takeEvery, takeLatest, takeLeading, throttle, debounce
• Resilience: retry

The watcher helpers return fork effects, so standard saga syntax works:

function* rootSaga() {
  yield takeEvery("AUDIT", auditWorker);
  yield takeLatest("SEARCH", searchWorker);
}

Generator, async, Promise-returning and synchronous workers are supported. Async generators are rejected explicitly.

put also accepts a channel:

const jobs = channel<number>();

function* example() {
  yield put(jobs, 10);
  const value: number = yield take(jobs);
}

Because native EventTarget.dispatchEvent is synchronous, putResolve is an API-compatible alias of put for bus events.

Channels and buffers

Channels:

• channel
• eventChannel
• multicastChannel
• stdChannel

Buffers:

• buffers.none()
• buffers.fixed(size)
• buffers.dropping(size)
• buffers.sliding(size)
• buffers.expanding(initialSize)

channel() uses a fixed buffer of 1024 items by default. Overflow throws rather than growing memory without a bound. Pass a buffer explicitly to choose another policy. This intentionally differs from redux-saga's expanding default:

const latestJobs = channel(buffers.sliding(100));
const losslessButUnbounded = channel(buffers.expanding());

buffers.expanding() is intentionally unbounded. Use it only when producers are controlled or when an external backpressure mechanism exists.

Closing a channel wakes pending takers with END. A regular take terminates its saga and executes finally; takeMaybe returns END to the saga. Closing does not discard buffered payloads: they remain available to take/flush until drained, so a retained closed channel can still retain those payloads.

const stream = eventChannel<number>((emit) => {
  const timer = setInterval(() => emit(Date.now()), 1000);
  return () => clearInterval(timer);
});

stream.close(); // subscription cleanup runs exactly once

Cleanup functions returned by channel.take, multicastChannel.take and runtime.bus.on are idempotent and detach their internal references after use. eventChannel.close() also drops its subscriber and emitter references.

actionChannel ownership and overflow

An actionChannel uses buffers.sliding(1024) by default. This bounds memory under a sustained producer burst and keeps the newest events; old buffered events can therefore be dropped. This is an overflow policy, not producer backpressure. For job queues where loss is unacceptable, choose the behavior explicitly:

function* worker() {
  const jobs = yield actionChannel("JOB", buffers.fixed(500));
  // Or opt into the redux-saga-style unbounded behavior:
  const allJobs = yield actionChannel("ALL_JOBS", buffers.expanding());
}

With buffers.fixed, overflow throws while the bus delivers the event; the runtime closes that actionChannel and reports the error through onError. It does not pause the producer.

Each created actionChannel belongs to the task that created it. It remains open while the generator and its finally blocks run, then closes before the task waits for attached children. Manual close() releases ownership immediately. A channel directly returned by a blocking call/raw nested generator is transferred to the waiting parent; forked and detached ownership is not transferred.

For a channel deliberately handed to a detached task, disable task ownership and close the channel manually:

const runtime = createSagaRuntime({
  autoCloseActionChannels: false,
  actionChannelBuffer: () => buffers.sliding(256),
});

actionChannelBuffer is a factory and is called for every channel. A buffer passed directly to actionChannel(pattern, buffer) has priority over it.

Cancellation

Cancellation propagates through attached forks, blocking calls, all, race and joins. Generator finally blocks are fully drained.

function* worker() {
  try {
    yield delay(60_000);
  } finally {
    if (yield cancelled()) {
      yield call(releaseResources);
    }
  }
}

Promises can expose the public CANCEL hook:

import { CANCEL, type CancelablePromise } from "saga-custom-events";

const controller = new AbortController();
const request = fetch("/api/data", {
  signal: controller.signal,
}) as CancelablePromise<Response>;
request[CANCEL] = () => controller.abort();

Tasks

runtime.run, fork, spawn and watcher helpers create tasks with:

• cancel()
• isRunning()
• isCancelled()
• result()
• error()
• setContext(object)
• toPromise()
• promise

Cancellation resolves promise/toPromise() and result() with the exported TASK_CANCEL symbol, so their TypeScript result is T | typeof TASK_CANCEL.

spawn creates a detached task. Its failure does not abort the parent and is reported through onError. Failure of an attached fork aborts the parent and its sibling tasks.

Runtime options

const runtime = createSagaRuntime({
  busOptions: {
    capacity: 4096,
    overflow: "throw",
  },
  actionChannelBuffer: () => buffers.sliding(1024),
  autoCloseActionChannels: true,
  context: { api },
  getState: () => applicationState,
  onError: (error, { sagaStack }) => {
    console.error(sagaStack, error);
  },
  sagaMonitor,
});

runtime.setContext({ api: replacementApi });

select is optional and reads from getState; Redux itself is not required. For standalone runSaga, pass an explicit bus when callers need to emit or inspect events; createSagaRuntime is the convenient owner of a newly created bus.

runtime.close() stops saga-side event acceptance, drains already queued events, then sends END to bus takers. It returns immediately. It does not cancel tasks blocked on delays, Promises or custom channels, does not cancel detached spawn tasks, and does not remove native EventTarget listeners. Cancel owned root tasks and run listener cleanup functions as part of application shutdown. Events emitted after close still reach native listeners, but are not queued for sagas.

Diagnostics and memory lifecycle

Diagnostics return plain snapshot records and arrays without live tasks, channels, callbacks or contexts:

const diagnostics = runtime.getDiagnostics();

console.log(diagnostics.activeTasks);
console.log(diagnostics.activeActionChannels);
console.log(diagnostics.ownedActionChannels);
console.log(diagnostics.bus.pendingEvents);
console.log(diagnostics.bus.droppedEvents);
console.log(diagnostics.bus.activeTakers);

tasks contains active tasks only. Task totals and dropped-item counters are cumulative. activeActionChannels includes manually managed channels, while ownedActionChannels includes only task-owned channels. bus.activeTakers counts saga-side bus takers, not native EventTarget listeners.

Built-in buffers and channels return InspectableBuffer and InspectableChannel, whose getDiagnostics() method is required. The method remains optional on the base Buffer and Channel interfaces so custom implementations stay structurally compatible. Buffer capacity: null means the buffer is unbounded.

Cancellation detaches the continuation from unresolved Promise/thenable and CPS callbacks before invoking their cleanup hook. A Promise can still expose CANCEL to stop the underlying operation. Completed tasks release their runtime, parent and context references; task results and errors remain available through the public task API.

The bounded defaults prevent accidental queue growth, but deliberately unbounded resources are still possible: buffers.expanding(), an actionChannel configured with it, an infinite spawn, or native addEventListener calls that are never removed. Those resources must have an application-level lifecycle. Diagnostics expose live counts and queue state; the separate GC regression suite verifies reference detachment.

Development

npm ci
npm run typecheck
npm test
npm run test:memory
npm pack --dry-run

The test suite covers events, channels, buffers, helpers, cancellation, error propagation, task lifecycle, bounded bursts, END, packaging types and concurrent effects. test:memory runs separate WeakRef regressions with --expose-gc for externally retained Promise, CPS and unsubscribe callbacks.