Replace

Replace is a token item filter based on FilterBase. It replaces objects from a stream passing the rest as is. Its filter is not called for subobjects of replaced objects. In certain conditions, it can replace items with nothing effectively ignoring values.

There is a helper filter Ignore, which is used to remove values completely. It is based on Replace.

Introduction

const Replace = require('stream-json/filters/Replace');
const {streamArray} = require('stream-json/streamers/StreamArray');
const {chain} = require('stream-chain');
const fs = require('fs');

// our data stream consists of items like that:
// {total: 123456789, meta: {...}, data: [...]}
// we want to replace all 'meta' properties with 'null'

const pipeline = chain([
  fs.createReadStream('sample.json'),
  Replace.withParser({filter: /\bmeta\b/i}),
  streamArray()
]);

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

The same example, but we want to replace all meta with 0:

const {replace} = require('stream-json/filters/Replace');
const {parser} = require('stream-json/Parser');
const {streamArray} = require('stream-json/streamers/StreamArray');

const pipeline = chain([
  fs.createReadStream('sample.json'),
  parser(),
  replace({filter: /\bmeta\b/i, replacement: [
    {name: 'startNumber'},
    {name: 'numberChunk', value: '0'},
    {name: 'endNumber'},
    {name: 'numberValue', value: '0'}
  ]}),
  streamArray()
]);

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

Like before but we want to remove all meta properties (make sure that a filter gets packed keys, and don't forget to allow empty replacements):

const {replace} = require('stream-json/filters/Replace');
const {parser} = require('stream-json/Parser');
const {streamArray} = require('stream-json/streamers/StreamArray');

const pipeline = chain([
  fs.createReadStream('sample.json'),
  parser(), // packs keys by default, otherwise {packKeys: true}
  replace({filter: /\bmeta\b/i, replacement: [], allowEmptyReplacement: true}),
  // better: ignore({filter: /\bmeta\b/i}),
  streamArray()
]);

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

API

Replace has no special API. Based on FilterBase it uses the full set of options:

• filter
  • If it returns a truthy value, the current object is removed completely with all its possible subobjects and replaced with an array of tokens.
  • It is called only for the following tokens: startObject, startArray, startString, startNumber, stringValue, numberValue, nullValue, trueValue, falseValue.
• pathSeparator
• once
• replacement
• allowEmptyReplacement
  • This options is required in order to remove objects.
  • The upstream should produce packed keys (keyValue tokens) for it to work.
• Streaming options:
  • streamValues
  • streamKeys

See their definitions in FilterBase.

Static methods and properties

replace(options) and make(options)

make() and replace() are two aliases of the factory function. It takes options described above, and return a new instance of Replace. replace() helps to reduce a boilerplate when creating data processing pipelines:

const {chain}   = require('stream-chain');
const {parser}  = require('stream-json/Parser');
const {replace} = require('stream-json/filters/Replace');

const fs = require('fs');

const pipeline = chain([
  fs.createReadStream('sample.json'),
  parser(), // packs keys by default
  replace({filter: /\bmeta\b/}, replace: [], allowEmptyReplacement: true)
]);

let metaCounter = 0;
pipeline.on('data', data => data.name === 'keyValue' && data.value === 'meta' && ++metaCounter);
pipeline.on('end', console.log(`Found ${metaCounter} objects. Should be 0.`));

make.Constructor

Constructor property of make() (and replace()) is set to Replace. It can be used for indirect creating of filters or metaprogramming if needed.

withParser()

withParser() takes one argument:

• options is an object described in Parser's options. It is used to initialize both streams (a Parser instance and a stream returned by make()).

It returns a stream produced by stream-chain, which wraps the pipeline. The most important utility of withParser() is that it correctly sets object modes of the returned stream: object mode for the Readable part and text mode for the Writable part.

This static method is created using withParser() utility. It simplifies a case when a stream should be immediately preceded by a parser.

const Replace = require('stream-json/filters/Replace');
const fs = require('fs');

const pipeline = fs.createReadStream('sample.json')
  .pipe(Replace.withParser({filter: /\bmeta\b/}));

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