Pipes : connecter les streams entre eux

Le besoin de connecter les flux de donnees

Nous avons vu comment lire depuis des streams Readable et écrire dans des streams Writable. Souvent, l'objectif est de prendre les données issues d'une source (Readable) et de les diriger directement vers une destination (Writable), parfois en les transformant au passage (avec un stream Transform). Pensez à copier un fichier, télécharger un fichier depuis une URL et le sauvegarder sur disque, compresser des données avant de les envoyer sur le réseau, etc.

Manuellement, cela impliquerait d'écouter l'événement `data` (ou `readable`) sur le stream source, d'appeler `write()` sur le stream destination, et surtout, de gérer méticuleusement la contre-pression (backpressure) en surveillant la valeur de retour de `write()` et l'événement `drain` pour ne pas saturer la destination. Cela peut rapidement devenir complexe et source d'erreurs.

Heureusement, Node.js fournit une méthode élégante et puissante pour automatiser ce processus : la méthode `pipe()`.

Decouverte de la methode `pipe()`

La méthode `pipe()` est une méthode disponible sur tous les streams Readable. Sa fonction principale est de connecter la sortie d'un stream Readable à l'entrée d'un stream Writable (ou Duplex/Transform). Elle prend le stream de destination comme argument.

Sa syntaxe de base est d'une simplicité déconcertante :

readableSrc.pipe(writableDest);

Cette unique ligne de code met en place un mécanisme sophistiqué qui gère pour vous l'ensemble du transfert de données entre les deux streams.

Ce que `pipe()` fait automatiquement pour vous

Utiliser `pipe()` délègue plusieurs tâches complexes à Node.js, rendant votre code plus lisible et moins sujet aux erreurs :

• Transfert des données : `pipe()` lit automatiquement les données du `readableSrc` dès qu'elles sont disponibles.
• Ecriture des données : Il écrit ensuite ces données dans le `writableDest` en appelant sa méthode `write()`.
• Gestion de la contre-pression (Backpressure) : C'est l'un des avantages majeurs. Si le `writableDest` devient trop lent pour accepter les données (par exemple, `write()` retourne `false`), `pipe()` détecte cela et met automatiquement en pause le `readableSrc` pour éviter d'accumuler les données en mémoire. Lorsque le `writableDest` est de nouveau prêt (il émet l'événement `drain`), `pipe()` reprend automatiquement la lecture sur le `readableSrc`.
• Gestion de la fin du flux (`end`) : Lorsque le `readableSrc` a fini d'émettre des données (il émet l'événement `end`), `pipe()` appelle automatiquement la méthode `writableDest.end()`. Cela signifie que la destination sait quand l'écriture est terminée (sauf si l'option `{ end: false }` est utilisée).
• Gestion basique des erreurs : Par défaut, si une erreur (`'error'`) se produit sur le `readableSrc`, `pipe()` s'assure que le `writableDest` est détruit (sa méthode `destroy()` est appelée avec l'erreur). Cela aide à prévenir les fuites de ressources. Attention : l'inverse n'est pas vrai par défaut (une erreur sur le Writable ne détruit pas le Readable), d'où l'importance de `stream.pipeline` ou de gestionnaires d'erreurs explicites.

Exemples d'utilisation de `pipe()`

La simplicité de `pipe()` le rend idéal pour de nombreuses tâches courantes :

Copie de fichier :

const fs = require('fs');

const reader = fs.createReadStream('source.txt');
const writer = fs.createWriteStream('destination.txt');

reader.pipe(writer);

// Il est toujours bon d'écouter les erreurs individuellement
reader.on('error', err => console.error('Erreur lecture:', err));
writer.on('error', err => console.error('Erreur écriture:', err));
writer.on('finish', () => console.log('Copie terminée.'));

Répondre à une requête HTTP avec un fichier :

const http = require('http');
const fs = require('fs');

const server = http.createServer((req, res) => {
  const fileStream = fs.createReadStream('mon_fichier.html');

  // fileStream (Readable) pipe vers res (ServerResponse, Writable)
  fileStream.pipe(res);

  fileStream.on('error', (err) => {
    console.error(err);
    res.statusCode = 500;
    res.end('Erreur interne du serveur.');
  });
});

server.listen(3000, () => console.log('Serveur écoute sur le port 3000'));

Chainage des `pipe()` pour creer des pipelines

La méthode `pipe()` retourne le stream de destination sur lequel elle est appelée. Cette caractéristique permet de chaîner les appels à `pipe()` de manière fluide, créant ainsi des pipelines de traitement de données complexes. C'est particulièrement utile avec les streams Transform.

Par exemple, pour lire un fichier, le compresser en Gzip, puis l'écrire dans un autre fichier :

const fs = require('fs');
const zlib = require('zlib');

const reader = fs.createReadStream('fichier_volumineux.log');
const gzip = zlib.createGzip(); // Transform stream
const writer = fs.createWriteStream('fichier_volumineux.log.gz');

// Chaînage : reader -> gzip -> writer
reader.pipe(gzip).pipe(writer);

// Gestion des erreurs (simplifiée, idéalement utiliser stream.pipeline)
reader.on('error', err => console.error('Erreur lecture:', err));
gzip.on('error', err => console.error('Erreur compression:', err));
writer.on('error', err => console.error('Erreur écriture:', err));
writer.on('finish', () => console.log('Compression terminée.'));

Ce pipeline élégant lit le fichier source, le compresse à la volée, et écrit le résultat compressé, tout en gérant la contre-pression à chaque étape.

Considerations et alternative robuste : `stream.pipeline`

Bien que `pipe()` soit extrêmement pratique, il a une limitation principale : sa gestion des erreurs n'est pas entièrement symétrique. Comme mentionné, une erreur sur le stream Writable (ou un stream Transform au milieu du pipeline) ne détruit pas automatiquement les streams précédents dans la chaîne par défaut. Cela peut conduire à des fuites de ressources ou à des états incohérents.

Pour une gestion d'erreurs plus robuste et un nettoyage correct de tous les streams en cas de problème n'importe où dans le pipeline, il est fortement recommandé d'utiliser l'utilitaire `stream.pipeline()` (ou sa version Promesse `stream.promises.pipeline`).

const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');

pipeline(
  fs.createReadStream('fichier_volumineux.log'),
  zlib.createGzip(),
  fs.createWriteStream('fichier_volumineux.log.gz'),
  (err) => {
    if (err) {
      console.error('Le pipeline a échoué :', err);
    } else {
      console.log('Le pipeline s\'est terminé avec succès.');
    }
  }
);

En conclusion, `pipe()` est un outil fondamental pour connecter facilement les streams en Node.js, idéal pour les cas simples. Cependant, pour des pipelines plus complexes ou pour une robustesse maximale en production, privilégiez `stream.pipeline` qui offre une meilleure gestion des erreurs et du nettoyage des ressources.