Comment développer en node.js avec les flux

Tous les différents types de flux utilisent .pipe() pour appairer les entrées avec les sorties.

.pipe() n'est qu'une fonction qui prend un flux source en lecture src et branche sa sortie sur un flux destination en écriture dst :

src.pipe(dst)

.pipe(dst) retourne dst afin que vous puissiez chaîner de multiples appels .pipe() ensemble :

a.pipe(b).pipe(c).pipe(d)

qui est équivalent à

a.pipe(b);
b.pipe(c);
c.pipe(d);

ce qui est beaucoup plus comparable à ce que vous pourriez faire en ligne de commande pour enchaîner les programmes ensemble :

a | b | c | d

Sauf que ça fonctionne en shell, pas avec node !

Les flux en lecture produisent des données qui peuvent être injectées dans un flux writable, transform ou duplex en appelant .pipe() :

readableStream.pipe(dst)

III-B-1. Créer un flux en lecture ▲

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

var rs = new Readable;
rs.push('beep ');
rs.push('boop\n');
rs.push(null);

rs.pipe(process.stdout);

$ node read0.js
beep boop

var Readable = require('stream').Readable;
var rs = Readable();

var c = 97;
rs._read = function () {
    rs.push(String.fromCharCode(c++));
    if (c > 'z'.charCodeAt(0)) rs.push(null);
};

rs.pipe(process.stdout);

$ node read1.js
abcdefghijklmnopqrstuvwxyz

var Readable = require('stream').Readable;
var rs = Readable();

var c = 97 - 1;

rs._read = function () {
    if (c >= 'z'.charCodeAt(0)) return rs.push(null);
    
    setTimeout(function () {
        rs.push(String.fromCharCode(++c));
    }, 100);
};

rs.pipe(process.stdout);

process.on('exit', function () {
    console.error('\n_read() called ' + (c - 97) + ' times');
});
process.stdout.on('error', process.exit);

$ node read2.js | head -c5
abcde
_read() called 5 times

III-B-2. Consommer un flux en lecture▲

process.stdin.on('readable', function () {
    var buf = process.stdin.read();
    console.dir(buf);
});

$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume0.js 
<Buffer 61 62 63 0a>
<Buffer 64 65 66 0a>
<Buffer 67 68 69 0a>
null

process.stdin.on('readable', function () {
    var buf = process.stdin.read(3);
    console.dir(buf);
});

$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume1.js 
<Buffer 61 62 63>
<Buffer 0a 64 65>
<Buffer 66 0a 67>

process.stdin.on('readable', function () {
    var buf = process.stdin.read(3);
    console.dir(buf);
    process.stdin.read(0);
});

$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume2.js 
<Buffer 61 62 63>
<Buffer 0a 64 65>
<Buffer 66 0a 67>
<Buffer 68 69 0a>

var offset = 0;

process.stdin.on('readable', function () {
    var buf = process.stdin.read();
    if (!buf) return;
    for (; offset < buf.length; offset++) {
        if (buf[offset] === 0x0a) {
            console.dir(buf.slice(0, offset).toString());
            buf = buf.slice(offset + 1);
            offset = 0;
            process.stdin.unshift(buf);
            return;
        }
    }
    process.stdin.unshift(buf);
});

$ tail -n +50000 /usr/share/dict/american-english | head -n10 | node lines.js 
'hearties'
'heartiest'
'heartily'
'heartiness'
'heartiness\'s'
'heartland'
'heartland\'s'
'heartlands'
'heartless'
'heartlessly'

Un flux en écriture est un flux que l'on peut passer en paramètre à un .pipe(), mais à partir duquel on ne peut pas invoquer cette méthode :

src.pipe(writableStream)

III-C-1. Créer un flux en écriture ▲

var Writable = require('stream').Writable; 
var ws = Writable(); 
ws._write = function (chunk, enc, next) { 
    console.dir(chunk); 
    next(); 
}; 
process.stdin.pipe(ws);

$ (echo beep; sleep 1; echo boop) | node write0.js 
<Buffer 62 65 65 70 0a> 
<Buffer 62 6f 6f 70 0a>

III-C-2. Écrire sur un flux en écriture ▲

