Node.js EventEmitter va Streams — pipe, backpressure (5.4-bob)

5-QISM — Node.js Backend · 4-mavzu

1. Kirish va motivatsiya

Core modullarni 5.3-bob bildik. Endi Node'ning ikki fundamental, event-asoslangan tushunchasini chuqur o'rganamiz — bularsiz Node'ning ichki ishlashini to'liq tushunib bo'lmaydi:

1. EventEmitter — "hodisa chiqarish va tinglash" mexanizmi. Bu shunchaki bir modul emas — Node'ning butun arxitekturasi event-driven (hodisaga asoslangan). HTTP server (req, res), streams, process 5.1-bob — hammasi EventEmitter. 2.16-bobdagi brauzer eventlari (DOM) ning server tomondagi ukasi.

2. Streams — katta ma'lumotni xotirani to'ldirmasdan, bo'lak-bo'lak (chunk) qayta ishlash (0.1: RAM cheklangan). 5.1-bobda "5 GB faylni readFileSync bilan o'qimang — RAM to'lib ketadi" degandik. Streams aynan shu muammoni hal qiladi: ma'lumot oqim kabi oqib o'tadi, hech qachon to'liq xotirada bo'lmaydi.

O'xshatish (EventEmitter): radio stansiya. Stansiya (emitter) signal chiqaradi (emit); tinglovchilar (listeners) o'sha to'lqinga ulanib (on), eshitadi. Stansiya kim eshitayotganini bilmaydi — shunchaki chiqaradi. Bu — decoupling (qismlarni ajratish): chiqaruvchi va tinglovchi bir-biriga bog'liq emas.

O'xshatish (Streams): hovuzni chelaklab ko'chirish vs shlang orqali. readFileSync — butun hovuzni bitta ulkan idishga quyib, keyin ko'chirish (idish yetmasligi mumkin — RAM). Stream — shlang: suv oqib o'tadi, hech qachon hammasi bir joyda turmaydi. 100 GB faylni 50 MB RAM bilan ham qayta ishlaysiz.

Nega muhim?

