stream-fork

stream-fork is a Writable stream, which writes into dependent Writable streams properly handling backpressure. It is a way to make forks in a linear pipeline of streams.

Originally stream-fork was used internally with stream-chain and stream-json to create flexible data processing pipelines.

stream-fork is a lightweight, no-dependencies micro-package. It is distributed under New BSD license.

Intro

const Fork = require('stream-fork');
const fs = require('fs');
const zlib = require('zlib');
 
const gzip = zlib.createGzip();
gzip.pipe(fs.createWriteStream('log.txt.gz'));
 
// push data to both the gzip chain and stdout
const forkStream = new Fork([gzip, process.stdout], {});
dataSource.pipe(forkStream);
 
// now we have data on our screen and as a compressed log file

Installation

npm i --save stream-fork
# or: yarn add stream-fork 

Documentation

Fork, which is returned by require('stream-fork'), is a specialized Writable stream. It propagates every piece of data downstream to its dependent Writable streams (including Transform and Duplex streams).

Many details about this package can be discovered by looking at test files located in tests/ and in the source code (main.js).

Constructor: new Fork(outputs[, options])

The constructor accepts following arguments:

• outputs is an array of Writable streams, which will be used duplicate written chunks or items.
• options is an options object, which is used to create a Writable stream. Read all about it in Implementing a Writable stream. If it is not specified or falsy, {objectMode: true} is assumed. This default is useful for creating object mode streams.
  • Additionally following custom options are recognized:
    • ignoreErrors is a flag. When its value is truthy, a Fork instance never fails on write() silently ignoring downstream errors. Otherwise, the first encountered downstream error is reported upstream as its own error. The default: false.

const forkObj = new Fork(streams);     // object mode
const forkChk = new Fork(streams, {}); // chunk mode (text or buffers)

Property: outputs

It is an array of Writable streams supplied in the constructor above. If a stream fails on writing a chunk, eventually it will be removed from the array.

const forkStream = new Fork(streams);
forkStream.outputs.length === streams.length; // true

Method: isEmpty()

It returns true if outputs property is empty, and false otherwise. If isEmpty() is true, it means that the stream do not duplicate data.

const forkStream = new Fork([]);
forkStream.isEmpty(); // true

Static method: fork(outputs[, options])

It is a factory function, which accepts the same arguments as the constructor, and returns a fully constructed Fork object.

// replicating the introduction example above
const {fork} = require('stream-fork');
dataSource.pipe(fork([gzip, process.stdout], {}));

Release History

• 1.0.5 technical release.
• 1.0.4 bugfix: forward errors correctly, thx dbubovych.
• 1.0.3 technical release to support Node 14.
• 1.0.2 workaround for Node 6: use 'finish' event instead of _final().
• 1.0.1 improved documentation.
• 1.0.0 the initial release.