Node.js Flux vs Observables

Les deux Observables et node.js's Flux de vous permettre de résoudre le même problème sous-jacent: processus asynchrone d'une séquence de valeurs. La principale différence entre les deux, je crois, est liée au contexte qui a motivé son apparence. Ce contexte se reflète dans la terminologie et de l'API.

Sur le Observables côté, vous avez une extension à EcmaScript qui introduit les réactifs modèle de programmation. Il tente de combler le fossé entre la valeur de la génération et de l'asynchronicité avec le minimaliste et composable concepts de Observer et Observable.

Sur node.js et Flux côté vous avez voulu créer une interface pour l'asynchrone et performant de traitement de flux de réseau et les fichiers locaux. La terminologie en découle contexte initial et vous obtenez pipe, chunk, encoding, flush, Duplex, Buffer, etc. En ayant une approche pragmatique qui fournit un soutien explicite de l'utilisation particulière des cas, vous perdez la capacité de composer des choses parce que ce n'est pas aussi uniforme. Par exemple, vous utilisez push sur un Readable flux et write sur un Writable bien que, sur le plan conceptuel, que vous faites la même chose: la publication d'une valeur.

Donc, dans la pratique, si vous regardez les concepts, et si vous utilisez l'option { objectMode: true }, vous pouvez faire correspondre Observable avec le Readable flux et Observer avec le Writable flux. Vous pouvez même créer de simples adaptateurs entre les deux modèles.

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
});
}

Vous avez peut-être remarqué que j'ai changé quelques noms et utilisé le plus simple des concepts de Observer et Subscription, présenté ici, pour éviter la surcharge de responsabilités fait par Observables dans Generator. Fondamentalement, la Subscription vous permet de vous désabonner à partir de la Observable. De toute façon, avec le code ci-dessus, vous pouvez avoir un pipe.

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

Comparé avec process.stdin.pipe(process.stdout), ce que vous avez est un moyen de combiner, de filtre et de transformer les flux qui fonctionne également pour toute autre séquence de données. Vous pouvez la réaliser avec Readable, Transform, et Writable flux mais l'API faveurs de sous-classement au lieu d'enchaîner Readables et de l'application de fonctions. Sur le Observable modèle, Par exemple, la transformation des valeurs correspond à l'application d'un transformateur fonction du flux. Il ne nécessite pas un nouveau sous-type de 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))

La conclusion? Il est facile d'introduire le modèle réactif et le Observable concept n'importe où. Il est plus difficile de mettre en œuvre une bibliothèque entière autour de ce concept. Toutes ces petites fonctions doivent travailler ensemble de manière cohérente. Après tout, la ReactiveX projet est toujours en cours à elle. Mais si vous avez vraiment besoin d'envoyer le contenu du fichier pour le client, de traiter avec le codage, et la fermeture-éclair puis le support, c'est là, dans NodeJS, et ça marche plutôt bien.