@streamparser/json-whatwg

Fast dependency-free library to parse a JSON stream using utf-8 encoding in Node.js, Deno or any modern browser. Fully compliant with the JSON spec and JSON.parse(...).

tldr;

import { JSONParser } from '@streamparser/json-whatwg';

const inputStream = new ReadableStream({
  async start(controller) {
    controller.enqueue('{ "test": ["a"] }');
    controller.close();
  },
});

const parser = new JSONParser();
const reader = inputStream.pipeThrough(parser).pipeTo(destinationStream)

// Or manually getting the values

const reader = inputStream.pipeThrough(parser).getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  processValue(value);
  // There will be 3 value:
  // "a"
  // ["a"]
  // { test: ["a"] }
}

@streamparser/json ecosystem

There are multiple flavours of @streamparser:

• The @streamparser/json package allows to parse any JSON string or stream using pure JavaScript.
• The @streamparser/json-whatwg wraps @streamparser/json into a WHATWG TransformStream.
• The @streamparser/json-node wraps @streamparser/json into a node Transform stream.

Dependencies / Polyfilling

@streamparser/json requires a few ES6 classes:

• Uint8Array
• TextEncoder
• TextDecoder
• TransformStream

If you are targeting browsers or systems in which these might be missing, you need to polyfill them.

Components

Tokenizer

A JSON compliant tokenizer that parses a utf-8 stream into JSON tokens

import { Tokenizer } from '@streamparser/json-whatwg';

const tokenizer = new Tokenizer(opts, writableStrategy, readableStrategy);

Writable and readable strategy are standard WhatWG Stream settings (see MDN).

The available options are:

{
  stringBufferSize: <number>, // set to 0 to don't buffer. Min valid value is 4.
  numberBufferSize: <number>, // set to 0 to don't buffer.
  separator: <string>, // separator between object. For example `\n` for nd-js.
  emitPartialTokens: <boolean> // whether to emit tokens mid-parsing.
}

If buffer sizes are set to anything else than zero, instead of using a string to append the data as it comes in, the data is buffered using a TypedArray. A reasonable size could be 64 * 1024 (64 KB).

Buffering

When parsing strings or numbers, the parser needs to gather the data in-memory until the whole value is ready.

Strings are immutable in JavaScript so every string operation creates a new string. The V8 engine, behind Node, Deno and most modern browsers, performs many different types of optimization. One of these optimizations is to over-allocate memory when it detects many string concatenations. This increases significantly the memory consumption and can easily exhaust your memory when parsing JSON containing very large strings or numbers. For those cases, the parser can buffer the characters using a TypedArray. This requires encoding/decoding from/to the buffer into an actual string once the value is ready. This is done using the TextEncoder and TextDecoder APIs. Unfortunately, these APIs create a significant overhead when the strings are small so should be used only when strictly necessary.

TokenParser

A token parser that processes JSON tokens as emitted by the Tokenizer and emits JSON values/objects.

import { TokenParser} from '@streamparser/json-whatwg';

const tokenParser = new TokenParser(opts, writableStrategy, readableStrategy);

Writable and readable strategy are standard WhatWG Stream settings (see MDN).

The available options are:

{
  paths: <string[]>,
  keepStack: <boolean>, // whether to keep all the properties in the stack
  separator: <string>, // separator between object. For example `\n` for nd-js. If undefined, the token parser will end after parsing the first object. Whitespace between objects is always ignored. To parse multiple object without any delimiter just set it to the empty string `''`.
  emitPartialValues: <boolean>, // whether to emit values mid-parsing.
}

• paths: Array of paths to emit. Defaults to undefined which emits everything. The paths are intended to support jsonpath although at the time being it only supports the root object selector ($) and subproperties selectors including wildcards ($.a, $.*, $.a.b, , $.*.b, etc).
• keepStack: Whether to keep full objects on the stack even if they won't be emitted. Defaults to true. When set to false the it does preserve properties in the parent object some ancestor will be emitted. This means that the parent object passed to the onValue function will be empty, which doesn't reflect the truth, but it's more memory-efficient.
  • When streaming elements out of a large top-level array or object with paths (e.g. paths: ['$.*']), each event's parent/stack are snapshotted lazily, so a consumer that only reads value (the common case) stays linear regardless of size. Reading parent/stack on every one of many events is different: each read materializes a snapshot of everything parsed so far, which is quadratic overall -- set keepStack: false for that case, so already-emitted siblings are dropped instead of accumulating.

JSONParser

The full blown JSON parser. It basically chains a Tokenizer and a TokenParser.

import { JSONParser } from '@streamparser/json-whatwg';

const parser = new JSONParser();

Usage

You can use both components independently as

const tokenizer = new Tokenizer(opts);
const tokenParser = new TokenParser(opts);
const jsonParser = tokenizer.pipeThrough(tokenParser);

You can subscribe to the resulting data using the