• Node arxitekturasi event-driven — EventEmitter'ni bilmasdan, server (req/res), socket 5.13-bob tushunarsiz.
• Katta ma'lumot (fayl, video, DB natija, HTTP body) — streams'siz xotira muammosi 0.1-bob.
• Performance — streams xotira-samarali, tez (ma'lumot kelishi bilan qayta ishlanadi).
• HTTP 5.5-bob, file upload 5.11-bob, socket.io 5.13-bob — streams/events ustida.

2. Nazariya — chuqur tushuntirish (EventEmitter)

2.1. Event-driven arxitektura nima?

Event-driven (hodisaga asoslangan) — dastur "hodisalar"ga javob berib ishlaydi (so'rov keldi, fayl o'qildi, ulanish yopildi), ketma-ket buyruq bajarish o'rniga. Node'ning butun modeli shunday (5.1: event loop ham shu — eventlarni boshqaradi).

  ANANAVIY (imperativ):  qadam1 → qadam2 → qadam3 (ketma-ket)
  EVENT-DRIVEN:          "so'rov" eventi → handler; "xato" eventi → handler
                         (qachon nima sodir bo'lishi oldindan noma'lum)

2.2. EventEmitter — asosiy mexanizm

EventEmitter — node:events core modulidagi class 5.3-bob. U "hodisa chiqarish" (emit) va "tinglash" (on) imkonini beradi (nodejs.org):

import { EventEmitter } from "node:events";

const emitter = new EventEmitter();

// Tinglash (listener qo'shish) — 2.16: addEventListener kabi
emitter.on("salom", (ism) => {
  console.log(`Salom, ${ism}!`);
});

// Chiqarish (event yuborish) — barcha "salom" listenerlari ishlaydi
emitter.emit("salom", "Ali");   // "Salom, Ali!"
emitter.emit("salom", "Vali");  // "Salom, Vali!"

2.3. EventEmitter metodlari

emitter.on("event", listener);        // tinglash (har emit'da)
emitter.once("event", listener);      // FAQAT BIR MARTA (keyin avtomatik o'chadi)
emitter.emit("event", arg1, arg2);    // chiqarish (sinxron! — 2.5)
emitter.off("event", listener);       // listener o'chirish (removeListener)
emitter.removeAllListeners("event");  // hammasi
emitter.listenerCount("event");       // nechta listener
emitter.eventNames();                  // barcha event nomlari
emitter.prependListener("event", fn); // birinchi qilib qo'shish

on vs once: on — har emitda ishlaydi (doimiy). once — faqat birinchi marta (masalan "ulanish ochildi", "server tayyor" — bir martalik hodisalar).

2.4. Eventlar — SINXRON (muhim nuans)

emit — sinxron: listenerlar darrov, navbat bilan (ro'yxatga qo'shilgan tartibda) ishlaydi, emit qatori tugaguncha (2.5: 5.1 event loop bilan adashtirmang):

emitter.on("ish", () => console.log("2: listener"));
console.log("1: emitdan oldin");
emitter.emit("ish");                 // listener DARROV ishlaydi (sinxron)
console.log("3: emitdan keyin");
// Tartib: 1 → 2 → 3 (emit listener'ni darrov, sinxron chaqiradi)

Agar listenerda async ish kerak bo'lsa — listener ichida async/Promise ishlating (lekin emit ularni kutmaydi).

2.5. error eventi — maxsus va xavfli

error eventi — EventEmitter'da maxsus: agar error chiqarilsa va listener yo'q bo'lsa, Node butun process'ni qulatadi (uncaught exception — 5.1, 2.12):

const emitter = new EventEmitter();

// ❌ error listener yo'q → emit("error") process'ni qulatadi
// emitter.emit("error", new Error("xato"));   // CRASH!

// ✅ doim error listener qo'shing
emitter.on("error", (err) => {
  console.error("Xato ushlandi:", err.message);   // (2.12)
});
emitter.emit("error", new Error("biror xato"));   // endi xavfsiz

Har EventEmitter'da error listener bo'lsin (stream, socket — 5.13). Aks holda bitta xato butun serverni o'ldiradi 5.1-bob.

2.6. Memory leak ogohlantirishi

Har EventEmitter'da default 10 ta listener chegarasi (ogohlantirish uchun). Agar ko'proq qo'shsangiz:

  MaxListenersExceededWarning: Possible EventEmitter memory leak detected.
  11 "data" listeners added.

Sababi: odatda listener'larni off qilmasdan qayta-qayta on qilish (memory leak — 0.1). Yechimi: kerakmas listener'larni off qiling; haqiqatan ko'p kerak bo'lsa emitter.setMaxListeners(n).

2.7. Custom EventEmitter (o'z class — 2.10)

O'z class'ingizni EventEmitter'dan meros qilib (2.10: inheritance), event chiqaruvchi qilasiz:

import { EventEmitter } from "node:events";

class Buyurtma extends EventEmitter {          // (2.10: extends)
  yarat(mahsulot) {
    // ... biznes logika ...
    this.emit("yaratildi", mahsulot);          // event chiqarish
  }
  tolov() {
    this.emit("tolandi");
  }
}

const buyurtma = new Buyurtma();
buyurtma.on("yaratildi", (m) => console.log(`Buyurtma: ${m}`));   // email yuborish (5.19)
buyurtma.on("tolandi", () => console.log("To'lov qabul qilindi")); // SMS (5.18)
buyurtma.yarat("Telefon");   // "Buyurtma: Telefon"

Nega kuchli (decoupling): yarat metodi email/SMS yuborishni bilmaydi — u faqat event chiqaradi. Email/SMS logikasi alohida listener'da. Yangi reaksiya qo'shsangiz (masalan log), yaratni o'zgartirmaysiz (9.1: Open/Closed). Bu — observer pattern 9.2-bob.

2.8. Async Listeners va captureRejections (Promise xatolarini xavfsiz boshqarish)

EventEmitter dastlab callback'lar davrida yaratilgan. Agar siz async listener yozsangiz va uning ichida xatolik yuz bersa (reject bo'lsa), EventEmitter buni bilmaydi va ushlay olmaydi! Bu unhandledRejection hodisasiga olib keladi:

const ee = new EventEmitter();
ee.on("tozala", async () => {
  throw new Error("DB ulanmadi!"); // UnhandledPromiseRejection! Process qulashi mumkin!
});
ee.emit("tozala");

Senior Yechim — captureRejections: true: Node.js ga hodisa uzatgichni yaratishda captureRejections optsiyasini berish orqali barcha async funksiyalardagi reject'larni avtomatik error hodisasiga aylantirish mumkin:

import { EventEmitter } from "node:events";

const ee = new EventEmitter({ captureRejections: true });

ee.on("tozala", async () => {
  throw new Error("DB xatoligi!"); // Avtomatik ushlanadi
});

// Reject bo'lgan barcha Promise'lar to'g'ri shu yerga yo'naltiriladi:
ee.on("error", (err) => {
  console.error("Async listener xatosi xavfsiz ushlandi:", err.message);
});

ee.emit("tozala");

3. Nazariya — chuqur tushuntirish (Streams)

3.1. Stream nima va nega kerak (0.1)

Stream (oqim) — ma'lumotni bo'lak-bo'lak (chunk), vaqt o'tishi bilan qayta ishlash (hammasini birdan emas):

  readFileSync (butun fayl xotiraga):
  [████████████ 5 GB ████████████]  → RAM'da 5 GB! (portlaydi — 0.1)

  Stream (chunk-chunk):
  [64KB] → qayta ishla → tashla → [64KB] → qayta ishla → ...
  → RAM'da bir vaqtda faqat 64KB (5 GB faylni 64KB bilan!)

Foydalari: xotira-samarali (katta fayl/cheksiz ma'lumot), tez (birinchi chunk kelishi bilan boshlaydi, hammasini kutmaydi), kompozitsion (pipe bilan ulanadi).

3.2. Stream — EventEmitter (bog'lanish)

Barcha streamlar — EventEmitter (3-bo'lim, nodejs.org). Ular holatni eventlar bilan bildiradi:

  Readable: "data" (chunk keldi), "end" (tugadi), "error"
  Writable: "drain" (yana yozsa bo'ladi), "finish" (tugadi), "error"

Shuning uchun 2-bo'lim (EventEmitter) — streamlar uchun poydevor.

3.3. To'rt stream turi

Node'da to'rt fundamental stream turi (nodejs.org):

3.4. Readable stream — o'qish

import { createReadStream } from "node:fs";

const stream = createReadStream("katta.txt", { encoding: "utf8", highWaterMark: 64 * 1024 });
// highWaterMark — chunk hajmi (default 64KB)

stream.on("data", (chunk) => {       // har chunk kelganda (2.2: on)
  console.log(`Chunk: ${chunk.length} belgi`);
});
stream.on("end", () => {             // hammasi o'qildi
  console.log("Fayl to'liq o'qildi");
});
stream.on("error", (err) => {        // xato (2.5: MAJBURIY)
  console.error("Xato:", err.message);
});

3.5. Writable stream — yozish

import { createWriteStream } from "node:fs";

const stream = createWriteStream("natija.txt");

stream.write("Birinchi qator\n");    // yozish (chunk)
stream.write("Ikkinchi qator\n");
stream.end("Oxirgi qator\n");         // tugatish (oxirgi yozuv + yopish)

stream.on("finish", () => console.log("Yozish tugadi"));   // (3.2)
stream.on("error", (err) => console.error(err));

3.6. Backpressure — eng muhim tushuncha

Backpressure (orqa bosim) — Readable tez ma'lumot bersa, lekin Writable sekin yozsa, ma'lumot to'planib qoladi (xotira to'ladi — 0.1). Node buni boshqaradi (nodejs.org):

  Readable (tez) ──→ Writable (sekin)
                      buffer to'ldi! (highWaterMark oshdi)
  write() → false qaytaradi → Readable'ni TO'XTAT (pause)
  Writable bo'shadi → "drain" eventi → Readable'ni DAVOM ETTIR (resume)

// write() false qaytarsa — to'xtatish kerak (qo'lda murakkab)
const ok = writable.write(chunk);
if (!ok) {
  readable.pause();                    // to'xtatish (buffer to'la)
  writable.once("drain", () => readable.resume());   // bo'shaganda davom
}
// ✅ Lekin buni QO'LDA qilmang — pipe/pipeline avtomatik qiladi (3.7)

Backpressure'ni qo'lda boshqarish murakkab va xatoga moyil. Shuning uchun pipe/pipeline ishlatamiz — ular backpressure'ni avtomatik boshqaradi 3.7-bob. Bu — streams'ning eng katta amaliy qoidasi.

3.7. pipe va pipeline — ulash (eng amaliy)

pipe — Readable'ni Writable'ga ulaydi; backpressure avtomatik 3.6-bob:

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

// Faylni nusxalash (xotira-samarali — har qanday hajm)
createReadStream("manba.txt").pipe(createWriteStream("nusxa.txt"));
// Readable → Writable, backpressure avtomatik (3.6)

Lekin pipe xatolarni to'g'ri boshqarmaydi va resurslarni tozalamaydi. Production uchun pipeline (xato + tozalash + callback/Promise — nodejs.org):

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

// ✅ Production — pipeline (xato boshqaruvi + tozalash — 2.12)
await pipeline(
  createReadStream("manba.txt"),
  createWriteStream("nusxa.txt")
);
// Xato bo'lsa — throw qiladi (try/catch bilan ushlanadi), resurslar yopiladi

Production'da pipe emas, pipeline ishlating (nodejs.org). pipeda xato bo'lsa — stream "osilib" qoladi, memory leak. pipeline — xatoni uzatadi va tozalaydi.

3.8. Transform stream — ma'lumotni o'zgartirish

Transform — o'qib, o'zgartirib, yozadi (gzip, shifrlash, formatlash):

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

// Faylni siqish (gzip — Transform stream)
await pipeline(
  createReadStream("katta.log"),
  createGzip(),                       // Transform: siqadi (3.3)
  createWriteStream("katta.log.gz")
);
// O'qiladi → siqiladi → yoziladi, hammasi oqim bilan (xotira-samarali)

3.9. O'z Transform stream'ingiz

import { Transform } from "node:stream";

// Har chunkni katta harfga aylantiruvchi Transform
const kattaHarf = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());   // o'zgartirib uzat (2.6)
    callback();                                   // tugadi (xato bo'lsa callback(err))
  }
});
// Ishlatish: readStream.pipe(kattaHarf).pipe(writeStream)

