Emitter

Signature: tokens → events — a sink

Emitter is a sink that consumes a token stream and re-emits each token as a named event. It ships in two substrate-specific shapes:

• Node — a Writable stream that emits events via EventEmitter. Subscribe with .on(name, fn).
• Web — an EventTarget with a .writable WritableStream attached. Each token is dispatched as a CustomEvent. Subscribe with .addEventListener(name, ev => ev.detail).

Same model — token name as event name, token value as event payload — different runtime APIs. EventTarget and CustomEvent are universal across modern Node, Bun, Deno, and browsers, so the Web shape needs no polyfill.

Introduction

Node

import {chain} from 'stream-chain';
import {parser} from 'stream-json/parser.js';
import emitter from 'stream-json/emitter.js';
import fs from 'node:fs';

const e = emitter();
chain([fs.createReadStream('data.json'), parser.asStream(), e]);

let counter = 0;
e.on('startObject', () => ++counter);
e.on('finish', () => console.log(counter, 'objects'));

Web

import {chain} from 'stream-chain/web';
import {parser} from 'stream-json/web/parser.js';
import emitter from 'stream-json/web/emitter.js';

const e = emitter();
const pipeline = chain([source, parser.asWebStream(), e]);

let counter = 0;
e.addEventListener('startObject', () => ++counter);
e.addEventListener('keyValue', ev => console.log('key:', ev.detail));
await pipeline.readable.pipeTo(e.writable);
console.log(counter, 'objects');

Subscribing happens via addEventListener; event.detail carries the token value (for valued tokens like keyValue, stringValue, numberValue; structural tokens like startObject have event.detail === undefined).

API

emitter(options) — Node

options is an optional object passed to the Writable constructor; see node.js' Stream documentation. No custom options are used.

Returns a new Writable stream.

Subscribe with .on(name, value => …). The value is passed positionally; structural tokens fire with value === undefined.

const emitter = options => {
  const stream = new Writable({
    ...options,
    objectMode: true,
    write(chunk, _, callback) {
      stream.emit(chunk.name, chunk.value);
      callback(null);
    }
  });
  return stream;
};

emitter(options) — Web

Imported from stream-json/web/emitter.js. Returns an EventTarget with a .writable WritableStream attached. Per-token, dispatches new CustomEvent(chunk.name, {detail: chunk.value}).

options.strategy is an optional QueuingStrategy applied to the writable side.

const emitter = options => {
  const target = new EventTarget();
  target.writable = new WritableStream(
    {
      write(chunk) {
        target.dispatchEvent(new CustomEvent(chunk.name, {detail: chunk.value}));
      }
    },
    options?.strategy
  );
  return target;
};

Static methods and properties

emitter.emitter(options)

Alias of the factory function (same on Node and Web entries).

emitter.asStream(options) — Node only

Identity alias for the Node factory, for API consistency with other components.

emitter.asWebStream(options)

On the Node entry (stream-json/emitter.js), this is a delegate to the Web factory, returning the EventTarget shape. On the Web entry it is a self-alias for the factory itself.

Zero-allocation alternative for hot paths

The Web emitter dispatches synchronously per token and allocates a fresh CustomEvent for every token. For very high-throughput streams that overhead matters. The Web Streams substrate already exposes an async-iterable interface, so consumers who don't need the subscribe-style API can drain the parser directly with a for await loop and dispatch by name with a plain object lookup — no event objects, no listener registry:

import {parser} from 'stream-json/web/parser.js';

const handlers = {
  startObject: () => {},
  keyValue: value => {},
  stringValue: value => {},
  numberValue: value => {},
  endObject: () => {}
  // …other tokens as needed
};

const {readable, writable} = parser.asWebStream();
sourceReadable.pipeTo(writable);

for await (const tok of readable) handlers[tok.name]?.(tok.value);

This is the same model emitter exposes — token name selects a handler, value is the payload — but with one function call per token and zero per-token allocations. Use the emitter when API ergonomics matter (multiple independent subscribers, dynamic add/remove); use the for await form when raw throughput matters.