stream.pipeline în Node.js

import { pipeline } from "node:stream/promises";
await pipeline(sursă, ...transformări, destinație);

Ce face stream.pipeline

Stream-urile prelucrează datele bucată cu bucată, nu tot fișierul deodată. Așa poți comprima sau copia un fișier de câțiva GB cu foarte puțină memorie. pipeline leagă o sursă (stream de citire), zero sau mai multe transformări și o destinație (stream de scriere).

Față de vechiul sursa.pipe(destinatie), pipeline are două avantaje: dacă oricare stream dă eroare, le închide pe toate și îți dă eroarea, iar varianta din node:stream/promises se poate aștepta cu await. Respectă și presiunea inversă (backpressure): nu citește mai repede decât poate scrie destinația.

Între stream-uri poți pune și o funcție generator asincronă (async function*) care primește bucățile și produce altele: este cel mai simplu mod de a scrie o transformare.

Parametri

Parametru	Descriere
sursă	Un stream de citire sau orice iterabil (inclusiv un generator).
transformări	Stream-uri Transform sau funcții generator asincrone.
destinație	Un stream de scriere (sau o funcție asincronă care consumă datele).
{ signal }	Opțional, ca ultim argument: anulează tot lanțul.

Ce returnează

O promisiune care se rezolvă când s-a terminat scrierea.

Exemple

Exemplul 1: Comprimarea unui fișier

import { createReadStream, createWriteStream, writeFileSync, statSync } from "node:fs";
import { createGzip } from "node:zlib";
import { pipeline } from "node:stream/promises";

writeFileSync("jurnal.txt", "comandă nouă din Iași\n".repeat(5000));

await pipeline(
  createReadStream("jurnal.txt"),
  createGzip(),
  createWriteStream("jurnal.txt.gz"),
);
console.log("Original:", statSync("jurnal.txt").size, "octeți");
console.log("Comprimat mai mic de 1%:", statSync("jurnal.txt.gz").size < 1200);

Rezultat

Original: 125000 octeți
Comprimat mai mic de 1%: true

Exemplul 2: Transformare cu un generator

Generatorul primește bucăți de text și le trimite mai departe cu litere mari.

import { createReadStream, createWriteStream, writeFileSync, readFileSync } from "node:fs";
import { pipeline } from "node:stream/promises";

writeFileSync("orase.txt", "iași\ncluj\nbrașov\n");

await pipeline(
  createReadStream("orase.txt", "utf8"),
  async function* (sursa) {
    for await (const bucata of sursa) yield bucata.toUpperCase();
  },
  createWriteStream("orase-mari.txt"),
);
console.log(readFileSync("orase-mari.txt", "utf8"));

Rezultat

IAȘI
CLUJ
BRAȘOV

Exemplul 3: Erorile ajung într-un singur loc

import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";

try {
  await pipeline(createReadStream("lipsa.csv"), createWriteStream("copie.csv"));
} catch (eroare) {
  console.log("Copierea a eșuat:", eroare.code);
}

Rezultat

Copierea a eșuat: ENOENT

Greșeli frecvente

• Folosești a.pipe(b).pipe(c) fără tratarea erorilor: o eroare în a nu închide c, iar fișierele rămân deschise. Folosește pipeline.
• Importi pipeline din node:stream și pui await în fața lui: varianta de acolo cere callback. Pentru await, importă din node:stream/promises.
• Citești tot fișierul cu readFile ca să-l copiezi sau comprimi: la fișiere mari consumă multă memorie. Stream-urile citesc pe bucăți (implicit 64 KB).

Comenzi înrudite

• stream.Readable: Stream-ul din care citești date bucată cu bucată: fișiere, cereri HTTP, stdin. Readable.from îl creează din orice iterabil.
• stream.Transform: Stream care primește date, le modifică și le dă mai departe: comprimare, criptare, împărțire pe rânduri, conversii.
• zlib (gzip, brotli): Comprimă și decomprimă date cu gzip, deflate sau Brotli: fișiere .gz, răspunsuri HTTP mai mici.
• fs.readFile: Citește tot conținutul unui fișier: ca text dacă dai codificarea („utf8”), altfel ca Buffer cu octeții bruți.