Node.js pipeline va stream xatolari: pipe, finished va bekor qilish

Qisqacha: a.pipe(b) ma'lumotni uzatadi, lekin xatoni uzatmaydi. Manba a buzilsa, b ochiq qoladi. Uni kutayotgan kod esa abadiy kutadi. stream/promises dagi pipeline(a, b, c) esa birinchi xatoda hamma oqimni yopadi va Promise'ni rad etadi — uni try/catch bilan ushlaysiz. finished(oqim) bitta oqim tugashini kutadi, signal esa butun zanjirni tashqaridan to'xtatadi.

Bu darsda

• .pipe() xatoni nega uzatmasligini haqiqiy chiqish bilan ko'rsata olasiz.
• pipeline ni stream/promises dan olib, xatoni try/catch bilan ushlay olasiz.
• finished bilan bitta oqim tugashini (yoki buzilishini) kuta olasiz.
• AbortSignal va destroy() bilan oqimlar zanjirini to'xtata olasiz.
• Xato bo'lganda yarim yozilgan faylni qoldirmaslikni bilasiz.

Oldin bilishingiz kerak: Stream nima: to'rt turi va rejimlari, Backpressure, Bekor qilish: AbortController, Xavfsiz fayl operatsiyalari.

1. Nega bu kerak?

Oldingi darsda backpressure bilan tanishdingiz: tez manba sekin qabul qiluvchini bosib ketmasligi kerak. U yerda pipe() va pipeline buni o'zi boshqarishini ko'rdingiz. Bugun ikkinchi savol: zanjirning o'rtasida xato chiqsa nima bo'ladi?

valyuta CLI'da shunday buyruq bo'ladi: valyuta tahlil eur.csv.gz. U siqilgan CSV faylni ochadi, ochib (gunzip) qatorlarni o'qiydi. Foydalanuvchi fayl nomini xato yozdi — eur.csv.gz o'rniga euro.csv.gz. Dastur nima qilishi kerak? To'g'ri javob: "fayl topilmadi" deb chiqib ketishi.

Kanonni yozganimizda dastlabki kodda fayl.pipe(gunzip) bor edi. Test yo'q fayl bilan ishga tushdi — va tugamadi. Xato chiqmadi, natija ham chiqmadi, dastur shunchaki kutib turaverdi. Bu darsning markaziy topilmasi shu: .pipe() xatoni keyingi oqimga uzatmaydi.

Hayotiy o'xshatish: suv quvuri uchta bo'lakdan iborat — bak, filtr, kran. Bak yorilib ketdi. Oddiy quvurda kran buni bilmaydi: u ochiq turaveradi, siz esa stakan bilan suv kutasiz. pipeline esa aqlli quvur: bitta bo'lak yorilsa, hamma jo'mrak yopiladi va sizga "bak yorildi" degan xabar keladi.

2. .pipe() xatoni uzatmaydi

2.1 Tajriba: fayl yo'q

Keling, buni o'z ko'zimiz bilan ko'ramiz. Mavjud bo'lmagan yoq.csv.gz faylni ochamiz va ochuvchiga (createGunzip) ulaymiz. Ikkala oqimga ham error tinglovchisini qo'yamiz — kim xatoni bilishini ko'rish uchun.

import { createReadStream } from "node:fs";
import { createGunzip } from "node:zlib";

const fayl = createReadStream("yoq.csv.gz");
const ochuvchi = createGunzip();
fayl.on("error", (x) => console.log("fayl xatosi:", x.code));
ochuvchi.on("error", (x) => console.log("gunzip xatosi:", x.code));
fayl.pipe(ochuvchi);

setTimeout(() => {
  console.log("fayl yopildimi:", fayl.destroyed);
  console.log("gunzip yopildimi:", ochuvchi.destroyed);
}, 100);

Konsolda:

fayl xatosi: ENOENT
fayl yopildimi: true
gunzip yopildimi: false