import { JSONParser } from '@streamparser/json-whatwg';

const parser = new JSONParser({ stringBufferSize: undefined, paths: ['$'] });

const inputStream = new ReadableStream({
  async start(controller) {
    // The value can arrive split across several chunks.
    controller.enqueue('"');
    controller.enqueue('Hello');
    controller.enqueue(' ');
    controller.enqueue('world!');
    controller.enqueue('"');
    controller.close();
  },
});

const reader = inputStream.pipeThrough(parser).getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(value);
}

Write is always a synchronous operation so any error during the parsing of the stream will be thrown during the write operation. After an error, the parser can't continue parsing.

import { JSONParser } from '@streamparser/json-whatwg';

const parser = new JSONParser({ stringBufferSize: undefined });

const inputStream = new ReadableStream({
  async start(controller) {
    controller.enqueue('"""');  // invalid JSON
    controller.close();
  },
});

try {
  const reader = inputStream.pipeThrough(parser).getReader();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    console.log(value);
  }
} catch (err) {
  console.log(err); // logs the parsing error
}

Examples

Parsing a JSON array from a file

Imagine a large file containing a JSON array of objects ([{"id":1},{"id":2},{"id":3},...]) that you want to process one element at a time without loading the whole file into memory.

@streamparser/json-whatwg is a WHATWG TransformStream, so you pipe any ReadableStream of the file's bytes through it. How you obtain that stream from a file is runtime-specific.

Browser (a File from an <input type="file">)

  import { JSONParser } from '@streamparser/json-whatwg';

  const parser = new JSONParser({ paths: ['$.*'], keepStack: false });

  const reader = file.stream().pipeThrough(parser).getReader();
  while (true) {
    const { done, value: parsedElementInfo } = await reader.read();
    if (done) break;

    const { value } = parsedElementInfo;
    // TODO process element
  }

Deno (Deno.open(...).readable is a ReadableStream)

  import { JSONParser } from '@streamparser/json-whatwg';

  const parser = new JSONParser({ paths: ['$.*'], keepStack: false });

  using file = await Deno.open('arrayOfObjects.json', { read: true });

  const reader = file.readable.pipeThrough(parser).getReader();
  while (true) {
    const { done, value: parsedElementInfo } = await reader.read();
    if (done) break;

    const { value } = parsedElementInfo;
    // TODO process element
  }

Stream-parsing a fetch request returning a JSONstream

Imagine an endpoint that send a large amount of JSON objects one after the other ({"id":1}{"id":2}{"id":3}...).

  import { JSONParser} from '@streamparser/json-whatwg';

  const parser = new JSONParser();

  const response = await fetch('http://example.com/');
  const reader = response.body.pipeThrough(parser).getReader();
  while(true) {
    const { done, value } = await reader.read();
    if (done) break;
    // TODO process element
  }

Stream-parsing a fetch request returning a JSON array

Imagine an endpoint that send a large amount of JSON objects one after the other ([{"id":1},{"id":2},{"id":3},...]).

  import { JSONParser } from '@streamparser/json-whatwg';

  const parser = new JSONParser({ stringBufferSize: undefined, paths: ['$.*'], keepStack: false });

  const response = await fetch('http://example.com/');

  const reader = response.body.pipeThrough(parser).getReader();
  while(true) {
    const { done, value: parsedElementInfo } = await reader.read();
    if (done) break;

    const { value, key, parent, stack } = parsedElementInfo;
    // TODO process element
  }

Stream-parsing a fetch request returning a very long string getting previews of the string

Imagine an endpoint that send a large amount of JSON objects one after the other ("Once upon a midnight <...>").

  import { JSONParser } from '@streamparser/json-whatwg';

  const parser = new JSONParser({ stringBufferSize: undefined, paths: ['$.*'], keepStack: false });

  const response = await fetch('http://example.com/');

  const reader = response.body.pipeThrough(parser).getReader();
  while(true) {
    const { done, value: parsedElementInfo } = await reader.read();
    if (done) break;

    const { value, key, parent, stack, partial } = parsedElementInfo;
    if (partial) {
      console.log(`Parsing value: ${value}... (still parsing)`);
    } else {
      console.log(`Value parsed: ${value}`);
    }
  }

Backpressure

When the input arrives in chunks (the normal case for a stream), the parser takes part in the standard WHATWG stream backpressure: if a downstream consumer is slow, the writer is throttled (its desiredSize drops) and only a bounded number of parsed values are held in the readable buffer at a time. Just pipeThrough/pipeTo as usual and it works.

One caveat: a single write() is processed in one synchronous pass, so all the values contained in that one chunk are emitted (and buffered, if unread) at once — backpressure applies between chunks, not within one. In practice this only matters if you hand the parser an entire large document as a single write; if you already have the whole document in memory as one string that's usually fine, but to get backpressure over a large input, feed it in chunks (which is what piping from a fetch/file stream does automatically).

License

See LICENSE.md.