`stream.Transform` în Node.js

Evenimente, stream-uri și date binare Node.js 0.9.4; constructorul simplificat din 1.2

Stream care primește date, le modifică și le dă mai departe: comprimare, criptare, împărțire pe rânduri, conversii.

Sintaxă

import { Transform } from "node:stream";
new Transform({
  transform(bucata, codificare, callback) { callback(null, rezultat); },
  flush(callback) { ... }
})

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 }]);