process.stdout.write('beep boop\n');

var fs = require('fs'); 
var ws = fs.createWriteStream('message.txt'); 

ws.write('beep '); 

setTimeout(function () { 
    ws.end('boop\n'); 
}, 1000);

$ node writing1.js 
$ cat message.txt 
beep boop

Les flux de transformation sont une catégorie de flux duplex (à la fois en lecture et écriture). La différence vient du fait que dans un flux de transformation, la sortie est en quelque sorte calculée depuis l'entrée.

Vous pourrez aussi entendre parler de flux de transformation sous le nom de flux traversants.

Les flux traversants sont de simples filtres lecture/écriture qui transforment une entrée et produisent une sortie.

Les flux duplex sont en lecture/écriture et les deux extrémités du flux sont partie prenante d'un dialogue dans les deux sens, envoyant des messages en avant et en arrière comme un téléphone. Un dialogue RPC est un bon exemple de flux duplex. À chaque fois que vous verrez des choses comme : a.pipe(b).pipe(a), vous aurez affaire à un flux duplex.

Les flux classiques sont les anciennes interfaces qui apparurent en premier dans node 0.4. Vous rencontrerez probablement ce genre de flux pendant un bon moment, aussi il est bon de savoir comment ils fonctionnent.

Sitôt qu'un flux s'est abonné à un écouteur d'événements « données », il bascule en mode classique et se comporte conformément à l'ancienne API.

III-F-1. Les flux classiques en lecture▲

var Stream = require('stream');
var stream = new Stream;
stream.readable = true;

var c = 64;
var iv = setInterval(function () {
    if (++c >= 75) {
        clearInterval(iv);
        stream.emit('end');
    }
    else stream.emit('data', String.fromCharCode(c));
}, 100);

stream.pipe(process.stdout);

$ node classic0.js
ABCDEFGHIJ

process.stdin.on('data', function (buf) {
    console.log(buf);
});
process.stdin.on('end', function () {
    console.log('__END__');
});

$ (echo beep; sleep 1; echo boop) | node classic1.js 
<Buffer 62 65 65 70 0a>
<Buffer 62 6f 6f 70 0a>
__END__

var through = require('through');
process.stdin.pipe(through(write, end));

function write (buf) {
    console.log(buf);
}
function end () {
    console.log('__END__');
}

$ (echo beep; sleep 1; echo boop) | node through.js 
<Buffer 62 65 65 70 0a>
<Buffer 62 6f 6f 70 0a>
__END__

var concat = require('concat-stream');
process.stdin.pipe(concat(function (body) {
    console.log(JSON.parse(body));
}));

$ echo '{"beep":"boop"}' | node concat.js 
{ beep: 'boop' }

III-F-2. Flux classiques en écriture▲

V-A-1. process.stdin▲

V-A-2. process.stdout▲

V-A-3. process.stderr▲

V-A-4. Net▲

concat-stream

concat-stream accumulera les données du flux dans un unique buffer. concat(cb) prend une unique fonction de callback .cb(body) avec le 'body' bufférisé quand le flux a fini.

Par exemple, dans ce programme, la fonction de 'callback' concat se déclenche avec la chaîne de caractères « beep boop » dès que cs.end() est appelé. Le programme prend ce corps de texte et le met en majuscules, affichant alors «  BEEP BOOB »

var concat = require('concat-stream');

var cs = concat(function (body) {
    console.log(body.toUpperCase());
});
cs.write('beep ');
cs.write('boop.');
cs.end();

$ node concat.js
BEEP BOOP.

Voici un exemple d'utilisation d'un concat-steam qui parsera les données d'un formulaire entrant encodées dans l'URL et renverra les données de celui-ci en JSON « stringifié ».

var http = require('http');
var qs = require('querystring');
var concat = require('concat-stream');

var server = http.createServer(function (req, res) {
    req.pipe(concat(function (body) {
        var params = qs.parse(body.toString());
        res.end(JSON.stringify(params) + '\n');
    }));
});
server.listen(5005);

$ curl -X POST -d 'beep=boop&dinosaur=trex' http://localhost:5005
{"beep":"boop","dinosaur":"trex"}

