nodejs-streams.js

// ref.: https://medium.freecodecamp.org/node-js-streams-everything-you-need-to-know-c9141306be93

var stream = require('stream');
var Writable = stream.Writable;
var Readable = stream.Readable;
var Duplex = stream.Duplex;
var Transform = stream.Transform;

function consoleWriterStream(){
    var writeStream = new Writable({
        'write': function(chunk, encoding, callback){
            console.log(chunk.toString());
            callback();
        }
    });

    return writeStream;
}

function consoleProducerStream(){
    process.stdin.pipe(writeStream);
}

function inMemoryProducerStream(){
    var texts = ['t1', 't2', 't3', 't4', 't5'];
    var nextTextIdx = 0;

    var readableStream = new Readable({
        'read': function(size){
            if(texts[nextTextIdx]) {
                this.push(texts[nextTextIdx++]);
            } else {
                this.push(null);
            }
        }
    });

    return readableStream;
}

function duplexStream(){
    var texts = ['t1', 't2', 't3', 't4', 't5'];
    var nextTextIdx = 0;

    var inoutStream = new Duplex({
        'write': function(chunk, encoding, callback) {
            console.log(chunk.toString());
            callback();
        },

        'read': function(size) {
            if(texts[nextTextIdx]) {
                this.push(texts[nextTextIdx++] + "\n");
            } else {
                this.push(null);
            }
        }
    });

    return inoutStream;
}

function transformStream(){
    var upperCaseTr = new Transform({
        /*readableObjectMode: true,
        writableObjectMode: true,*/
        transform(chunk, encoding, callback) {
            this.push(chunk.toString().toUpperCase());
            callback();
        }
    });

    return upperCaseTr;
}

function run1(){
    var inStream = inMemoryProducerStream();
    var outStream = consoleWriterStream();

    inStream.pipe(outStream);
}

function run2(){
    var inoutStream = duplexStream();
    var transform = transformStream();
    inoutStream.pipe(transform).pipe(inoutStream);
}

run2();