Node.
JS
Le module stream (flux)
Définition
Le stream en [Link] (flux de données) est un mécanisme qui permet de lire ou écrire
des données progressivement, morceau par morceau (chunks), au lieu de tout charger en
mémoire d’un coup.
C’est très important pour les fichiers volumineux, les transferts réseau, la vidéo, etc., car
il améliore la performance et la consommation mémoire.
Au lieu de charger tout un fichier de 2 Go en mémoire, puis le traiter. On le lire morceau
par morceau.
Comme boire de l’eau avec une paille au lieu d’un verre entier.
Prof. Mohamed NABIL 71
[Link]
Le module stream (flux)
Les streams offrent trois avantages:
Gain de mémoire : Lire un gros fichier et tout charger en RAM implique un gros risque de mémoire
saturée. Avec streams la lecture/écriture par petits morceaux permet de garder la mémoire stable
avec une meilleure performance liée à la mémoire.
Temps de réponse amélioré : Les streams permettent un traitement immédiat des données.
Lorsqu’un bloc de données arrive, il peut être traité sans attendre la réception de la totalité des
données. Cela réduit la latence et améliore la réactivité globale de votre application.
Scalabilité pour le traitement en temps réel : En traitant les données par blocs, les streams [Link]
peuvent gérer efficacement de grandes quantités de données avec des ressources limitées. Cette
scalabilité rend les flux idéaux pour les applications qui traitent de gros volumes de données en
temps réel.
Ces avantages font les streams un outil puissant pour la création d'applications [Link]
performantes et évolutives, notamment lors du traitement de grands ensembles de données ou de
données en temps réel.
Prof. Mohamed NABIL 72
[Link]
Le module stream (flux)
Importer le module
//const stream = require('stream');
import stream from 'stream';
Ou importer des classes spécifiques :
//const { Readable, Writable, Duplex, Transform, pipeline } = require('stream');
import { Readable, Writable, Duplex, Transform, pipeline } from 'stream';
Prof. Mohamed NABIL 73
[Link]
Le module stream (flux)
Les classes principales
Le module stream de [Link] repose sur quatre classes de base fondamentales et une
classe utilitaire simplifiée. Toutes ces classes héritent de la classe EventEmitter.
Classe Rôle Exemple d'utilisation
Lecture d'un fichier
[Link] Source de données depuis laquelle on lit.
([Link]).
Destination vers laquelle on écrit des Écriture dans un fichier
[Link]
données. ([Link]).
Flux qui est à la fois Readable et Writable
[Link] Sockets réseau (TCP).
(comme un téléphone).
Type de Duplex qui modifie la donnée Compression de données (zlib)
[Link]
pendant son passage. ou chiffrement.
Prof. Mohamed NABIL 74
[Link]
Le module stream (flux)
La classe Readable
Readable est la classe utilisée pour lire séquentiellement une source de données.
Dans l'API [Link], on trouve généralement des exemples de flux Readable :
[Link] pour la lecture de fichiers,
[Link] pour la lecture de requêtes HTTP
[Link] pour la lecture depuis l'entrée standard.
Prof. Mohamed NABIL 75
[Link]
Le module stream (flux)
La classe Readable
Un readable stream utilise plusieurs méthodes et événements principaux :
on(eventName, listener) où listener est une fonction appelée quand l'événement se produit.
on('data') : Cet événement est déclenché dès que des données sont disponibles dans le stream
(morceau (un chunk) est disponible dans le tampon de lecture).
on('end') : Émis lorsqu'il n'y a plus de données à lire dans le stream. Il signale la fin de la
transmission des données. Cet événement est déclenché uniquement lorsque toutes les
données du flux ont été consommées.
on('readable') : Cet événement est déclenché lorsque des données sont disponibles en lecture
dans le stream ou lorsque la fin du stream est atteinte. Il permet une lecture des données plus
contrôlée si nécessaire.
Prof. Mohamed NABIL 76
[Link]
Le module stream (flux)
La classe Readable
on('close') : Cet événement est émis lorsque le stream et ses ressources sous-jacentes
sont fermés et indique qu'aucun autre événement ne sera émis.
on('error') : Cet événement peut être émis à tout moment, signalant une erreur de
traitement. Un gestionnaire d'événements peut être utilisé pour éviter les exceptions
non interceptées.
Prof. Mohamed NABIL 77
[Link]
Le module stream (flux)
La classe Readable: mode flux (flowing mode): Les données arrivent automatiquement
dès qu’elles sont disponibles. [Link] pousse les chunks vers ton callback. Tu ne contrôles
pas la vitesse de lecture.
class MyStream extends Readable {
import { Readable } from 'stream'; #count = 0;
_read(size) {
const readable = new Readable({ [Link](':-)');
read(size) { if (++this.#count === 5) {
[Link]('Bonjour '); [Link](null);
[Link]('le monde'); }
[Link](null); // fin du flux }
} }
});
const stream = new MyStream();
[Link]('data', chunk => {
[Link]([Link]()); [Link]('data', chunk => {
}); [Link]([Link]());
});
Prof. Mohamed NABIL 78
[Link]
import { Readable } from 'stream';
//crée une classe qui hérite des capacités de lecture de [Link] Streams.
class MyStream extends Readable {
//une propriété privée qui suit le nombre d'éléments générés.
#count = 0;
//C'est le moteur du flux. [Link] l'appelle automatiquement dès qu'il a besoin de données.
_read(size) {
//Envoie la chaîne de caractères dans la file d'attente (le buffer).
[Link](':-)');
//Incrémente le compteur à chaque passage.
if (++this.#count === 5) {
//Le signal critique pour dire au flux que la génération est terminée.
// Sans cela, le flux resterait ouvert indéfiniment.
[Link](null);
}
}
}
const stream = new MyStream();
//active le mode "flowing". Chaque fois qu'un push est effectué dans _read, // l'événement est
déclenché et affiche :-) dans la console.
[Link]('data', chunk => {
[Link]([Link]());
}); Prof. Mohamed NABIL 79
[Link]
Le module stream (flux)
La classe Readable: mode lecture contrôlée (paused / non-flowing mode)
L’événement readable signifie qu'il y a des données disponibles à lire; mais c’est vous qui doivent
appeler à [Link]() pour les lire. Nous contrôlons quand et combien lire. Cet événement est
plus complexe, mais offre de meilleures performances pour certaines applications en permettant
un contrôle explicite du moment où les données sont lues à partir du stream.
L'événement readable
const stream = new MyStream(); n'est pas émis à chaque
fois que vous appelez
[Link]('readable', () => { [Link]() car il se
[Link]('>> readable event'); déclenche uniquement
let chunk; lorsque de nouvelles
while ((chunk = [Link]()) !== null) {
[Link]([Link]()); // Process the chunk données arrivent dans un
} tampon vide ou après que
}); vous ayez consommé
[Link]('end', () => [Link]('>> end event')); toutes les données
Prof. Mohamed NABIL disponibles. 80
[Link]
Le module stream (flux)
La classe Readable: mode lecture contrôlée (paused / non-flowing mode)
class MyStream extends Readable {
#count = 0;
_read(size) {
setImmediate(() => {
[Link](':-)');
if (++this.#count === 5) {
[Link](null);
}
});
}
}
const stream = new MyStream();
[Link]('readable', () => {
[Link]('>> readable event');
let chunk;
while ((chunk = [Link]()) !== null) {
[Link]([Link]()); // Process the chunk
}
});
[Link]('end', () => [Link]('>> end event')); Prof. Mohamed NABIL 81
[Link]
Le module stream (flux)
mode flux (flowing mode) VS mode lecture contrôlée (paused / non-flowing mode)
data readable
Mode flowing Mode paused
Données poussées automatiquement Vous appelez read()
Simple Plus de contrôle
Moins de contrôle mémoire Meilleure gestion mémoire
Usage courant Usage avancé
Prof. Mohamed NABIL 82
[Link]
Le module stream (flux)
La classe Writable (Flux en écriture)
Les Writable streams sont utiles pour créer des fichiers, télécharger des données ou
toute tâche impliquant la sortie séquentielle de données.
Un Writable Stream est une destination vers laquelle vous envoyez des données morceau
par morceau (chunks). C'est l'opposé du Readable Stream : au lieu de recevoir (lire), nous
émettons (écrire) vers une cible externe.
Alors que les Readable streams fournissent la source des données, les Writable streams
dans servent de destination pour nos données.
Des exemples typiques de Writable streams dans l'API [Link] sont:
[Link] : Pour écrire des données dans un fichier local.
réponses HTTP (res) : pour envoyer une réponse au client.
[Link] : Pour afficher du texte dans le terminal (la console).
Prof. Mohamed NABIL 83
[Link]
Le module stream (flux)
La classe Writable (Flux en écriture)
Pour envoyer des données à un Writable Stream, on utilise principalement deux
méthodes :
.write() : Cette méthode permet d'écrire un bloc de données dans le Stream. Elle gère les
données en les mettant en mémoire tampon jusqu'à une limite définie (highWaterMark)
et renvoie une valeur booléenne indiquant si d'autres données peuvent être écrites
immédiatement. Elle envoie un morceau de données vers la destination.
.end() : Cette méthode signale la fin du processus d'écriture des données. Elle indique au
Stream de terminer l'opération d'écriture et d'effectuer éventuellement le nettoyage
nécessaire. Elle envoie le dernier morceau et signale que l'écriture est terminée.
Prof. Mohamed NABIL 84
[Link]
Le module stream (flux)
write() est le moteur des Writable Streams. C'est elle qui
permet d'envoyer des données dans un Steam (comme un
La classe Writable (Flux en écriture)
fichier, une réponse HTTP ou [Link]).
import { Writable } from 'node:stream'; chunk (Le morceau de donnée): C'est la donnée que vous
voulez écrire. Généralement une String ou un Buffer
const writable = new Writable({
(données binaires).
write(chunk, encoding, callback) {
[Link]([Link]()); Note : Si le flux est configuré en objectMode: true, le chunk
callback(); peut être n'importe quel objet JavaScript (un tableau, un
} JSON, etc.).
}); encoding (L'encodage): Cet argument n'est utilisé que si le
chunk est une String. Il indique comment convertir la chaîne
[Link]('Hello');
[Link](); en octets.
Valeur par défaut : 'utf8'.
Autres options : 'ascii', 'base64', 'hex'.
callback() signifie : « J'ai fini de traiter
Note : Si le chunk est un Buffer, cet argument est ignoré.
ce morceau, envoyez-moi le suivant ».
callback: C'est une fonction qui est appelée une fois que
Si vous oubliez cette ligne, le flux
l'écriture est terminée. Il reçoit un argument error si
restera bloqué indéfiniment après le
l'écriture a échoué. Prof. Mohamed NABIL 85
premier morceau de donnée.
[Link]
Le module stream (flux)
La classe Writable (Flux en écriture)
import { Writable } from 'node:stream';
class MyWritable extends Writable {
// _write prend 3 arguments :
// 1. chunk : le morceau de donnée reçu (Buffer ou String)
// 2. encoding : l'encodage si c'est une chaîne
// 3. callback : la fonction à appeler quand l'écriture est finie
_write(chunk, encoding, callback) {
const content = [Link]().toUpperCase();
[Link](`[Ecriture] : ${content}`);
callback();
}
}
const writer = new MyWritable();
// Utilisation
[Link]('Bonjour ');
[Link]('le monde !');
[Link](' - Fin du flux.'); Prof. Mohamed NABIL 86
[Link]
Le module stream (flux)
La classe Writable (Flux en écriture)
import { once } from 'node:events';
import { Writable } from 'node:stream';
class MyStream extends Writable {
constructor() {
super({ highWaterMark: 10 /* 10 bytes */ });
}
_write(chunk, encode, cb) {
[Link]([Link]().toUpperCase() + '\n', cb);
}
}
const stream = new MyStream();
for (let i = 0; i < 10; i++) {
const waitDrain = ;
if (waitDrain) {
[Link]('>> wait drain');
await once(stream, 'drain');
}
}
[Link]('world'); Prof. Mohamed NABIL 87
[Link]
Le module stream (flux)
La classe Duplex
Les duplex streams implémentent les interfaces de lecture et d'écriture.
Les duplex streams implémentent toutes les méthodes et tous les événements décrits
dans la section Readable et Writable streams .
Un bon exemple de duplex stream est la classe Socket du module net.
Prof. Mohamed NABIL 88
[Link]
Le module stream (flux)
La classe Duplex
import net from 'node:net';
// Create a TCP server import net from 'node:net';
const server = [Link](socket => {
[Link]('Hello from server!\n'); // Connect to the server at localhost:8080
const client = [Link]({ port:
[Link]('data', data => { 8080 }, () => {
[Link](`Client says: ${[Link]()}`); [Link]('Hello from client!\n');
}); });
// Handle client disconnection [Link]('data', data => {
[Link]('end', () => { [Link](`Server says: ${[Link]()}`);
[Link]('Client disconnected'); });
});
}); // Handle the server closing the connection
[Link]('end', () => {
// Start the server on port 8080 [Link]('Disconnected from server');
[Link](8080, () => { });
[Link]('Server listening on port 8080');
}); Prof. Mohamed NABIL 89
[Link]
Le module stream (flux)
La classe Transform
Les Transform streams sont des Duplex streams, où la sortie est calculée à partir de
l'entrée.
Comme leur nom l'indique, ils sont généralement utilisés entre un Readable stream et un
writable stream pour transformer les données lors de leur passage.
Outre toutes les méthodes et tous les événements des duplex streams, on trouve :
_transform : Cette fonction est appelée en interne pour gérer le flux de données entre les
parties de lecture et d'écriture.
Elle ne doit en aucun cas être appelée par le code de l'application.
Prof. Mohamed NABIL 90
[Link]
L'objet upper est une instance de la classe
Le module stream (flux) Transform. Contrairement à un Readable stream
ou Writable stream, il fait les deux : il reçoit des
La classe Transform données, les modifie, puis les renvoie.
transform(data, enc, cb) : C'est la fonction cœur
import { Transform } from 'node:stream'; qui s'exécute à chaque fois qu'un morceau de
donnée (chunk) arrive.
const upper = new Transform({ data : Le morceau de donnée reçu (un Buffer).
transform(data, enc, cb) { enc : L'encodage (si la donnée est une chaîne).
[Link]([Link]().toUpperCase()); cb : La fonction de rappel (callback) à appeler
cb(); impérativement pour signaler que le traitement
}, est fini.
}); [Link](...) : Cette méthode envoie la donnée
transformée vers la sortie du flux.
[Link]('bonjour'); [Link]('bonjour') : On envoie
[Link]('data', data => [Link]([Link]())); manuellement la chaîne 'bonjour' dans l'entrée
du flux. Cela déclenche immédiatement la
fonction transform.
[Link]('data', ...) : On écoute l'événement
'data'. Dès que la transformation est terminée et
que [Link]() est appelé, le résultat
('BONJOUR') est éjecté du flux et affiché dans la
Prof. Mohamed NABIL console. 91
[Link]
Le module stream (flux)
Les méthodes .pipe() et . pipeline()
Lorsqu'on travaille avec des flux de données, on souhaite généralement lire des données
d'une source et les écrire dans une destination, en ayant parfois besoin de transformer
les données entre les deux.
La méthode .pipe() concatène un readable stream à un writable stream (ou transform).
[Link](transform).pipe(writable);
Bien que cela paraisse simple, elle délègue la gestion des erreurs au programmeur, ce qui
complique sa mise en œuvre.
import { pipeline } from ‘node:stream’);
pipeline(readable, transform, writable, err => {if (err) [Link](err);});
Prof. Mohamed NABIL 92
[Link]
Le module stream (flux)
Les méthodes .pipe() et . pipeline()
import fs from 'node:fs';
import { Transform } from 'node:stream';
const upper = new Transform({
transform(data, enc, cb) {
[Link]([Link]().toUpperCase());
cb();
},
});
// Création des flux de lecture et d'écriture
const reader = [Link]([Link]);
const writer = [Link]('[Link]');
// Connexion des flux entre eux
[Link](upper).pipe(writer);
[Link]('finish', () => [Link]('Fichier transformé avec succès !'));
Prof. Mohamed NABIL 93
[Link]
Le module stream (flux)
[Link]((req, res) => {
Les méthodes .pipe() et . pipeline()
// Configuration des en-têtes de réponse
[Link](200, { 'Content-Type': 'text/html;
import fs from 'node:fs';
charset=utf-8' });
import { Transform } from 'node:stream'; // Pipeline : Lecture -> Transformation -> Réponse
import http from 'node:http'; const reader = [Link]('[Link]');
const upper = new Transform({ [Link](upper).pipe(res);
transform(chunk, encoding, callback) { [Link]('error', (err) => {
callback(null, [Link]()); [Link](err);
[Link]();
}
});
});
}).listen(3000, () => [Link]('Serveur sur
[Link]
Prof. Mohamed NABIL 94
,
[Link]
Le module stream (flux)
Les méthodes .pipe() et . pipeline()
const readStream =
import fs from 'node:fs'; [Link]([Link], {
import { Transform, pipeline } from 'node:stream'; highWaterMark: 1,
});
let errorCount = 0;
const upper = new Transform({ const writeStream = [Link];
transform(data, enc, cb) {
if (errorCount === 10) { pipeline(readStream, upper, writeStream, err => {
return cb(new Error('BOOM!')); if (err) {
} return [Link]('Pipeline error:');
errorCount++; }
[Link]([Link]().toUpperCase()); [Link]('Pipeline succeeded');
cb(); });
[Link]('close', () => {
},
[Link]('\nTransform stream closed');
}); });
Prof. Mohamed NABIL 95
[Link]
Le module stream (flux)
La méthode .pipe() import fs from 'fs';
La méthode pipeline() import { pipeline } from 'stream/promises';
possède également une await pipeline(
version async pipeline(), qui [Link]([Link]),
async function* (source) {
n'accepte pas de rappel mais for await (const chunk of source) {
renvoie plutôt une promesse yield [Link]().toUpperCase();
}
qui est rejetée si le pipeline },
échoue. [Link]
);
[Link]
Le module stream (flux)
Object Mode: Streams peuvent transporter des objets JS. Dans ce cas, l’option
highWaterMark fait référence au nombre d’objets, et non à leur taille en octets.
const r = new Readable({
objectMode: true,
read() {}
});
[Link]({ id: 1 });
[Link]
Le module stream (flux)
Backpressure (Gestion de la contre-pression)
Lors de l'utilisation de streams, il est crucial d'éviter que le producteur ne surcharge le
consommateur. À cette fin, le mécanisme de Backpressure est implémenté dans tous les
flux de l'API [Link], et il incombe aux développeurs de maintenir ce comportement.
Si la mémoire tampon de données dépasse le seuil `highWaterMark` ou si la file d'attente
d'écriture est occupée, la méthode `.write()` renvoie `false`.
Lorsqu'une valeur `false` est renvoyée, le système de Backpressure s'active.
Il suspend l'envoi de données du flux `Readable` entrant et attend que le consommateur
soit de nouveau prêt.
Une fois la mémoire tampon de données vidée, un événement `drain` est émis pour
reprendre le flux de données entrant.
[Link]
Le module stream (flux)
Backpressure (Gestion de la contre-pression)
if () {
[Link]();
}
[Link]('drain', () => [Link]());
Si le buffer dépasse une certaine limite (highWaterMark), write() renvoie false, indiquant : "Stop !
Je suis plein, ne m'envoie plus rien.«
Pour éviter de saturer la mémoire RAM, on met généralement le flux de lecture en pause
([Link]()) dès que le buffer d'écriture est plein.
resume() permet de relancer la lecture une fois que le danger est passé.
Le flux Writable émet l'événement 'drain' uniquement lorsque :
Son buffer était plein (write avait renvoyé false).
Il a fini de vider ce buffer et est à nouveau prêt à recevoir des données.