Les scuttlebut (ragots) peuvent être utilisés pour de la synchronisation d'état en peer-to-peer sur des topologies de réseau dans lesquelles les nœuds ne sont connectés qu'à travers des intermédiaires, et où aucun nœud ne dispose d'une version de référence de toutes les données.

Ce genre de réseau peer-to-peer distribué que scuttlebutt fournit est particulièrement utile quand les nœuds de part et d'autre d'une barrière de réseaux ont besoin de partager et de mettre à jour un même état.

Un exemple de ce genre de réseau serait des navigateurs qui s'envoient des messages via un serveur HTTP, qui fait tourner des processus auxquels les navigateurs ne peuvent pas se connecter directement. Un autre cas d'utilisation serait celui des systèmes qui étendent les réseaux internes puisque les adresses IPV4 sont insuffisantes.

Si scuttlebut utilise un protocole bavard (gossip protocol) pour échanger des messages entre les nœuds connectés, c'est afin que l'état à travers tous les nœuds finisse par converger partout vers la même valeur.

En utilisant l'interface scuttlebutt/model, nous pouvons créer des nœuds et les connecter entre eux pour créer toutes les sortes de réseaux que nous voulons :

var Model = require('scuttlebutt/model');
var am = new Model;
var as = am.createStream();

var bm = new Model;
var bs = bm.createStream();

var cm = new Model;
var cs = cm.createStream();

var dm = new Model;
var ds = dm.createStream();

var em = new Model;
var es = em.createStream();

as.pipe(bs).pipe(as);
bs.pipe(cs).pipe(bs);
bs.pipe(ds).pipe(bs);
ds.pipe(es).pipe(ds);

em.on('update', function (key, value, source) {
    console.log(key + ' => ' + value + ' from ' + source);
});

am.set('x', 555);

Le réseau que nous avons créé est un graphe non orienté qui ressemble à :

Notons que les nœuds a et e ne sont pas directement connectés, mais lorsque nous exécutons ce script :

$ node model.js
x => 555 from 1347857300518

la valeur que le nœud a impose trouve son chemin via les nœuds b et d. Ici tous les nœuds sont dans le même processus, mais parce que scuttlebutt utilise une simple interface de flux, les nœuds peuvent être placés sur n'importe quel processus ou serveur et connectés avec n'importe quel flux de transports qui peut prendre en charge une chaîne de caractères.

Dans un second temps, nous pouvons donner un exemple plus réaliste qui connecte au réseau et implémente une variable compteur.

Voici le serveur qui initialisera la variable count à 0 et l'incrémentera toutes les 320 millisecondes, imprimant toutes les mises à jour du compteur :

var Model = require('scuttlebutt/model');
var net = require('net');

var m = new Model;
m.set('count', '0');
m.on('update', function (key, value) {
    console.log(key + ' = ' + m.get('count'));
});

var server = net.createServer(function (stream) {
    stream.pipe(m.createStream()).pipe(stream);
});
server.listen(8888);

setInterval(function () {
    m.set('count', Number(m.get('count')) + 1);
}, 320);

Maintenant nous pouvons écrire le client qui se connecte au serveur, met à jour le compteur et imprime toutes les mises à jour qu'il reçoit :

var Model = require('scuttlebutt/model');
var net = require('net');

var m = new Model;
var s = m.createStream();

s.pipe(net.connect(8888, 'localhost')).pipe(s);

m.on('update', function cb (key) {
    // wait until we've gotten at least one count value from the network
    if (key !== 'count') return;
    m.removeListener('update', cb);
    
    setInterval(function () {
        m.set('count', Number(m.get('count')) + 1);
    }, 100);
});

m.on('update', function (key, value) {
    console.log(key + ' = ' + value);
});

Le client est légèrement plus malin, puisqu'il pourrait attendre jusqu'à ce qu'il reçoive une mise à jour de quelqu'un d'autre pour débuter la mise à jour du compteur lui-même, voire réinitialiser son compteur à zéro.

Une fois que le serveur et quelques clients tournent, nous devrions voir une séquence de ce type :

count = 183
count = 184
count = 185
count = 186
count = 187
count = 188
count = 189

Plus rarement, sur certains nœuds, nous pourrions observer une séquence avec des valeurs qui se répètent comme :

