Library of The Week: Highland.js

A few weeks ago, writing about the increasingly obsolete Async.js, I promised to talk about Highland.js. This library is by Caolan McMahon, who also wrote Async.js, and can be considered its successor.

Highland can be used as an alternative to promises for handling asynchronous actions, but its focus is a bit different. Highland is based on Node.js streams and aims to provide consistent functionality regardless of the type of stream. Promises are just one option: a stream giving you just a single value. As such, Highland is similar conceptually to RxJS. Because it is based on Node, you need Browserify or equivalent to use it in the browser.

To create a stream, you simple call Highland with any stream-like data: an array, a Node stream or a promise. One interesting use case is to tie it to an object that emits events:

_('click', btn).each(handleEvent);

(By convention _ is the alias for the Highland object.)

Or use it with a generator function, which allows you to create infinite streams. The following example creates an infinite stream of Fibonacci numbers and then prints the first twenty:

var fibGenerator = function() {
    var a = 0, b = 1;
    return function(push, next) {
        push(null, b);
        b = b+a; a = b-a;
        next();
    };
}();

_(fibGenerator).take(20).each(console.log.bind(console));

Streams are lazy, with values computed on demand. Also worthy of note is the "back-pressure" feature, when the producer stream is paused until the consumer is ready for more data. If the source stream does not support pausing, Highland will buffer the results for you automatically.

Once you have a stream, you can transform it. Besides the obligatory map(), there are many handy convenience methods like pluck(), as well as methods for stream filtering and slicing like latest() and first(). Don't forget that it's a stream, so you can add additional items during processing.

_(['a','b','c']).append('d').append('e');

You don't need to perform a one-to-one transformation. The following code duplicates each item:

function double(err, x, push, next) {
    // pass error and stream end normally, otherwise duplicate
    if (err) { push(err); next(); }
    else if (x === _.nil) { push(null, x); }
    else {
        push(null, x);
        push(null, x);
        next();
    }
}

_([1, 2]).consume(double);  // => [1, 1, 2, 2]

High-level functions operate on top of the streams themselves, allows you to merge them, fork them, have one stream wait for another or handle items in parallel.

Lastly, I would like to mention the nifty currying feature. All stream methods also exist on the global _ object and take an additional stream argument. So _([1,2,3,4,5]).filter(isEven); is equivalent to _.filter(isEven, _([1,2,3,4,5])). That's not very useful by itself, but this method also supports currying and can be called in sequence, e.g. _.filter(isEven)(_([1,2,3,4,5])). This allows you to create a filter instance before you have the stream itself. So you can apply the transformation later, potentially without even knowing that the original method was filter().