3.10. Stream rejimlari va Buffer (0.1)

• Buffer — binary ma'lumot (0.1: bayt); stream chunk'lari default Buffer (encoding berilmasa).
• flowing vs paused — Readable'ning ikki rejimi: pipe/on("data") — flowing (avtomatik oqadi); read() — paused (qo'lda).
• Object mode — chunk Buffer emas, obyekt bo'lishi mumkin (objectMode: true) — DB natijalari uchun.

3.11. Async Iterables (for await...of), AbortSignal va highWaterMark nozikliklari

Callback va event listenerlar (stream.on('data')) o'rniga zamonaviy JavaScript streamlarni asinxron iterator sifatida aylanish imkonini beradi:

import { createReadStream } from "node:fs";

async function processLog(filePath) {
  const stream = createReadStream(filePath, { encoding: "utf8" });

  // ✅ Toza, sinxrondek o'qiladigan va xavfsiz oqim
  for await (const chunk of stream) {
    if (chunk.includes("FATAL_ERROR")) {
      console.log("Xavfli xato topildi, to'xtatamiz!");
      break; // break qilinganda stream avtomatik ravishda destroy() bo'ladi va resurs tozalanadi!
    }
  }
}

AbortController bilan Streamlarni bekor qilish (Lifecycle nazorati)