Uch narsaga qarang. Birinchi: xatoni faqat fayl oldi (ENOENT — "error no entry", fayl yo'q). Ikkinchi: gunzip xatosi qatori umuman chiqmadi. Uchinchi: fayl yopildi (destroyed: true), ochuvchi esa ochiq qoldi (false).

pipe() ning vazifasi tor: data hodisasini olib write() ga beradi, end kelsa end() chaqiradi. Xato kelsa — hech narsa qilmaydi. Manba buzilganda end ham kelmaydi. Demak ochuvchi "ma'lumot hali keladi" deb kutib turaveradi.

2.2 Kutayotgan kod abadiy kutadi

Endi ochuvchi dan for await bilan o'qiymiz (asinxron iteratsiya). Manbaga xatoni yozib qo'yadigan tinglovchi bor — xuddi "logga yozib qo'ydik, bo'ldi" degan odatiy kod kabi. Yarim soniyadan keyin tekshirib, dasturni o'zimiz to'xtatamiz:

import { createReadStream } from "node:fs";
import { createGunzip } from "node:zlib";

const fayl = createReadStream("yoq.csv.gz");
fayl.on("error", (x) => console.log("log:", x.code));
const oqim = fayl.pipe(createGunzip());
setTimeout(() => {
  console.log("0.5 s o'tdi — for await hali kutyapti");
  process.exit(0);
}, 500);
for await (const bolak of oqim) console.log(bolak.length);
console.log("tugadi");

Konsolda:

log: ENOENT
0.5 s o'tdi — for await hali kutyapti

tugadi hech qachon chiqmaydi. Bizda taymer bor edi, shuning uchun dasturni o'zimiz to'xtatdik. Haqiqiy CLI'da taymer yo'q — u terminalda qotib qoladi. Test ichida esa test "osilib" qoladi, chunki test yuguruvchi (runner) boshqa ishlar bilan event loop'ni tirik ushlab turadi.

Taymerni olib tashlasak nima bo'ladi? Biz ishga tushirdik (Node 24.21.0, Windows 11):

o'qishni boshladik
log: ENOENT
Warning: Detected unsettled top-level await at file:///C:/Users/…/a5.mjs:8
for await (const bolak of oqim) console.log(bolak.length);
                 ^

Chiqish kodi — 13. Node event loop'da boshqa ish qolmaganini ko'radi va "yuqori darajadagi await hech qachon tugamadi" deb ogohlantiradi. Bu xato emas, ogohlantirish: dastur nima uchun to'xtaganini aytmaydi. Foydalanuvchi "fayl yo'q" degan aniq xabarni ko'rmaydi.

2.3 Tinglovchi umuman bo'lmasa

Manbada error tinglovchisi yo'q bo'lsa-chi? Unda Node EventEmitter qoidasi ishlaydi: tinglovchisiz error hodisasi — butun jarayonni yiqitadi. Biz ishga tushirdik:

node:events:505
    throw er; // Unhandled 'error' event
    ^

Error: ENOENT: no such file or directory, open 'C:\Users\…\yoq.csv.gz'
Emitted 'error' event on ReadStream instance at:

Unhandled 'error' event — "ishlov berilmagan 'error' hodisasi". Bu holat hech bo'lmasa halol: dastur yiqiladi va sabab ko'rinadi. Lekin server ichida bunday xato butun jarayonni o'ldiradi.

Uchala holatni jamlaymiz:

Diqqat: Uchala holatda ham keyingi oqimlar yopilmaydi. Fayl deskriptorlari, siqish buferlari xotirada qoladi. Server bir kunda minglab so'rovni qayta ishlasa, bu sizish (leak) jarayonni sekin o'ldiradi.

Tekshirib ko'ring: Nega ochuvchi.on("error", ...) tinglovchisi «Tajriba: fayl yo'q» bo'limida ishlamadi?

3. pipeline: xatoni ushlaydigan quvur

3.1 stream/promises dagi pipeline

pipeline — oqimlarni ketma-ket ulaydigan funksiya (pipe ning kattasi). Uni node:stream/promises dan olamiz — u Promise qaytaradi va siz uni await qila olasiz. O'sha tajribani pipeline bilan takrorlaymiz:

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

const ochuvchi = createGunzip();
const chiqish = createWriteStream("natija.csv");
try {
  await pipeline(createReadStream("yoq.csv.gz"), ochuvchi, chiqish);
  console.log("tayyor");
} catch (xato) {
  console.log("pipeline xatosi:", xato.code);
  console.log("gunzip yopildimi:", ochuvchi.destroyed);
  console.log("chiqish yopildimi:", chiqish.destroyed);
}

Konsolda:

pipeline xatosi: ENOENT
gunzip yopildimi: true
chiqish yopildimi: true

Farq aniq. Xato bitta joyda — catch ichida — keldi. Zanjirdagi hamma oqim yopildi. Dastur kutib qolmadi.

pipeline ichkarida uch ish qiladi:

1. Oqimlarni pipe qiladi va backpressure'ni boshqaradi (oldingi darsdagidek).
2. Har bir oqimning error hodisasini tinglaydi.
3. Birinchi xatoda hammasiga destroy(xato) chaqiradi va Promise'ni shu xato bilan rad etadi.

Quyidagi diagrammada ikki quvurni solishtiring: yuqorida xato qayerda to'xtaydi, pastda qayerga boradi.

flowchart LR
  subgraph P["a.pipe(b).pipe(c)"]
    A1[fayl] -- "ma'lumot" --> B1[gunzip]
    B1 --> C1[yozish]
    A1 -. "xato shu yerda qoladi" .-> X1((?))
  end
  subgraph Q["pipeline(a, b, c)"]
    A2[fayl] --> B2[gunzip] --> C2[yozish]
    A2 -. xato .-> K["catch: hammasi yopiladi"]
  end

3.2 Callback shakli

pipeline ning eski shakli node:stream modulida turadi va oxirgi argument sifatida callback oladi. Eski kodda uni ko'p uchratasiz:

import { pipeline, Readable, PassThrough } from "node:stream";

const manba = Readable.from(["osh\n", "manti\n"]);
pipeline(manba, new PassThrough(), (xato) => {
  if (xato) console.log("xato:", xato.message);
  else console.log("hammasi o'tdi");
});

Konsolda:

hammasi o'tdi

PassThrough — kirganini o'zgartirmasdan chiqaradigan eng oddiy Transform oqim. Bu yerda u "o'rtadagi bo'g'in" o'rnida turibdi.

Ikki modulni adashtirish oson. node:stream dagi pipeline callback'siz chaqirilsa, darhol xato beradi:

import { pipeline, Readable, PassThrough } from "node:stream";

await pipeline(Readable.from(["a"]), new PassThrough());
console.log("tugadi");

Konsolda:

TypeError [ERR_INVALID_ARG_TYPE]: The "streams[stream.length - 1]" property must be of type function. Received an instance of PassThrough

Tarjimasi: "streams[stream.length - 1] xususiyati funksiya bo'lishi kerak. PassThrough nusxasi berildi". Nega aynan "oxirgi element" haqida gapiryapti? Callback shaklida oxirgi argument callback deb kutiladi. Siz esa u yerga oqim berdingiz. Qoida: await kerak bo'lsa — node:stream/promises dan import qiling.

Maslahat: Yangi kodda doim import { pipeline } from "node:stream/promises". try/catch, async/await bilan o'qish oson va xato yo'qolmaydi.

3.3 Generator bilan pipeline

pipeline ga oqim o'rniga async generator funksiya ham berish mumkin. Bunday funksiya oldingi bo'g'inni for await bilan o'qiydi va natijani yield qiladi. Bu Transform yozmasdan oraliq ishlov berishning eng qisqa yo'li:

import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

const kurslar = Readable.from(["12010", "12020", "12030"]);
let jami = 0;
await pipeline(
  kurslar,
  async function* (manba) {
    for await (const kurs of manba) yield Number(kurs) + 5;
  },
  new Writable({
    objectMode: true,
    write(son, _kod, tayyor) { jami += son; tayyor(); },
  }),
);
console.log("jami:", jami);

Konsolda:

jami: 36075

Generator ichida throw qilsangiz, pipeline uni boshqa xatolar kabi ushlaydi va hammasini yopadi. O'z Transform'ingizni yozish va generator bilan farqini keyingi darsda ko'ramiz.

Tekshirib ko'ring: pipeline dan keyin try/catch yozmasangiz va xato chiqsa, nima bo'ladi?

4. finished: bitta oqim tugashini kutish

Ba'zan sizda zanjir emas, bitta oqim bor. Masalan, oqimni boshqa kutubxona o'qiyapti, sizga esa faqat "tugadimi?" degan savolga javob kerak. Buning uchun finished bor. U oqim to'liq tugaganda bajariladi, buzilganda esa xato bilan rad etiladi:

import { Readable } from "node:stream";
import { finished } from "node:stream/promises";

const oqim = Readable.from(["osh\n", "lag'mon\n", "manti\n"]);
oqim.on("data", (b) => process.stdout.write("bo'lak: " + b));
await finished(oqim);
console.log("oqim tugadi");

const buzuq = new Readable({ read() {} });
setTimeout(() => buzuq.destroy(new Error("disk uzildi")), 10);
try {
  await finished(buzuq);
} catch (xato) {
  console.log("finished xatosi:", xato.message);
}

Konsolda:

bo'lak: osh
bo'lak: lag'mon
bo'lak: manti
oqim tugadi
finished xatosi: disk uzildi

Nega end hodisasini o'zimiz tinglamaymiz? Oqim end, finish, close yoki error bilan tugashi mumkin — qaysi biri kelishi oqim turiga bog'liq. finished hammasini hisobga oladi.

5. Bekor qilish: signal va destroy

5.1 AbortSignal bilan vaqt chegarasi

valyuta tarix buyrug'i bir oylik kursni kunma-kun so'raydi. Kanonda har so'rov ~1 soniya oldi va 34 kun 36.2 soniya davom etdi. Foydalanuvchi shuncha kutishni xohlamasa-chi? AbortController darsidagi signal bu yerda ham ishlaydi: pipeline ning oxirgi argumenti — sozlamalar obyekti.

Quyidagi misolda har "kun" 20 ms oladi, 30 kun esa ~600 ms. Biz esa faqat 150 ms beramiz (timers/promises dagi setTimeout):

import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { setTimeout as kut } from "node:timers/promises";

async function* kunlar() {
  for (let kun = 1; kun <= 30; kun++) {
    await kut(20); // har kun — sekin so'rov
    yield `2026-10-${String(kun).padStart(2, "0")}\n`;
  }
}
let soni = 0;
const sanovchi = new Writable({
  write(bolak, _kod, tayyor) { soni++; tayyor(); },
});
try {
  await pipeline(Readable.from(kunlar()), sanovchi, {
    signal: AbortSignal.timeout(150),
  });
} catch (xato) {
  console.log(xato.name, xato.code);
  console.log("yozilgan kunlar 30 dan kam:", soni < 30);
}

Konsolda:

AbortError ABORT_ERR
yozilgan kunlar 30 dan kam: true

Signal ishga tushganda pipeline hamma oqimga destroy chaqiradi va AbortError bilan rad etiladi. Aniq necha kun yozilgani vaqtga bog'liq, shuning uchun biz faqat "30 dan kam" ekanini tekshirdik.

Kanondagi oqimniYoz funksiyasi ham signal qabul qiladi va uni pipeline ga uzatadi — chaqiruvchi xohlasa, yozishni bekor qila oladi. Ctrl+C esa hozircha (holat 35 gacha) boshqacha ishlaydi: jarayon 130 kodi bilan darhol chiqadi (process obyekti). Ctrl+C ni shu signal ga ulash keyinroq, Built-in fetch darsida qo'shiladi.

5.2 destroy() — qo'lda to'xtatish

destroy(xato) — oqimni darhol yopish. Xato bersangiz, oqim error hodisasini chiqaradi; bermasangiz — jim yopiladi (close). pipeline ichidagi istalgan oqimni destroy qilsangiz, butun zanjir to'xtaydi — bu ham bekor qilish usuli.

Kanondagi tuzatish aynan destroy ga tayanadi. valyuta tahlil .gz faylni shunday ochadi:

export function faylniOch(fayl: string): Readable {
  const oqim = createReadStream(fayl);
  if (!fayl.endsWith(".gz")) return oqim;
  // .pipe() xatoni uzatmaydi: fayl yo'q bo'lsa gunzip abadiy kutardi
  const ochuvchi = createGunzip();
  oqim.on("error", (xato) => ochuvchi.destroy(xato));
  return oqim.pipe(ochuvchi);
}

Nega bu yerda pipeline emas? Chunki funksiya oqimni qaytaradi, uni keyin boshqa kod (readline) o'qiydi. pipeline esa zanjirni oxirigacha o'zi yurgizadi. Qaytariladigan oqim uchun pipe + qo'lda xato uzatish kerak bo'ldi. Bitta qator — ochuvchi.destroy(xato) — o'qish xatosini ochuvchiga o'tkazadi. Endi for await darhol ENOENT bilan to'xtaydi.

Maslahat: Node'da stream.compose(a, b) ham bor — bir nechta oqimdan bitta oqim yasaydi va xatoni uzatadi. Node hujjatiga ko'ra u uzoq vaqt experimental (tajribaviy) edi va faqat Node 24.19.0 da barqaror deb belgilandi. Kanon uni ishlatmadi — pipe + destroy har qanday Node 24 versiyasida ishlaydi. Async iteratorlar va Web Streams darsida unga qaytamiz.

Tekshirib ko'ring: oqim.on("error", (x) => console.log(x)) va oqim.on("error", (x) => ochuvchi.destroy(x)) — farqi nima?

6. Xatodan keyin nima qoladi

6.1 Yarim yozilgan fayl

pipeline oqimlarni yopadi, lekin diskdagi faylni o'chirmaydi. Quyidagi misolda 5 000 qatordan keyin noto'g'ri qator keladi:

import { createWriteStream, existsSync, statSync } from "node:fs";
import { Readable, Transform } from "node:stream";
import { pipeline } from "node:stream/promises";

async function* qatorlar() {
  yield "sana,kurs\n";
  for (let i = 1; i <= 5000; i++) yield `2026-10-23,${12000 + i}\n`;
  yield "2026-10-24,xato\n";
}
const tekshiruvchi = new Transform({
  transform(bolak, _kod, tayyor) {
    const kurs = String(bolak).trim().split(",")[1];
    if (kurs === "xato") return tayyor(new Error("kurs son emas"));
    tayyor(null, bolak);
  },
});
try {
  await pipeline(
    Readable.from(qatorlar()),
    tekshiruvchi,
    createWriteStream("kurs.csv"),
  );
} catch (xato) {
  console.log("xato:", xato.message);
}
console.log("fayl qoldimi:", existsSync("kurs.csv"));
console.log("bo'shmi:", statSync("kurs.csv").size === 0);

Konsolda:

xato: kurs son emas
fayl qoldimi: true
bo'shmi: false

Xato to'g'ri ushlandi, lekin diskda yarim kurs.csv qoldi. Ertaga kimdir uni ochadi va "to'liq ma'lumot" deb o'ylaydi. Bu xatodan ham yomonroq: jim noto'g'ri natija. Transform'dagi tayyor(xato) qatori — keyingi darsning mavzusi; hozircha "shu bo'lakda xato" degan signal deb biling.

6.2 Yechim: vaqtinchalik fayl va rename

Xavfsiz fayl operatsiyalari darsidagi atomik yozishni eslang: avval vaqtinchalik faylga yoziladi, muvaffaqiyatli tugasa — rename. Kanonda 20 marta uzilgan yozishdan oddiy usulda 17 tasi buzilgan, temp + rename usulida — 0. Kanondagi oqimniYoz shu g'oyani pipeline bilan birlashtiradi:

const vaqtinchalik = `${s.fayl}.${String(process.pid)}.tmp`;
try {
  const fayl = createWriteStream(vaqtinchalik);
  await (s.gzip
    ? pipeline(manba, createGzip(), fayl, { signal: s.signal })
    : pipeline(manba, fayl, { signal: s.signal }));
  await rename(vaqtinchalik, s.fayl);
} catch (xato) {
  await rm(vaqtinchalik, { force: true });
  throw xato;
}

Qatorma-qator:

Natija: eur.csv.gz yo to'liq bor, yo umuman yo'q. Uchinchi holat — yarim fayl — bo'lmaydi.

Bitta halol istisno bor — Ctrl+C. «AbortSignal bilan vaqt chegarasi» bo'limida aytganimizdek, bu bosqichda valyuta Ctrl+C da darhol process.exit(130) qiladi. Unda catch dagi rm ishlashga ulgurmaydi va yonda .tmp fayl qolishi mumkin. Asosiy fayl esa buzilmaydi. Batafsil — katta fayl amaliyoti darsida.

7. Brauzerda ham xuddi shunday: pipeTo

Streams API darsida Web Streams bilan ishlagansiz. Ularda pipeline o'rnida pipeTo va pipeThrough bor va ular xatoni uzatadi. Bu blok «Ishga tushir» bilan brauzerda ham ishlaydi:

const manba = new ReadableStream({
  start(boshqaruv) {
    boshqaruv.enqueue("osh");
    setTimeout(() => boshqaruv.error(new Error("tarmoq uzildi")), 10);
  },
});
const qabul = new WritableStream({
  write(b) { console.log("keldi:", b); },
  abort(sabab) { console.log("qabul bekor:", sabab.message); },
});
try {
  await manba.pipeTo(qabul);
} catch (xato) {
  console.log("pipeTo xatosi:", xato.message);
}

Konsolda:

keldi: osh
qabul bekor: tarmoq uzildi
pipeTo xatosi: tarmoq uzildi

Manba buzilganda pipeTo qabul qiluvchini abort qildi va Promise'ni rad etdi. Web Streams standarti boshidanoq xato uzatish bilan loyihalangan. Node oqimlari bu standartdan oldin paydo bo'lgan, pipe() xulqi tarixiy — uni o'zgartirish eski kodni buzardi. Shuning uchun Node pipeline ni alohida qo'shdi.

8. Ko'p uchraydigan xatolar

8.1 .pipe() zanjiriga bitta on("error")

// ❌ xato faqat oxirgi oqimdan ushlanadi
fayl.pipe(ochuvchi).pipe(chiqish).on("error", xabarBer);

.pipe() keyingi oqimni qaytaradi, demak on("error") faqat chiqish ga ulandi. fayl yoki ochuvchi xatosi tinglovchisiz qoladi va jarayonni yiqitadi. Tuzatish — await pipeline(fayl, ochuvchi, chiqish).

8.2 pipeline ni await siz chaqirish

// ❌ funksiya "tugadi" deydi, lekin yozish hali davom etyapti
pipeline(manba, createWriteStream("kurs.csv"));
console.log("tayyor");

tayyor darhol chiqadi, fayl esa hali yozilmoqda. Xato bo'lsa — ushlanmagan rad etilish. await pipeline(...) va try/catch.

8.3 process.stdout ga ikki marta pipeline

pipeline oxirgi oqimni tugatadi (end()). process.stdout ni tugatsangiz, keyingi yozish hech qachon "tugadi" demaydi. Biz ishga tushirdik:

await pipeline(Readable.from(["birinchi\n"]), process.stdout);
console.log("pipeline'dan keyin");
await pipeline(Readable.from(["ikkinchi\n"]), process.stdout);
console.log("oxiri");

birinchi
pipeline'dan keyin
ikkinchi
Warning: Detected unsettled top-level await at file:///C:/Users/…/c1.mjs:6

oxiri chiqmadi, exit kodi 13. Tuzatish — { end: false }: pipeline(manba, process.stdout, { end: false }). Kanon valyuta tarix natijasini ekranga aynan shunday chiqaradi.

8.4 Yarim faylni unutish

catch ichida faqat xabar chiqarib, diskdagi faylni qoldirish — «Yarim yozilgan fayl» bo'limidagi tuzoq. Vaqtinchalik fayl, rename, xatoda rm.

8.5 Xatoni yutib yuborish

// ❌ xato yo'qoladi, dastur "muvaffaqiyatli" tugaydi
try { await pipeline(a, b); } catch {}

CLI exit kodi 0 bilan chiqadi, skript esa "hammasi joyida" deb o'ylaydi. Xatoni tushunarli xabar bilan chiqaring va process.exitCode = 1 qo'ying (process obyekti).

9. Mashqlar

1-mashq (oson): Kim xatoni oladi?

Quyidagi kod nima chiqaradi? Avval o'ylang, keyin o'z kompyuteringizda node bilan tekshiring (bu kod saytda ishlamaydi — Node API).

import { Readable, PassThrough } from "node:stream";

const manba = new Readable({ read() {} });
const orta = new PassThrough();
manba.on("error", () => console.log("manba: xato"));
orta.on("error", () => console.log("orta: xato"));
orta.on("close", () => console.log("orta: yopildi"));
manba.pipe(orta);
manba.destroy(new Error("uzildi"));
setTimeout(() => console.log("orta ochiqmi:", !orta.destroyed), 50);

Konsolda:

manba: xato
orta ochiqmi: true

2-mashq (o'rta): pipe dan pipeline ga

Quyidagi kod faylni siqadi, lekin fayl yo'q bo'lsa jarayon yiqiladi. Uni stream/promises dagi pipeline ga o'tkazing: muvaffaqiyatda siqildi, xatoda xato: <kod> chiqsin. Eslatma: createGzip — zlib darsidan.

import { createReadStream, createWriteStream } from "node:fs";
import { createGzip } from "node:zlib";

createReadStream("kurslar.csv")
  .pipe(createGzip())
  .pipe(createWriteStream("kurslar.csv.gz"));
console.log("siqildi");

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

async function siq(fayl) {
  try {
    await pipeline(
      createReadStream(fayl),
      createGzip(),
      createWriteStream(fayl + ".gz"),
    );
    console.log("siqildi");
  } catch (xato) {
    console.log("xato:", xato.code);
  }
}
await siq("kurslar.csv");

xato: ENOENT

3-mashq (qiyin): Atomik saqla

saqla(bolaklar, fayl, signal) funksiyasini yozing. U bo'laklar massivini (yoki async generatorni) faylga pipeline bilan yozsin: avval <fayl>.tmp ga, muvaffaqiyatda rename. Xato yoki bekor qilishda vaqtinchalik fayl o'chsin va xato yuqoriga uzatilsin. Sinash uchun ikki chaqiruv: oddiy (["a\n", "b\n"]) va 50 ms vaqt chegarali sekin generator. Ikkinchisidan keyin papkada na .tmp, na asosiy fayl qolmasligi kerak. Eslatma: rename, rm — node:fs/promises (fs: uch API shakli).

import { createWriteStream, existsSync, readFileSync } from "node:fs";
import { rename, rm } from "node:fs/promises";
import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { setTimeout as kut } from "node:timers/promises";

async function saqla(bolaklar, fayl, signal) {
  const tmp = `${fayl}.tmp`;
  try {
    await pipeline(Readable.from(bolaklar), createWriteStream(tmp), {
      signal,
    });
    await rename(tmp, fayl);
  } catch (xato) {
    await rm(tmp, { force: true });
    throw xato;
  }
}

await saqla(["a\n", "b\n"], "tez.txt");
console.log(JSON.stringify(readFileSync("tez.txt", "utf8")));

async function* sekin() {
  for (;;) { await kut(20); yield "x\n"; }
}
try {
  await saqla(sekin(), "sekin.txt", AbortSignal.timeout(50));
} catch (xato) {
  console.log(xato.name);
}
console.log(existsSync("sekin.txt.tmp"), existsSync("sekin.txt"));

"a\nb\n"
AbortError
false false

4-mashq: CLI qadami — oqimniYoz va faylniOch

Bu dars holat 35 ning bir qismi; hali commit qilmang — commit katta fayl amaliyoti darsida. valyuta loyihangizda:

1. src/tarix.ts ga oqimniYoz(bolaklar, { gzip, fayl, signal }) ni qo'shing: Readable.from(bolaklar) → (createGzip()) → <fayl>.<pid>.tmp → rename; xatoda rm va throw. Fayl berilmasa — process.stdout ga { end: false } bilan. Kod — «Vaqtinchalik fayl va rename» bo'limidagi parcha.
2. src/tahlil.ts ga faylniOch(fayl) ni qo'shing («destroy() — qo'lda to'xtatish» bo'limidagi parcha) — .gz bo'lsa ochuvchi, manba xatosi ochuvchi.destroy bilan uzatiladi.
3. Testga bitta holat qo'shing: yo'q .gz fayl ENOENT bilan rad etilsin (osilmasin).

const yoq = path.join(await vaqtinchalikPapka(), "yoq.csv.gz");
await assert.rejects(tahlilQil(faylniOch(yoq)), { code: "ENOENT" });

10. Real ishda

• Fayl yuklash va eksport: Excel/CSV hisobot yaratish, rasm siqish, log arxivlash — hammasi pipeline bilan yoziladi. Code review'da .pipe() zanjiri ko'rinsa, "xato qayerda ushlanadi?" degan savol odatiy.
• HTTP server'lar (25-qismda o'rganamiz): javobni oqim bilan yuborishda mijoz ulanishni uzsa, pipeline faylni ham yopadi. .pipe() bilan yozilgan server'da ochiq fayl deskriptorlari to'planadi — bu klassik sizish.
• Kutubxonalar: eski pump va end-of-stream paketlari pipeline va finished ning ajdodlari. Ular Node'ga kiritilgan, yangi kodda paket kerak emas.
• Intervyu savollari: ".pipe() va pipeline farqi?", "Stream xatosi bo'lsa resurslar qanday tozalanadi?", "Oqimni qanday bekor qilasiz?".

Xulosa

• .pipe() ma'lumotni uzatadi, lekin xatoni va yopilishni uzatmaydi — keyingi oqim abadiy kutishi mumkin.
• pipeline (node:stream/promises) birinchi xatoda hamma oqimni yopadi va Promise'ni rad etadi.
• finished(oqim) bitta oqim tugashini yoki buzilishini kutadi.
• { signal } va destroy() — zanjirni tashqaridan to'xtatish yo'llari.
• pipeline diskdagi yarim faylni o'chirmaydi — vaqtinchalik fayl, rename, xatoda rm.
• process.stdout ga pipeline — { end: false } bilan.

Keyingi dars: O'z Readable/Transform stream'ingizni yozish — _transform, _flush, Readable.from va objectMode bilan ma'lumotni yo'lda o'zgartiruvchi o'z oqimingizni yasaysiz.

Manbalar

• Node.js hujjati: Stream — pipeline va finished (v24).
• Node.js hujjati: stream/promises API.
• WHATWG: Streams Standard — pipeTo.