@portable/json-stream

Functions

`JsonStream.encodeStream`

Writes value as UTF-8 JSON text to a ReadableStream, resolving async values as it reaches them. The output is the text JSON.stringify would produce for the resolved tree, so any JSON parser can read it.

A promise or other thenable is replaced by what it resolves to.
A function is called with no arguments and replaced by its return value, so work such as a database query starts only when the writer reaches it.
A ReadableStream or async iterable is written as a JSON array of its values.

Text is written as far as the next unresolved value, then the stream pauses until that value is ready, so a reader receives the start of the document before the slow parts finish. Output is pulled: nothing is read from a nested stream until the reader asks for more. Cancelling the output cancels nested streams and calls return() on nested iterators. The output fails with the original error if a promise rejects or a nested stream errors, and with a TypeError on a circular structure or a BigInt, as JSON.stringify does.

function encodeStream(
  value: unknown,
  options: encodeStream.Options = {},
): ReadableStream<Uint8Array<ArrayBuffer>>;

Option	Type	Description
`replacer`	`Replacer \| readonly (string \| number)[]`	As in `JSON.stringify`: a function called with each key and resolved value, whose return value is written instead, or a list of the object keys to write.
`space`	`string \| number`	Indentation, as in `JSON.stringify`: a number of spaces (at most 10) or a string.
`signal`	`AbortSignal`	Errors the output with `signal.reason` when aborted, and cancels nested streams and iterators as cancelling the output does.

import * as JsonStream from '@portable/json-stream';
 
const body = JsonStream.encodeStream({
  user: getUser(id), // a promise
  posts: db.posts.cursor(), // an async iterable
});
return new Response(body, {
  headers: { 'content-type': 'application/json' },
});

Classes

`JsonStream.DecoderStream`

A TransformStream that parses JSON text as it arrives and emits the values found at the given JSONPath expressions, without holding the whole document in memory. Only matched values are built, with JSON.parse, and everything else is checked for syntax and dropped. Input may hold several top-level values separated by whitespace, as in NDJSON or concatenated JSON.

The stream fails with InvalidInputError on malformed JSON, with TruncatedError when the input ends inside a value, and with MaxLengthError on a matched value longer than maxLength. A structural error inside a matched value, such as a missing comma, is reported when that value ends.

class DecoderStream extends TransformStream<
  Uint8Array | string,
  DecoderStream.Match
> {
  constructor(options: DecoderStream.Options = {});
}

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` when aborted, which cancels the source it is piped from.
`paths`	`readonly string[]`	`['$']`	JSONPath expressions for the values to emit. Supported: `$` (the root), `.name`, `['name']`, `[0]`, `.`, `[]` and `..` (any depth). Every value that matches any path is emitted, and a nested match is emitted before the value that contains it.
`maxLength`	`number`	`Infinity`	Longest matched value, in UTF-16 code units of its JSON text. The stream holds a matched value's text until the value ends, so a longer one fails the stream with `MaxLengthError` as soon as it grows past the limit. Values outside `paths` are never held and don't count.

import * as JsonStream from '@portable/json-stream';
 
// {"total_rows":129,"rows":[{"id":"a"},{"id":"b"}]}
const response = await fetch('https://example.com/_all_docs');
const matches = response.body!.pipeThrough(
  new JsonStream.DecoderStream({
    paths: ['$.total_rows', '$.rows[*]'],
  }),
);
// { path: ['total_rows'], value: 129 }, { path: ['rows', 0], value: { id: 'a' } }, ...

Errors

`JsonStream.InvalidInputError`

Thrown when the input isn't valid JSON. The message gives the offset of the first bad character, or, for an error inside a matched value, the offset where that value starts.

class InvalidInputError extends Errors.InvalidInputError {
  readonly kind: string;
}

import * as JsonStream from '@portable/json-stream';
import * as Streams from '@portable/streams';
 
try {
  await Streams.toArray(
    body.pipeThrough(new JsonStream.DecoderStream()),
  );
} catch (error) {
  if (error instanceof JsonStream.InvalidInputError) {
    return new Response('Bad JSON', { status: 400 });
  }
  throw error;
}

`JsonStream.MaxLengthError`

Thrown when a value at one of the paths grows past maxLength.

class MaxLengthError extends Errors.MaxLengthError {
  constructor(options: Errors.MaxLengthError.Options);
  readonly kind: string;
}

import * as JsonStream from '@portable/json-stream';
import * as Streams from '@portable/streams';
 
const matches = body.pipeThrough(
  new JsonStream.DecoderStream({
    paths: ['$.rows[*]'],
    maxLength: 100_000,
  }),
);
try {
  await Streams.toArray(matches);
} catch (error) {
  if (error instanceof JsonStream.MaxLengthError) {
    return new Response(null, { status: 413 });
  }
  throw error;
}

`JsonStream.TruncatedError`

Thrown when the input ends inside a JSON value, such as a response body cut off mid-document.

class TruncatedError extends Errors.TruncatedError {
  constructor(
    message: string = 'JSON input ended inside a value.',
    options?: ErrorOptions,
  );
  readonly kind: string;
}

import * as JsonStream from '@portable/json-stream';
import * as Streams from '@portable/streams';
 
try {
  await Streams.toArray(
    body.pipeThrough(new JsonStream.DecoderStream()),
  );
} catch (error) {
  if (error instanceof JsonStream.TruncatedError) {
    console.log('connection dropped');
  }
}