Agar HTTP mijoz so'rovni bekor qilsa yoki jarayon vaqti tugasa (timeout), streamni zudlik bilan to'xtatish kerak:

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

const ac = new AbortController();

// 5 sekunddan keyin bekor qilish
setTimeout(() => ac.abort(new Error("Timeout: Jarayon juda ko'p vaqt oldi")), 5000);

try {
  await pipeline(
    createReadStream("huge_backup.tar"),
    createWriteStream("destination.tar"),
    { signal: ac.signal } // AbortSignal uzatildi!
  );
} catch (err) {
  if (err.name === "AbortError") {
    console.warn("Stream muvaffaqiyatli bekor qilindi va fayl deskriptorlari yopildi.");
  }
}

highWaterMark tuzog'i (Byte vs Object Mode)

highWaterMark — bu stream buferining sig'imi chegarasi.

• Binary/Byte streamlarda: default qiymat 64 * 1024 (fs uchun 64KB) yoki 16 * 1024 (HTTP/TCP uchun 16KB).
• objectMode: true streamlarda: default qiymat atigi 16 (ya'ni 16 ta JS obyekti)! Agar har bir obyektingiz 1MB hajmdagi katta ma'lumot bo'lsa va siz buni bilmasangiz, xotirada tezda yuzlab MB ushlanib qolishi mumkin. Shuning uchun ob'ekt rejimida highWaterMarkni obyektlar og'irligiga qarab ehtiyotkorlik bilan sozlash kerak.

4. Sintaksis — tez ma'lumotnoma

// EventEmitter
import { EventEmitter } from "node:events";
e.on(ev, fn)   e.once(ev, fn)   e.emit(ev, ...args)   e.off(ev, fn)
e.on("error", fn)   // MAJBURIY (2.5)
class X extends EventEmitter {}   // custom (2.7)

// Streams
import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";
rs.on("data"/"end"/"error", fn)   ws.write(x)   ws.end()
await pipeline(rs, transform, ws)   // production 3.7-bob — pipe emas!

5. Batafsil kod namunalari

Misol 1 — Custom EventEmitter (buyurtma tizimi — 2.7)

import { EventEmitter } from "node:events";

class BuyurtmaTizimi extends EventEmitter {
  yarat(buyurtma) {
    console.log(`Buyurtma #${buyurtma.id} yaratildi`);
    this.emit("yaratildi", buyurtma);          // event (2.2)
  }
  tolandi(buyurtma) {
    this.emit("tolandi", buyurtma);
  }
}

const tizim = new BuyurtmaTizimi();

// Turli reaksiyalar — decoupled (2.7)
tizim.on("yaratildi", (b) => console.log(`📧 Email: buyurtma #${b.id} 5.19-bob`));
tizim.on("yaratildi", (b) => console.log(`📊 Statistika yangilandi`));
tizim.once("tolandi", (b) => console.log(`✅ Birinchi to'lov!`));   // (2.3: once)
tizim.on("error", (err) => console.error("Xato:", err.message));   // (2.5)

tizim.yarat({ id: 1, mahsulot: "Telefon" });
tizim.tolandi({ id: 1 });
// yarat → ikkala "yaratildi" listener ishlaydi (sinxron — 2.4)

Misol 2 — Readable stream bilan katta fayl (3.4)

import { createReadStream } from "node:fs";

// 10 GB faylni ham xotirani to'ldirmasdan o'qish (0.1, 3.1)
let qatorlar = 0;
const stream = createReadStream("katta.log", { encoding: "utf8" });

stream.on("data", (chunk) => {                 // har 64KB (3.4)
  qatorlar += chunk.split("\n").length - 1;    // qatorlarni sanang (2.6)
});
stream.on("end", () => console.log(`Jami qator: ${qatorlar}`));
stream.on("error", (err) => console.error(err));   // (2.5)
// readFileSync 10 GB'ni RAM'ga sig'dirolmaydi; stream — bemalol

Misol 3 — pipeline bilan fayl nusxalash va siqish (3.7, 3.8)

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

// Xavfsiz nusxalash (backpressure + xato avtomatik — 3.6, 3.7)
async function nusxala(manba, manzil) {
  try {
    await pipeline(createReadStream(manba), createWriteStream(manzil));
    console.log("Nusxalandi");
  } catch (err) {
    console.error("Nusxalash xatosi:", err.message);   // (2.12)
  }
}

// Log faylni siqish (Transform — 3.8)
async function siq(fayl) {
  await pipeline(
    createReadStream(fayl),
    createGzip(),                      // siqadi (3.8)
    createWriteStream(`${fayl}.gz`)
  );
}

Misol 4 — O'z Transform stream (CSV → JSON — 3.9)

import { Transform } from "node:stream";

// Har qatorni (CSV) JSON'ga aylantiruvchi Transform
const csvToJson = new Transform({
  readableObjectMode: true,            // chiqish — obyekt (3.10)
  transform(chunk, enc, cb) {
    const qatorlar = chunk.toString().trim().split("\n");
    for (const qator of qatorlar) {
      const [ism, yosh] = qator.split(",");   // (2.6: split)
      this.push({ ism, yosh: Number(yosh) }); // obyekt uzat (2.8)
    }
    cb();                              // tugadi (3.9)
  }
});
// createReadStream("users.csv").pipe(csvToJson).on("data", u => console.log(u))

Misol 5 — HTTP server'da stream (5.5 ga ko'prik — 3.2)

