Transformer les données en flux
Objectif : Convertir les commandes lues en lignes CSV au fil de leur arrivée.
Un export de commandes peut contenir des centaines de milliers de lignes. Si vous construisez d’abord un grand tableau ou une chaîne CSV complète, la mémoire utilisée augmente avec le volume exporté.
Vous allez relier le parcours des commandes déjà défini dans src/services/exports.js à une transformation en CSV. Chaque commande sera convertie en ligne dès qu’elle est lue, sans conserver toutes les lignes précédentes.
Du parcours des commandes au flux CSV
Un flux lisible (Readable) fournit des morceaux de données que le consommateur peut lire progressivement. Un flux de transformation (Transform) reçoit ces morceaux, les convertit, puis émet le résultat correspondant. Ici, le premier flux produira les commandes du dépôt et le second convertira chacune en ligne CSV.
Le service d’export conserve son dépôt injecté et sa taille de lot. Il ajoute deux méthodes : creerFluxCommandes() fournit les enregistrements au fil du parcours, et creerFluxCsv() les transforme en texte CSV. Le parcours par curseur et l’ordre stable restent ceux de la leçon précédente.
Créer les flux dans le service
Remplacez le contenu de src/services/exports.js par le module complet ci-dessous. Il conserve parcourirCommandes(traiterCommande), utilisé pour parcourir les lots, et ajoute les flux Node.js. Les colonnes sont définies explicitement afin que l’export garde un format prévisible.
const { Readable, Transform } = require('node:stream');
function creerServiceExports({ depot, tailleLot }) {
if (!depot || typeof depot.lireLot !== 'function') {
throw new TypeError('Le dépôt doit fournir lireLot');
}
if (!Number.isInteger(tailleLot) || tailleLot < 1) {
throw new TypeError('La taille de lot doit être un entier positif');
}
async function parcourirCommandes(traiterCommande) {
if (typeof traiterCommande !== 'function') {
throw new TypeError('Le traitement doit être une fonction');
}
let apresId = null;
while (true) {
const commandes = await depot.lireLot({ apresId, limite: tailleLot });
if (!Array.isArray(commandes) || commandes.length > tailleLot) {
throw new TypeError('Le dépôt a retourné un lot invalide');
}
if (commandes.length === 0) return;
for (const commande of commandes) {
await traiterCommande(commande);
apresId = commande.id;
}
}
}
function echapperChampCsv(valeur) {
const texte = String(valeur ?? '');
if (/[",\r\n]/.test(texte)) {
return `"${texte.replace(/"/g, '""')}"`;
}
return texte;
}
function convertirCommandeEnCsv(commande) {
if (!commande || commande.id === undefined || commande.id === null) {
throw new TypeError('Chaque commande doit avoir un identifiant');
}
const champs = [
commande.id,
commande.clientId,
commande.statut,
commande.total
];
return `${champs.map(echapperChampCsv).join(',')}\r\n`;
}
function creerFluxCommandes() {
let erreurParcours;
const flux = new Readable({
objectMode: true,
read() {}
});
parcourirCommandes((commande) => {
if (!flux.push(commande)) {
return new Promise((resolve) => {
flux.once('resume', resolve);
});
}
}).then(() => {
flux.push(null);
}).catch((erreur) => {
erreurParcours = erreur;
flux.destroy(erreur);
});
flux.on('close', () => {
if (erreurParcours) return;
});
return flux;
}
function creerFluxCsv() {
const transformation = new Transform({
writableObjectMode: true,
transform(commande, encodage, rappel) {
try {
rappel(null, convertirCommandeEnCsv(commande));
} catch (erreur) {
rappel(erreur);
}
}
});
return { commandes: creerFluxCommandes(), csv: transformation };
}
return {
parcourirCommandes,
creerFluxCommandes,
creerFluxCsv
};
}
module.exports = { creerServiceExports };Le lecteur est en mode objet : il émet des commandes JavaScript et non des chaînes. La transformation accepte ces objets en entrée, puis émet des chaînes encodées en UTF-8. Elle échappe les champs nécessaires et termine chaque ligne par CRLF, séparateur de ligne usuel pour un fichier CSV.
Vérifier le résultat et les erreurs
Pour tester la conversion sans base de données, instanciez le service avec un dépôt qui retourne un lot puis un lot vide. Le code suivant est autonome et montre comment relier le lecteur à la transformation pour cet essai ; l’écriture finale vers une destination sera traitée dans la prochaine leçon.
const { creerServiceExports } = require('./src/services/exports');
const commandes = [
{ id: 101, clientId: 'C-7', statut: 'payée', total: 42.5 },
{ id: 102, clientId: 'C-8', statut: 'en attente, paiement', total: 18 }
];
let dejaLus = false;
const depot = {
async lireLot({ apresId, limite }) {
if (dejaLus) return [];
dejaLus = true;
return commandes
.filter((commande) => apresId === null || commande.id > apresId)
.slice(0, limite);
}
};
const service = creerServiceExports({ depot, tailleLot: 2 });
const { commandes: source, csv } = service.creerFluxCsv();
csv.on('data', (ligne) => process.stdout.write(ligne));
csv.on('error', (erreur) => {
console.error('Échec de transformation :', erreur.message);
process.exitCode = 1;
});
source.on('error', (erreur) => {
csv.destroy(erreur);
});
source.pipe(csv);101,C-7,payée,42.5
102,C-8,"en attente, paiement",18Une erreur du dépôt détruit le flux source ; une commande sans identifiant provoque une erreur dans la transformation. Dans les deux cas, l’événement error permet à l’appelant de détecter l’échec au lieu de produire silencieusement un export incomplet. Dans le serveur de production, le code appelant devra relier ces erreurs à la réponse et à la journalisation du service.
À retenir
- Un flux lisible fournit les commandes progressivement ; il s’appuie ici sur le parcours par lots déjà présent dans le service d’export.
- Un flux de transformation convertit chaque commande en texte CSV sans assembler le fichier complet en mémoire.
- Les champs contenant virgules, guillemets ou retours à la ligne doivent être échappés conformément au format CSV.
- Les erreurs du dépôt ou de conversion doivent remonter à l’appelant pour éviter de traiter un export incomplet comme réussi.
Dans la leçon suivante, vous relierez cette source et cette transformation à une destination en tenant compte du contrôle de flux des streams.