emit()

emit() wires a token-producing stream into an event-style subscribe API. It ships in two substrate-specific shapes mirroring Emitter:

• Node — decorates an existing token-producing Readable in place by attaching a 'data' listener that re-emits each token's name as an EventEmitter event. Returns the same stream so calls can be chained.
• Web — takes a ReadableStream<{name, value}> and returns a fresh EventTarget that's auto-piped from the readable. The Web side can't decorate a ReadableStream in place (no event API to attach), so the return value carries the subscribe surface instead.

Introduction

Node

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

const pipeline = fs.createReadStream('sample.json').pipe(parser.asStream());
emit(pipeline);

let objectCounter = 0;
pipeline.on('startObject', () => ++objectCounter);
pipeline.on('end', () => console.log(`Found ${objectCounter} objects.`));

Web

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

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

const target = emit(readable);
let objectCounter = 0;
target.addEventListener('startObject', () => ++objectCounter);
target.addEventListener('keyValue', ev => console.log('key:', ev.detail));
// `target` is an EventTarget; the underlying readable is being drained in the background.

Subscribing happens via addEventListener; event.detail carries the token value (for valued tokens). Structural tokens fire with event.detail === undefined.

API

emit(stream) — Node

Argument is a Node Readable that produces {name, value} tokens. emit() attaches a 'data' listener that re-emits each token as an event on the same stream and returns it for chaining.

const emit = stream => {
  stream.on('data', item => stream.emit(item.name, item.value));
  return stream;
};

emit(readable, options) — Web

Argument is a ReadableStream<{name, value}>. Returns a fresh EventTarget. The readable is piped through an internal WritableStream that dispatches new CustomEvent(chunk.name, {detail: chunk.value}) per token.

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

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

Pipe errors are swallowed by the .catch(() => {}); if you need observable errors, drain the readable yourself in a for await loop (see below) and handle exceptions in the loop body.

Zero-allocation alternative for hot paths

For high-throughput streams the for await form is strictly cheaper than going through emit() — no CustomEvent per token, no listener-registry indirection, and exceptions propagate normally:

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

const handlers = {
  startObject: () => {},
  keyValue: value => {},
  stringValue: value => {},
  numberValue: value => {},
  endObject: () => {}
};

const {readable, writable} = parser.asWebStream();
sourceReadable.pipeTo(writable);
for await (const tok of readable) handlers[tok.name]?.(tok.value);

Use emit() when the subscribe-style API matters (multiple independent subscribers, dynamic add/remove); use the for await form when raw throughput matters.