import http from "node:http";
import { createReadStream } from "node:fs";

// Katta faylni stream bilan yuborish (xotira-samarali — 3.1)
http.createServer((req, res) => {       // req — Readable, res — Writable! (3.3)
  if (req.url === "/video") {
    // ❌ readFile bilan: butun video RAM'ga, keyin yuborish
    // ✅ stream bilan: oqim, RAM-samarali
    createReadStream("katta-video.mp4").pipe(res);   // (3.7)
  }
}).listen(3000);
// req va res — streamlar (EventEmitter — 3.2); pipe avtomatik backpressure

Misol 6 — error event xavfi (2.5)

import { createReadStream } from "node:fs";

const stream = createReadStream("yoq-fayl.txt");

// ❌ error listener bo'lmasa → process QULAYDI (2.5)
// ✅ doim error tinglang
stream.on("error", (err) => {
  if (err.code === "ENOENT") console.log("Fayl topilmadi");   // (5.3: ENOENT)
  else console.error(err);
});
stream.on("data", (c) => console.log(c.length));

Misol 7 — once va listener boshqaruvi (2.3, 2.6)

import { EventEmitter } from "node:events";
const server = new EventEmitter();

// once — faqat birinchi marta (2.3)
server.once("ready", () => console.log("Server tayyor (bir marta)"));

