PipelineJS

PipelineJS is a data pipeline library for Javascript. It is intended to bring some of the design pattern advantages of channel based piplines from lanuages like Go to Javascript for projects that benifit from Javascripts other advantages.

Note: the Javascript event loop limits what performance gains can be achieved with a channel based pipeline. However, the design pattern is still very helpful for certain project types.

Installation

Install with npm or yarn

yarn add @scottsloane/pipelinejs

npm i --save @scottsloane/pipelinejs

Usage

Import the Pipeline class from the library

import Pipeline from '@scottsloane/pipelinejs' 

Create you segment functions.

See example below

Insance a new pipeline passing it your segment functions.

const pipeline = new Pipeline(catalyst, [func1, func2]); 

run the pipeline

Inside an async scope:

await pipeline.run();

Example

import Pipeline from 'pipelinejs';

const Data = [];
for (let i = 0; i < 100; i++) {
    Data.push(i);
}

function* catalist() {
    while (Data.length > 0) {
        yield Data.shift();
    }
}

function square(data) {
    console.log(`Square: ${data}`);
    return data * data;
}

function consumer (data) {
    console.log(`Consumer: ${data}`);
}

(async () => {
    const pipeline = new Pipeline(catalist, [square, consumer]);
    await pipeline.run();
    console.log('Pipeline: done');
})();

The Pipeline Class

The pipeline class takes three arguments.

1. function*: Catalyst
2. Array<segment/subpipeline>: Segments
3. (otional) function: Terminator

Note: If no terminator is specified, a noop terminator will be added to prevent back stalling on the channels.

The pipeline class has two public member functions:

setContext(data)

Set context allows the addition of context data that will be made available to every segment. This is helpful for things like sharing a database connection pool across the entire pipeline.

async run(void)

Runs the pipline, starting all segments.

Subpipelines

A Subpipeline is a pipeline that is embeded in the middle of a parent pipeline. These subpipelines allow for parallel processing within a pipeline.

Note: a Subpipeline can not contain a catalyst or terminator.

A Subpipeline can be added in the Pipeline constructor by adding a segment that is an array of two items:

1. Number: the number of parallel pipes to create
2. Array<segment>: The segments in the Subpipeline

const pipeline = new Pipeline(catalyst, [func1, [4,[subfunc1, subfunc2]], func2]);

Segment Types

Catalyst*

Generator function with a channel output, used to start a pipeline.

Function (Segment)

A standard segment function. Takes data from a channel and outputs a return result to another channel.

Function* (GeneratorSegment)

A generator segment function. Takes data from a channel and yields output to another channel. Useful if the function generates a lot of data.

Terminator

A function segment without an output channel. Used to terminate (or end) a pipeline. (needed to prevent back stalling the channels)