@portable/jsonl

Classes

`Jsonl.DecoderStream`

A TransformStream that decodes NDJSON (newline-delimited JSON, also called JSON Lines) into values, one per line. Lines may end in LF or CRLF. Blank and whitespace-only lines are skipped. Each line is parsed with JSON.parse, so a null line is emitted as null.

The stream fails with InvalidInputError on a line that isn't valid JSON, unless invalid is 'skip', and with MaxLengthError on a line longer than maxLength.

class DecoderStream extends TransformStream<
  Uint8Array | string,
  unknown
> {
  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.
`invalid`	`"error" \| "skip"`	`'error'`	What to do with a line that isn't valid JSON: fail the stream, or drop the line and continue.
`maxLength`	`number`	`Infinity`	Longest line, in UTF-16 code units, line ending excluded. A longer line fails the stream with `MaxLengthError` as soon as it grows past the limit, so a line without a newline can't use unbounded memory.

import * as Jsonl from '@portable/jsonl';
 
const response = await fetch('https://example.com/events.ndjson');
const events = response.body!.pipeThrough(new Jsonl.DecoderStream());

`Jsonl.EncoderStream`

A TransformStream that encodes each value as one line of NDJSON: JSON.stringify(value) followed by \n, as UTF-8 bytes. If JSON.stringify can't encode a value, such as undefined, a function, a symbol, a BigInt or a circular structure, the stream fails with a TypeError instead of dropping the line or writing a malformed one.

class EncoderStream extends TransformStream<
  unknown,
  Uint8Array<ArrayBuffer>
> {
  constructor(options: EncoderStream.Options = {});
}

Option	Type	Default	Description
`signal`	`AbortSignal`		Errors the stream with `signal.reason` when aborted, which cancels the source it is piped from.

import * as Jsonl from '@portable/jsonl';
import * as Streams from '@portable/streams';
 
const body = Streams.fromIterable(rows).pipeThrough(
  new Jsonl.EncoderStream(),
);
return new Response(body, {
  headers: { 'content-type': 'application/x-ndjson' },
});

Errors

`Jsonl.InvalidInputError`

Thrown when a line isn't valid JSON. The message gives the line number, and cause holds the SyntaxError from JSON.parse.

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

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

`Jsonl.MaxLengthError`

Thrown when a line grows past maxLength, before it is parsed.

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

import * as Jsonl from '@portable/jsonl';
import * as Streams from '@portable/streams';
 
try {
  await Streams.toArray(
    body.pipeThrough(new Jsonl.DecoderStream({ maxLength: 1_000_000 })),
  );
} catch (error) {
  if (error instanceof Jsonl.MaxLengthError) {
    return new Response(null, { status: 413 });
  }
  throw error;
}