// Listener'ni keyin o'chirish (memory leak'dan saqlanish — 2.6)
function logHandler(data) { console.log("Log:", data); }
server.on("log", logHandler);
// ... keyin kerak bo'lmasa:
server.off("log", logHandler);          // o'chirish (2.6)

server.emit("ready");   // ishlaydi
server.emit("ready");   // ENDI ishlamaydi (once — 2.3)

6. To'g'ri va noto'g'ri holatlar

1) Katta faylni readFile bilan

// ❌ butun fayl RAM'ga — katta faylda xotira portlaydi (0.1, 3.1)
const data = await readFile("5gb.mp4");

// ✅ stream
createReadStream("5gb.mp4").pipe(res);   // (3.7)

2) Production'da pipe (pipeline o'rniga)

// ❌ pipe — xato boshqarilmaydi, resurs oqishi (3.7)
rs.pipe(ws);

// ✅ pipeline — xato + tozalash
await pipeline(rs, ws);

3) error listener'siz stream/emitter

// ❌ error → process qulaydi (2.5)
stream.on("data", ...);   // error yo'q

// ✅ doim error
stream.on("error", err => ...);

4) Listener'larni off qilmaslik (memory leak)

// ❌ qayta-qayta on, off yo'q → MaxListenersExceededWarning (2.6)
setInterval(() => emitter.on("tick", fn), 1000);

