Ce face stream.Transform
Un Transform este în același timp stream de scriere (primește date) și de citire (dă date transformate). zlib.createGzip() și crypto.createHash() sunt exemple din Node.js. Pe al tău îl creezi dând funcția transform.
transform(bucata, codificare, callback) este apelată pentru fiecare bucată. Trimiți rezultatul cu callback(null, rezultat) sau cu this.push(rezultat) (de mai multe ori, dacă din o bucată ies mai multe), apoi callback(). O eroare se trimite cu callback(eroare). flush(callback) rulează la final, ca să trimiți ce a rămas în așteptare.
Cu objectMode: true (sau readableObjectMode / writableObjectMode) poți transforma obiecte, nu doar text și Buffer. Pentru transformări simple, o funcție generator în pipeline este adesea mai scurtă.
Parametri
| Parametru | Descriere |
|---|---|
transform(bucată, cod, cb) |
Prelucrează o bucată. |
flush(cb) |
Opțional: rulează după ultima bucată. |
objectMode |
Lucrează cu obiecte în loc de octeți. |
Exemple
Exemplul 1: Fișier CSV transformat în obiecte
Transformarea ține minte rândul neterminat dintr-o bucată (rest) și îl completează cu bucata următoare. flush trimite ultimul rând.
import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
import { createReadStream, writeFileSync } from "node:fs";
writeFileSync("note.csv", "Ana,10\nIon,7\nMaria,9\nVlad,5");
let rest = "";
const randuri = new Transform({
readableObjectMode: true,
transform(bucata, _cod, callback) {
const linii = (rest + bucata).split("\n");
rest = linii.pop();
for (const linie of linii) this.push(linie);
callback();
},
flush(callback) {
if (rest) this.push(rest);
callback();
},
});
const elevi = [];
await pipeline(
createReadStream("note.csv", { encoding: "utf8", highWaterMark: 8 }),
randuri,
async function (sursa) {
for await (const linie of sursa) {
const [nume, nota] = linie.split(",");
elevi.push({ nume, nota: Number(nota) });
}
},
);
console.log(elevi.filter((e) => e.nota >= 9));
Rezultat
[ { nume: 'Ana', nota: 10 }, { nume: 'Maria', nota: 9 } ]
Exemplul 2: Transformare de obiecte cu eroare
Un stream nu poate fi refolosit după ce s-a terminat, de aceea cuTva() creează unul nou la fiecare apel. La a doua listă, eroarea oprește tot lanțul, iar „ouă” nu mai este procesat.
import { Transform, Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
const cuTva = () => new Transform({
objectMode: true,
transform(produs, _cod, callback) {
if (produs.pret < 0) return callback(new Error(`Preț negativ la ${produs.nume}`));
callback(null, { ...produs, pretTva: +(produs.pret * 1.21).toFixed(2) });
},
});
async function afiseaza(produse) {
try {
await pipeline(Readable.from(produse), cuTva(), async function (sursa) {
for await (const p of sursa) console.log(p.nume, p.pretTva);
});
} catch (eroare) {
console.log("Oprit:", eroare.message);
}
}
await afiseaza([{ nume: "pâine", pret: 5 }, { nume: "lapte", pret: 8.5 }]);
await afiseaza([{ nume: "greșit", pret: -1 }, { nume: "ouă", pret: 12 }]);
Rezultat
pâine 6.05
lapte 10.29
Oprit: Preț negativ la greșit
Greșeli frecvente
- Uiți să apelezi
callback()întransform: stream-ul se blochează după prima bucată și programul se termină fără eroare, dar fără rezultat. - Presupui că fiecare bucată conține rânduri întregi: bucățile se taie oriunde. Păstrează restul neterminat, ca în primul exemplu, sau folosește
readline. - Trimiți obiecte printr-un stream fără
objectMode: primeștiERR_INVALID_ARG_TYPE, pentru că stream-urile obișnuite acceptă doar șiruri șiBuffer.
Comenzi înrudite
stream.pipeline: Leagă stream-uri în lanț (citire → transformare → scriere), cu tratarea erorilor și închiderea tuturor la final.stream.Readable: Stream-ul din care citești date bucată cu bucată: fișiere, cereri HTTP, stdin. Readable.from îl creează din orice iterabil.zlib (gzip, brotli): Comprimă și decomprimă date cu gzip, deflate sau Brotli: fișiere .gz, răspunsuri HTTP mai mici.readline (citire pe rânduri): Citește un fișier sau tastatura rând cu rând și pune întrebări utilizatorului în terminal (question).