count = 147
count = 148
count = 149
count = 149
count = 150
count = 151

Ces valeurs sont dues à l'algorithme historique de scuttlebutt de résolution des conflits avec lequel il est difficile de garantir un état du système à travers des nœuds parfaitement cohérents.

Notez que le serveur, dans cet exemple, n'est qu'un nœud de plus avec les mêmes privilèges que les clients avec lesquels il est connecté. Les termes clients et serveur, ici, n'affectent en rien la façon dont la synchronisation est effectuée, mais juste celui qui initie la connexion. Les protocoles ayant cette propriété sont souvent appelés protocoles symétriques (cf. dnode pour d'autres exemples de protocoles symétriques).

JSONStream

Utilisez ce module pour parser et « stringifier » des données JSON depuis un flux.

Si vous avez besoin d'envoyer une collection JSON volumineuse en passant par une liaison bas débit ou si vous avez un objet qui se peuplera lentement, ce module vous permettra de parser les données au fur et à mesure de leur arrivée.

dnode vous permet d'appeler des fonctions distantes à travers tout type de flux.

Voici un serveur dnode basique :

var dnode = require('dnode');
var net = require('net');

var server = net.createServer(function (c) {
    var d = dnode({
        transform : function (s, cb) {
            cb(s.replace(/[aeiou]{2,}/, 'oo').toUpperCase())
        }
    });
    c.pipe(d).pipe(c);
});

server.listen(5004);

Maintenant, vous pouvez écrire un simple client qui appelle la fonction .transform() de ce serveur :

var dnode = require('dnode');
var net = require('net');

var d = dnode();
d.on('remote', function (remote) {
    remote.transform('beep', function (s) {
        console.log('beep => ' + s);
        d.end();
    });
});

var c = net.connect(5004);
c.pipe(d).pipe(c);

Lancez le serveur et lorsque vous exécutez le client, vous devriez voir :

$ node client.js
beep => BOOP

Le client à envoyé 'beep' à la fonction transform() du serveur et le serveur a appelé la fonction de 'callback' du client avec le résultat, élégant !

L'interface de flux que dnode fourni ici est un flux duplex puisque le client et le serveur sont connectés l'un à l'autre ( c.pipe(d).pipe(c) ) avec les requêtes et les réponses échangées dans les deux sens.

Là où ça devient fou, c'est quand vous commencez à passer des arguments de fonction qui chaînent des fonctions de 'callback'. Voici une version mise à jour du serveur précédant avec une cinématique d'appels des fonctions de 'callback' emboîtés :

var dnode = require('dnode');
var net = require('net');

var server = net.createServer(function (c) {
    var d = dnode({
        transform : function (s, cb) {
            cb(function (n, fn) {
                var oo = Array(n+1).join('o');
                fn(s.replace(/[aeiou]{2,}/, oo).toUpperCase());
            });
        }
    });
    c.pipe(d).pipe(c);
});
server.listen(5004);

Voici le client mis à jour :

var dnode = require('dnode');
var net = require('net');

var d = dnode();
d.on('remote', function (remote) {
    remote.transform('beep', function (cb) {
        cb(10, function (s) {
            console.log('beep:10 => ' + s);
            d.end();
        });
    });
});

var c = net.connect(5004);
c.pipe(d).pipe(c);

Après avoir relancé le serveur (nouvelle version), voilà ce que nous obtenons en exécutant le client :

$ node client.js
beep:10 => BOOOOOOOOOOP

Ça fonctionne, tout simplement !

L'idée de base, c'est qu'il vous suffit de placer des fonctions dans des objets et de les appeler depuis l'autre côté d'un flux pour que les fonctions puissent faire les allers-retours et prendre leur place dans l'enchaînement des appels comme si le processus était local. Le plus extraordinaire, c'est que lorsque vous passez des fonctions à des fonctions souches en tant qu'arguments, ces fonctions se retrouvent en qualité de souches de l'autre côté !

Cette approche qui consiste à empiler des appels de fonctions récursivement est désormais connue sous le nom de la "manœuvre des tortues qui se marchent sur la carapace". La valeur de retour de chacune de ces fonctions sera ignorée et seule une énumération des propriétés des objets sera envoyée, façon JSON.