// ✅ once yoki off
emitter.once("tick", fn);

7. Keng tarqalgan xatolar va yechimlari

Xato 1 — MaxListenersExceededWarning

Sababi: 10+ listener (off qilinmagan — memory leak — 2.6). Yechimi: kerakmas listenerlarni off; haqiqatan ko'p bo'lsa setMaxListeners.

Xato 2 — Process qulaydi (uncaught error)

Sababi: stream/emitter error eventi listener'siz 2.5-bob. Yechimi: har emitter/stream'da on("error", ...).

Xato 3 — Xotira to'lib ketadi (katta fayl)

Sababi: readFile/butun ma'lumotni xotiraga 3.1-bob. Yechimi: stream + pipeline 3.7-bob.

Xato 4 — Stream "osilib" qoladi (tugamaydi)

Sababi: pipeda xato yoki backpressure noto'g'ri (3.6, 3.7). Yechimi: pipeline (avtomatik tozalash).

Xato 5 — emit natijasini async kutish

Sababi: emit sinxron — async listenerni kutmaydi 2.4-bob. Yechimi: async natija kerak bo'lsa Promise/callback alohida; events.once(emitter, "event") (Promise).

8. Integratsiya — bu mavzu stack'ning qayerida uchraydi

• Node arxitekturasi 5.1-bob: event-driven; event loop eventlarni boshqaradi.
• Core modullar 5.3-bob: fs stream'lari; crypto stream'lari.
• HTTP 5.5-bob: req (Readable), res (Writable) — streamlar va EventEmitter (Misol 5).
• File upload 5.11-bob: Multer — stream ustida.
• Socket.io 5.13-bob: event-driven (real-time).
• DOM events 2.16-bob: brauzer tomondagi o'xshashi.
• Observer pattern 9.2-bob: EventEmitter — observer'ning amali.
• Buffer 0.1-bob: stream chunk'lari.

9. Eng yaxshi amaliyotlar (best practices)

• Katta/cheksiz ma'lumot — stream (readFile emas — 3.1, xotira).
• Production'da pipeline (pipe emas — xato + tozalash — 3.7).
• Har emitter/stream'da error listener (process qulashidan saqlaydi — 2.5).
• Listener'larni off/once bilan boshqaring (memory leak — 2.6).
• emit sinxron ekanini eslang 2.4-bob; async kerak bo'lsa alohida.
• Custom EventEmitter bilan decoupling (observer — 2.7, 9.2).
• Transform bilan oqim ichida o'zgartiring (gzip, format — 3.8).
• Backpressure'ni qo'lda boshqarmang — pipe/pipeline qiladi 3.6-bob.

10. Amaliy loyiha: "Log Qayta Ishlash Quvuri (Stream Pipeline)"

EventEmitter va Streams'ni amalda mustahkamlash.

Maqsad

EventEmitter (decoupling, error), va Streams (Readable/Writable/Transform, pipeline, backpressure)ni birlashtirib, katta ma'lumotni xotira-samarali qayta ishlash.

Talablar (requirements)

