StreamFlow

Encapsulate streams for use in object-oriented programming.

Replace transform stream

const StreamFlow = require("stream-workflow");
const { Transform, pipeline } = require("stream");
const JSONStream = require("JSONStream");
const fs = require("fs");

fs.createReadStream(`data.json`)
    .pipe(JSONStream.parse("*"))
    .pipe(new Transform({
        objectMode: true,
        transform(chunk, enc, cb) {
            cb(null, chunk);
        }
    }));

by stream-workflow

class CustomStream extends StreamFlow { {
    constructor({
        objectMode: true,
        init(stream) {              // init function must return the last stream of the pipeline.
            return pipeline(        
                stream,             // pathtrought stream
                JSONStream.parse(), // Transform 1 in diagram
                new Transform({     // Transform 2 in diagram
                    objectMode: true,
                    transform(chunk, enc, cb) {
                        cb(null, chunk);
                    }
                }),
                error => error && this.emit("error", error)
            )
        }
    })
}

fs.createReadStream(`data.json`)
    .pipe(new CustomStream());

Test

To run our tests, clone the stream-workflow repo and install the dependencies.

$ git clone https://github.com/BenoitClaveau/stream-workflow --depth 1
$ cd stream-workflow
$ npm install
$ cd tests
$ node.exe "../node_modules/mocha/bin/mocha" .