C'est la pyramide des tortues !

Comme dnode s'exécute dans node.js ou dans un navigateur à travers n'importe quel flux, il est facile d'appeler des fonctions définies n'importe où, et particulièrement utiles lorsqu'elles sont associées avec mux-demux pour multiplexer un flux RPC afin de contrôler un flux de données très volumineux.

Le module append-only nous apporte un tableau append-only (ajout uniquement) confortable par-dessus la couche scuttlebutt qui permet d'écrire très simplement un chat distribué et cohérent entre plusieurs nœuds pouvant survivre à des coupures partielles du réseau.

Voyons comment implémenter ce chat.

Nous pouvons construire une API pour émettre des événements à la mode Socket.io en utilisant les flux vus plus haut dans cet article.

Tout d'abord, nous pouvons utiliser shoe pour créer un nouveau gestionnaire de websocket côté serveur et avec emit-stream qui transformera un émetteur d'événements en un flux qui émet des objets. Le flux objet peut alors être alimenté par JSONStream pour sérialiser des objets, et de là, le flux sérialisé peut être raccordé à un navigateur distant.

var EventEmitter = require('events').EventEmitter;
var shoe = require('shoe');
var emitStream = require('emit-stream');
var JSONStream = require('JSONStream');

var sock = shoe(function (stream) {
    var ev = new EventEmitter;
    emitStream(ev)
        .pipe(JSONStream.stringify())
        .pipe(stream)
    ;
    ...
});

Dans la fonction de 'callback' de shoe, nous pouvons émettre des événements pour la fonction ev. Ici, nous ne ferons qu'émettre différents types d'événements à intervalles réguliers :

var intervals = [];

intervals.push(setInterval(function () {
    ev.emit('upper', 'abc');
}, 500));

intervals.push(setInterval(function () {
    ev.emit('lower', 'def');
}, 300));

stream.on('end', function () {
    intervals.forEach(clearInterval);
});

In fine, l'instance de shoe devra juste être liée au serveur http :

var http = require('http');
var server = http.createServer(require('ecstatic')(__dirname));
server.listen(8080);

sock.install(server, '/sock');

Pendant ce temps, côté navigateur, il suffit de parser le flux shoe json et de transmettre le flux objet résultant à eventStream(). eventStream() ne fait que retourner un émetteur d'événements qui les émet coté serveur :

var shoe = require('shoe');
var emitStream = require('emit-stream');
var JSONStream = require('JSONStream');

var parser = JSONStream.parse([true]);
var stream = parser.pipe(shoe('/sock')).pipe(parser);
var ev = emitStream(stream);

ev.on('lower', function (msg) {
    var div = document.createElement('div');
    div.textContent = msg.toLowerCase();
    document.body.appendChild(div);
});

ev.on('upper', function (msg) {
    var div = document.createElement('div');
    div.textContent = msg.toUpperCase();
    document.body.appendChild(div);
});

Utilisez browserify pour compiler le code source du navigateur afin que vous puissiez appeler require() sur tous ces modules malins côté navigateur :

$ browserify main.js -o bundle.js

Il ne reste plus qu'à indiquer <script src="/bundle.js"></script> dans le code html et l'exécuter dans un navigateur pour voir les événements côté serveur se propager côté navigateur.

Avec cette approche orientée flux, vous pouvez vous fier à des composants légers et réutilisables qui ont juste besoin de savoir comment dialoguer avec un flux. Au lieu de router des messages à travers un système de gestion globale d'événements à la mode socket.io, vous pouvez vous concentrer sur le fractionnement de votre application en petites unités qui implémentent correctement une fonctionnalité unique.

Par exemple, vous pouvez trivialement remplacer JSONStream dans cet exemple stream-serializer pour obtenir une sérialisation différente pour un codage différent. Vous pouvez abandonner les couches au-dessus de shoe pour gérer les reconnexions et pulsations de réseau, et utiliser des interfaces simples de streaming. Vous pouvez même ajouter un flux dans une chaîne pour utiliser namespaced events avec eventemitter2 au lieu de EventEmitter du noyau.

