stream

stream.js

function Stream () {
    this.fns = [];
}

Stream.prototype.push = function (element) {
    var accumulator = this.fns[0](element);
    for (var i = 1, len = this.fns.length; i < len; i++) {
        if (accumulator) { accumulator = this.fns[i](accumulator); } else { break; }
    }
};

Stream.prototype.map = function (fn) {
    this.fns.push(fn);
    return this;
};

Stream.prototype.filter = function (fn) {
    this.fns.push(function (value) {
        return fn(value) ? value : null;
    });
    return this;
};

Stream.prototype.onValue = function (fn) {
    this.fns.push(fn);
    return this;
};

var s = new Stream();

s.map(function (x) { return x + 1; }).filter(function (x) { return !(x % 2); }).onValue(function(x) {
    console.log('onValue: ', x);
});

s.push(1);
s.push(2);
s.push(3);