StreamValues

Signature: tokens → {key, value} items

StreamValues assumes that a token stream represents subsequent values and stream them out one by one.

1 "a" [] {} true
// StreamValues will produce an object stream:
{key: 0, value: 1}
{key: 1, value: 'a'}
{key: 2, value: []}
{key: 3, value: {}}
{key: 4, value: true}

StreamValues is a usual companion for Pick, which streams selected subobjects individually. It is also the standard way to terminate a token stream into a stream of plain JavaScript values — e.g. to hand each one to third-party code.

Each item arrives wrapped as {key, value} with key a running counter. To get bare values, append a functional stage: chain([…, streamValues(), data => data.value]). One caveat: a functional stage that returns null/undefined drops the item, so keep the wrapper when top-level null values matter.

As every streamer, it assumes that individual objects can fit in memory, but the whole file, or any other source, should be streamed.

It reads only packed tokens (keyValue, stringValue, numberValue) — the parser's default. See StreamBase.

Introduction

The prelude:

import {streamValues} from 'stream-json/streamers/stream-values.js';
import fs from 'node:fs';

Take a JSON streaming pipeline and pipe it into the parser:

const pipeline = fs.createReadStream('sample.json').pipe(streamValues.withParserAsStream());

pipeline.on('data', data => console.log(data));

Alternatively, first create the streaming pipeline and then pipe JSON messages into it:

const q = streamValues.withParserAsStream();
q.on('data', data => console.log(data));
fs.createReadStream('sample_a.json').pipe(q);

API

Being based on StreamBase, StreamValues has no special API.

Static methods and properties

streamValues(options)

streamValues() is the factory function. It takes options described above and returns a function for use in chain():

import chain from 'stream-chain';
import {parser} from 'stream-json';
import {streamValues} from 'stream-json/streamers/stream-values.js';

import fs from 'node:fs';

const pipeline = chain([fs.createReadStream('sample.json'), parser(), streamValues()]);

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

streamValues.asStream(options)

Returns a Duplex stream (object-mode both sides) wrapping streamValues() for .pipe() usage.

withParser()

withParser() takes one argument:

• options — combined Parser and streamer options. Passed to both the parser and streamValues().
  • jsonStreaming is always set to true because otherwise it doesn't make sense to stream just one object.

Returns a Duplex stream (text-mode writable, object-mode readable) wrapping a parser({jsonStreaming: true}) + streamValues() pipeline via stream-chain.

Built with the withParser() utility.

import {streamValues} from 'stream-json/streamers/stream-values.js';
import fs from 'node:fs';

const pipeline = fs.createReadStream('sample.json').pipe(streamValues.withParserAsStream());

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

Web Streams

streamValues ships in two substrate-specific entries with the same factory shape:

• Node — stream-json/streamers/stream-values.js. Has asStream, asWebStream, withParser, withParserAsStream, withParserAsWebStream.
• Web — stream-json/web/streamers/stream-values.js. Has asWebStream, withParser, withParserAsWebStream. Pulls in no Node-stream imports.

Both factories return the same flushable, so chain on either substrate auto-wraps it. Use chain from stream-chain on Node and from stream-chain/web on Web.

// Web
import {chain} from 'stream-chain/web';
import {parser} from 'stream-json/web/parser.js';
import {streamValues} from 'stream-json/web/streamers/stream-values.js';

const pipeline = chain([source, parser({jsonStreaming: true}), streamValues()]);
for await (const item of pipeline.readable) console.log(item);

withParserAsWebStream sets jsonStreaming: true automatically:

const {readable, writable} = streamValues.withParserAsWebStream();
sourceReadable.pipeTo(writable);
for await (const item of readable) console.log(item);