Node.js Streams vs. Observables

Both Observables and node.js's Streams allow you to solve the same underlying problem: asynchronously process a sequence of values. The main difference between the two, I believe, is related to the context that motivated its appearance. That context is reflected in the terminology and API.

On the Observables side you have an extension to EcmaScript that introduces the reactive programming model. It tries to fill the gap between value generation and asynchronicity with the minimalist and composable concepts of Observer and Observable.

On node.js and Streams side you wanted to create an interface for the asynchronous and performant processing of network streams and local files. The terminology derives from that initial context and you get pipe, chunk, encoding, flush, Duplex, Buffer, etc. By having a pragmatic approach that provides explicit support for particular use cases you lose some ability to compose things because it's not as uniform. For example, you use push on a Readable stream and write on a Writable although, conceptually, you are doing the same thing: publishing a value.

So, in practice, if you look at the concepts, and if you use the option { objectMode: true }, you can match Observable with the Readable stream and Observer with the Writable stream. You can even create some simple adapters between the two models.

var Readable = require('stream').Readable;
var Writable = require('stream').Writable;
var util = require('util');

var Observable = function(subscriber) {
    this.subscribe = subscriber;
}

var Subscription = function(unsubscribe) {
    this.unsubscribe = unsubscribe;
}

Observable.fromReadable = function(readable) {
    return new Observable(function(observer) {
        function nop() {};

        var nextFn = observer.next ? observer.next.bind(observer) : nop;
        var returnFn = observer.return ? observer.return.bind(observer) : nop;
        var throwFn = observer.throw ? observer.throw.bind(observer) : nop;

        readable.on('data', nextFn);
        readable.on('end', returnFn);
        readable.on('error', throwFn);

        return new Subscription(function() {
            readable.removeListener('data', nextFn);
            readable.removeListener('end', returnFn);
            readable.removeListener('error', throwFn);
        });
    });
}

var Observer = function(handlers) {
    function nop() {};

    this.next = handlers.next || nop;
    this.return = handlers.return || nop;
    this.throw = handlers.throw || nop;
}

Observer.fromWritable = function(writable, shouldEnd, throwFn) {
    return new Observer({
        next: writable.write.bind(writable), 
        return: shouldEnd ? writable.end.bind(writable) : function() {}, 
        throw: throwFn
    });
}

You may have noticed that I changed a few names and used the simpler concepts of Observer and Subscription, introduced here, to avoid the overload of reponsibilities done by Observables in Generator. Basically, the Subscription allows you to unsubscribe from the Observable. Anyway, with the above code you can have a pipe.

Observable.fromReadable(process.stdin).subscribe(Observer.fromWritable(process.stdout));

Compared with process.stdin.pipe(process.stdout), what you have is a way to combine, filter, and transform streams that also works for any other sequence of data. You can achieve it with Readable, Transform, and Writable streams but the API favors subclassing instead of chaining Readables and applying functions. On the Observable model, For example, transforming values corresponds to applying a transformer function to the stream. It does not require a new subtype of Transform.

Observable.just = function(/*... arguments*/) {
    var values = arguments;
    return new Observable(function(observer) {
        [].forEach.call(values, function(value) {
            observer.next(value);
        });
        observer.return();
        return new Subscription(function() {});
    });
};

Observable.prototype.transform = function(transformer) {
    var source = this;
    return new Observable(function(observer) {
        return source.subscribe({
            next: function(v) {
                observer.next(transformer(v));
            },
            return: observer.return.bind(observer),
            throw: observer.throw.bind(observer)
        });
    });
};

Observable.just(1, 2, 3, 4, 5).transform(JSON.stringify)
  .subscribe(Observer.fromWritable(process.stdout))

The conclusion? It's easy to introduce the reactive model and the Observable concept anywhere. It's harder to implement an entire library around that concept. All those little functions need to work together consistently. After all, the ReactiveX project is still going at it. But if you really need to send the file content to the client, deal with encoding, and zip it then the support it's there, in NodeJS, and it works pretty well.

Recommended topics

Hot tags