1. Custom EventEmitter: BuyurtmaTizimi — yarat/tolandi/bekor eventlari; bir nechta decoupled listener (email/SMS/log/statistika); error listener (Misol 1, 2.7).
2. once: "birinchi buyurtma" kabi bir martalik hodisa (Misol 7, 2.3).
3. Katta fayl o'qish: katta log faylni stream bilan o'qib, qator/xato/maxsus so'z sanash (Misol 2, 3.4); xotira o'lchovini ko'ring (process.memoryUsage()).
4. Transform: o'z Transform stream — har qatorni parse qilib (masalan log darajasi: INFO/ERROR), filtrlash yoki formatlash (Misol 4, 3.9).
5. pipeline: o'qish → transform (filtr) → yozish (faqat ERROR qatorlarni alohida faylga — 3.7, 3.8).
6. Gzip: natija faylni siqish (Transform — Misol 3, 3.8).
7. Backpressure namoyishi: tez Readable + sekin Writable; pipeline avtomatik boshqarishini ko'rsating 3.6-bob.
8. Xato boshqaruvi: mavjud bo'lmagan fayl, transform xatosi — error/try-catch (2.5, 2.12).
9. Hammasi xotira-samarali (readFile emas — 3.1); xotira o'lchovini taqqoslab ko'rsating.

Maslahatlar (hint)

• EventEmitter: class X extends EventEmitter 2.7-bob; har emit'da bir nechta on.
• Stream: createReadStream/createWriteStream (3.4, 3.5); pipeline from stream/promises 3.7-bob.
• Transform: new Transform({ transform(chunk, enc, cb) {...} }) 3.9-bob; cb() chaqiring.
• Xotira: process.memoryUsage().heapUsed 5.1-bob — stream vs readFile farqi.
• error listener — har stream/emitter 2.5-bob.
• Katta test fayl: fs.appendFile bilan ko'p qator yarating.

"Tayyor" mezonlari (acceptance criteria)

• Custom EventEmitter bir nechta decoupled listener bilan ishlaydi.
• once bir martalik hodisada ishlaydi.
• Katta fayl stream bilan o'qiladi (xotira past).
• O'z Transform stream parse/filtr qiladi.
• pipeline bilan o'qish→transform→yozish ishlaydi.
• Natija gzip bilan siqiladi.
• Har stream/emitter'da error boshqarilgan.
• Xotira-samaradorlik (readFile bilan farq) ko'rsatilgan.

Yechim kodi ataylab berilmagan — bu loyihani o'zingiz yozib ko'ring.

11. Xulosa va keyingi bobga ko'prik

Bu bobda Node'ning ikki event-asoslangan ustunini chuqur o'rgandik:

• EventEmitter — hodisa chiqarish (emit) / tinglash (on/once/off); Node arxitekturasining asosi (event-driven). emit sinxron; error event listener'siz → process qulaydi; memory leak (10 listener); custom emitter — decoupling (observer — 9.2).
• Streams — ma'lumotni bo'lak-bo'lak (chunk) qayta ishlash; xotira-samarali (katta fayl — 0.1). Hammasi EventEmitter. To'rt tur: Readable, Writable, Duplex, Transform.
• Backpressure — tez manba/sekin manzil muvozanati; qo'lda boshqarmang — pipe/pipeline avtomatik boshqaradi.
• Production'da pipeline (pipe emas — xato + tozalash).

Keyingi bob — 5.5-bob: HTTP modul — server yaratish, request/response (qo'lda). Node'ning ichki mexanizmlarini (event, stream) bildik; endi 0.4-bobdagi HTTP nazariyasini amalga oshiramiz: faqat core http modul bilan (Express'siz!) server yaratamiz, so'rov va javobni qo'lda boshqaramiz. Bu — Express 5.6-bob "sehrini" tushunish uchun zarur poydevor.

Foydalanilgan rasmiy/ishonchli manbalar

• nodejs.org/api/events — EventEmitter (on/emit/once, error event, maxListeners)
• nodejs.org/api/stream — Readable/Writable/Duplex/Transform, pipe, pipeline
• nodejs.org/learn — Backpressuring in Streams, How to use Streams