Si vous voulez des flux différents qui agissent de façon différente, il sera également avantageusement simple d'exécuter mux-demux à travers cet exemple pour créer des canaux dissociés pour chacun des différents types de flux dont vous avez besoin.

Au fur et à mesure de l'évolution de vos spécifications, vous pourrez remplacer ces composants orientés flux autant que cela sera nécessaire sans les risques liés aux frameworks dogmatiques.

Nous pouvons utiliser des modules de streaming pour réutiliser la même logique de rendu pour les client/serveur html. Cette approche est indexable, compatible SEO, et nous apporte les mises à jour en temps réel.

Notre renderer prend des lignes de json en entrée et retourne des chaînes html en sortie. Le texte, l'interface universelle !

Render.js :

var through = require('through');
var hyperglue = require('hyperglue');
var fs = require('fs');
var html = fs.readFileSync(__dirname + '/static/row.html', 'utf8');

module.exports = function () {
    return through(function (line) {
        try { var row = JSON.parse(line) }
        catch (err) { return this.emit('error', err) }
        
        this.queue(hyperglue(html, {
            '.who': row.who,
            '.message': row.message
        }).outerHTML);
    });
};

Nous pouvons utiliser brfs pour appeler à la volée fs.readFileSync() pour le code navigateur et hyperglue pour mettre à jour le html basé sur des sélecteurs css. Vous n'avez pas nécessairement besoin d'utiliser hyperglue ici ; tout ce qui peut retourner une chaîne avec du html à l'intérieur marchera.

Le fichier row.html utilisé n'est qu'un simple stub (souche) :

row.html :

<div class="row">
  <div class="who"></div>
  <div class="message"></div>
</div>

Le serveur utilisera juste slice-file pour que tout reste simple. slice-file est à peine plus évolué que la célèbre API tail/tail -f avec en plus des interfaces en osmose avec les bases de données relationnelles et NoSQL comme CouchDB

server.js :

var http = require('http');
var fs = require('fs');
var hyperstream = require('hyperstream');
var ecstatic = require('ecstatic')(__dirname + '/static');

var sliceFile = require('slice-file');
var sf = sliceFile(__dirname + '/data.txt');

var render = require('./render');

var server = http.createServer(function (req, res) {
    if (req.url === '/') {
        var hs = hyperstream({
            '#rows': sf.slice(-5).pipe(render())
        });
        hs.pipe(res);
        fs.createReadStream(__dirname + '/static/index.html').pipe(hs);
    }
    else ecstatic(req, res)
});
server.listen(8000);

var shoe = require('shoe');
var sock = shoe(function (stream) {
    sf.follow(-1,0).pipe(stream);
});
sock.install(server, '/sock');

La première partie du serveur prend en charge la route / et envoie les cinq dernières lignes depuis data.txt vers la div #rows.

La seconde partie du serveur prend en charge les mises à jours en temps réel vers #rows en utilisant shoe, une simple prothèse d'émulation de websocket .

Ensuite nous pouvons écrire du code navigateur simple pour peupler les mises à jour temps réel depuis shoe vers la div #rows

var through = require('through');
var render = require('./render');

var shoe = require('shoe');
var stream = shoe('/sock');

var rows = document.querySelector('#rows');
stream.pipe(render()).pipe(through(function (html) {
    rows.innerHTML += html;
}));

Reste à compiler avec browserify et brfs  :

$ browserify -t brfs browser.js > static/bundle.js

et le tour est joué ! Maintenant, nous pouvons peupler data.txt avec des données inutiles :

$ echo '{"who":"substack","message":"beep boop."}' >> data.txt
$ echo '{"who":"zoltar","message":"COWER PUNY HUMANS"}' >> data.txt

puis, lançons le serveur :

$ node server.js

et rendons-nous à l'URL http://localhost:8000 pour y voir notre contenu. Si nous en ajoutons :

$ echo '{"who":"substack","message":"oh hello."}' >> data.txt
$ echo '{"who":"zoltar","message":"HEAR ME!"}' >> data.txt

alors la page s'actualise automatiquement en temps réel avec la mise à jour, hourra !

Maintenant, nous utilisons exactement la même logique de rendu sur le client comme sur le serveur pour présenter du contenu compatible SEO et indexable en temps réel. Hourra !