Maîtriser les flux de données sous Node : partie 2

Si vous avez déjà utilisé le module RequestRequest, vous avez sans doute remarqué que son appel retourne un objet de flux synchrone que vous pouvez utiliser directement. Pour mieux comprendre, ce tutoriel explique comment vous devriez normalement gérer les réponses HTTP :

var http = require('http');

http.get('http://www.google.com', function onResponse(response) {
  response.pipe(destinationStream);
});

Comparons cet exemple à celui de l'utilisation du module Request :

var request = require('request');

request('http://www.google.com').pipe(destinationStream);

Il est plus facile à comprendre, plus court et nécessite un niveau d'indentation en moins. Dans cet article, je vais vous expliquer comment cela se passe pour que vous puissiez faire de même.

Tout d'abord, il est extrêmement important de comprendre le fonctionnement de l'API stream. Si ce n'est pas encore le cas, jetez un œil à la documentationAPI stream de l'API, je vous promets que ce n'est pas trop long.

Nous allons jeter un œil aux flux de lecture. Un flux de lecture peut être suspendu (paused()) et redémarré (resume()). Si nous utilisons un autre objet pour le représenter temporairement lorsqu'il n'est pas disponible, la chose raisonnable à faire serait de garder une propriété pause sur cet objet, mise à jour correctement quand pause() et resume() sont appelés. Certains flux de lecture ont également des méthodes destroy() et setEncoding(). Encore une fois, la première chose qui pourrait venir à l'esprit est de garder les propriétés destroyed et encoding sur le flux temporaire.

Mais, tous les flux de lecture ne sont pas créés de la même manière. Certains pourraient avoir plus de méthodes ou ne pas avoir une méthode destroy(). La technique la plus fiable que j'ai trouvée est de regarder le prototype du flux, d'effectuer une itération sur les fonctions y compris celles qui sont héritées, et mettre en mémoire tous les appels à celles-ci jusqu'à ce que le flux réel soit disponible. Cela fonctionne pour les méthodes write() et end() d'un flux en écriture, ainsi que pour les méthodes de l'émetteur d'événements telles que on().

Les méthodes des flux standards ne retournent rien à l'exception de write() qui retourne la valeur false si la mémoire d'écriture est pleine. Dans ce cas, on obtient false aussi longtemps que le flux réel n'est pas disponible.

Un autre cas particulier est celui de pipe(). Chaque méthode de flux de lecture fonctionne de la même manière. Elle n'a pas besoin d'être réécrite ou en file d'attente. Lorsque la méthode pipe() est appelée, elle écoute les événements des flux de source et de destination. Elle écrit dans le flux de destination chaque fois que les données sont émises à partir de la source et elle s'arrête et reprend la lecture de la source selon les besoins. Nous avons déjà en file d'attente les appels aux méthodes héritées de l'émetteur de l'événement.

Qu'en est-il de l'émission d'un événement avant que le flux de la source réelle soit disponible ? Vous ne pouvez pas le faire si vous mettez en file d'attente les appels à emit(). Les événements se déclencheraient uniquement lorsque le flux réel sera disponible. Si vous êtes perfectionniste, vous pouvez examiner ce cas très rare et trouver une solution.

StreamifyStreamify fait tout cela pour vous, alors vous n'avez pas à gérer les complexités mais juste à bénéficier des avantages d'une API plus agréable. Notre exemple HTTP précédent peut être réécrit pour fonctionner comme Request.

var http = require('http');
var streamify = require('streamify');

var stream = streamify();
http.get('http://www.google.com', function onResponse(response) {
  stream.resolve(response);
});

// "stream" peut déjà être chaîné
stream.pipe(destinationStream);

Vous pensez que c'est inutile puisque Request existe déjà et qu'il le fait déjà. N'oubliez pas que Request n'est qu'un module qui gère un type de flux. Ceci peut être utilisé avec n'importe quel type de flux de données qui n'est pas immédiatement disponible dans la boucle actuelle d'itération des événements.

Vous pourriez même faire quelque chose de fou comme ceci :

var http = require('http');
var fs = require('fs');
var streamify = require('streamify');

function uploadToFirstClient() {
  var stream = streamify({ superCtor: http.ServerResponse });

  var server = http.createServer(function onRequest(request, response) {
    response.writeHead(200);
    stream.resolve(response);
  }).listen(3000);

  stream.on('pipe', function onpipe(source) {
    source.on('end', server.close.bind(server));
  });

  return stream;
}

fs.createReadStream('/path/to/myfile').pipe(uploadToFirstClient);

Dans l'exemple précédent, j'ai utilisé le module HTTP standard de Node, mais il pourrait facilement être remplacé par Request. Streamify fonctionne très bien avec Request.

Streamify vous aide également lorsque vous avez besoin de faire plusieurs demandes avant que le flux que vous attendez soit disponible :

var request = require('request');
var streamify = require('streamify');

module.exports = function myModule() {
  var stream = streamify();

  request.get('http://somesite.com/authenticate', function onAuthenticate(err, response) {
    if (err) return stream.emit('error', err);
    
    var options = { uri: 'http://somesite.com/listmyfiles', json: true };
    request.get(options, function onList(err, result) {
      if (err) return stream.emit('error', err);
      stream.resolve(request.get('http://somesite.com/download/' + result.file));
    });
  });

  return stream;
};

Cela fonctionne à merveille pour n'importe quel cas d'usage dans lequel nous voulons travailler avec un flux qui sera utilisé plus tard, mais qui est précédé d'une ou plusieurs opérations asynchrones.

Partie 1 : cette partie parle du modèle courant, du support d'autres flux et de streamin.

Cet article a été publié avec l'aimable autorisation de Roly Fentanes pour l'article original Mastering Node Streams: Part 2 publié sur le site DailyJS.
Je remercie également ClaudeLELOUP pour sa relecture attentive et assidue.