Stream nima — Readable, Writable, Duplex, Transform

~20 daq · Node.js · Stream va Buffer

Muammodan boshlaylik

14-darsda fs.createReadStream ni ko'rdik — fayl o'qish uchun stream. Lekin stream faqat fayl bilan emas. HTTP so'rovi ham stream. HTTP javobi ham stream. Tarmoq ulanishi ham stream. Zip arxivlash ham stream.

Bularning hammasi bir xil interfeys — Stream. Shuning uchun ularni bir-biriga ulash mumkin: fayldan o'qi → zip qil → HTTP ga yubor. Har biri stream bo'lgani uchun pipe bilan zanjir hosil qilish mumkin.

Bu darsda stream ning to'rt turini tushuntiramiz.

Stream nima — chuqurroq tushuntirish

14-darsda stream ni "bo'laklarga bo'lib o'qish" deb tushuntirdik. Bu to'g'ri, lekin to'liq emas.

Stream — aslida EventEmitter. Ya'ni 16-darsda o'rgangan EventEmitter dan meros olgan. Faqat fayl emas — istalgan ma'lumot manbai stream bo'lishi mumkin: tarmoq, klaviatura, boshqa dastur, hatto xotiradagi ma'lumot.

Stream ning to'rt turi bor:

Readable   — faqat o'qish mumkin   (fayl o'qish, HTTP so'rov)
Writable   — faqat yozish mumkin   (fayl yozish, HTTP javob)
Duplex     — o'qish ham, yozish ham (TCP socket)
Transform  — o'qib, o'zgartirib, yozadi (zip, shifrlash)

Readable Stream

Readable stream — ma'lumot manbai. Undan ma'lumot olinadi.

Misollari: fs.createReadStream, http.IncomingMessage (kelgan so'rov), process.stdin (klaviatura).

Readable stream ikki rejimda ishlaydi:

Flowing mode — ma'lumot avtomatik keladi, data eventi orqali:

const { Readable } = require('stream');

const readable = new Readable({
  read() {} // kerak bo'lganda chaqiriladi
});

readable.on('data', (chunk) => {
  console.log('Keldi:', chunk.toString());
});

readable.on('end', () => {
  console.log('Tugadi');
});

// Ma'lumot qo'shish:
readable.push('Birinchi bo\'lak\n');
readable.push('Ikkinchi bo\'lak\n');
readable.push(null); // null — tugadi degani

Paused mode — ma'lumot so'ralganda keladi, read() bilan:

readable.on('readable', () => {
  let chunk;
  while ((chunk = readable.read()) !== null) {
    console.log('O\'qildi:', chunk.toString());
  }
});

Amalda ko'pincha pipe yoki for await...of ishlatiladi — rejimni o'zingiz boshqarmasdan.

for await...of — zamonaviy usul:

const fs = require('fs');

async function processFile() {
  const stream = fs.createReadStream('data.txt', 'utf8');

  for await (const chunk of stream) {
    console.log('Bo\'lak:', chunk);
  }

  console.log('Tugadi');
}

Bu eng o'qilishi oson usul — async/await bilan stream o'qish.

O'z Readable stream ini yaratish

Ba'zan ma'lumot manbai fayl emas — masalan, ma'lumotlar bazasidan qatorlarni stream qilib yuborish:

const { Readable } = require('stream');

class DatabaseStream extends Readable {
  constructor(query) {
    super({ objectMode: true }); // ob'ektlar bilan ishlash
    this.query  = query;
    this.offset = 0;
    this.limit  = 100; // har safar 100 ta qator
  }

  async _read() {
    const rows = await db.query(this.query, this.offset, this.limit);

    if (rows.length === 0) {
      this.push(null); // ma'lumot tugadi
      return;
    }

    rows.forEach(row => this.push(row));
    this.offset += this.limit;
  }
}

// Ishlatish:
const stream = new DatabaseStream('SELECT * FROM users');
for await (const user of stream) {
  console.log(user.name);
}
// 1 million foydalanuvchi bo'lsa ham — xotira sarfi minimal

objectMode: true — standart bo'yicha stream faqat Buffer yoki string bilan ishlaydi. objectMode da ixtiyoriy JavaScript ob'ekti uzatish mumkin.

Writable Stream

Writable stream — ma'lumot maqsadi. Unga ma'lumot yoziladi.

Misollari: fs.createWriteStream, http.ServerResponse (javob), process.stdout (terminal).

const { Writable } = require('stream');

const writable = new Writable({
  write(chunk, encoding, callback) {
    // chunk — kelgan bo'lak
    // callback — "yozib bo'ldim, keyingisini yubor" degani
    console.log('Yozildi:', chunk.toString());
    callback(); // xato bo'lsa: callback(new Error('...'))
  }
});

writable.write('Birinchi\n');
writable.write('Ikkinchi\n');
writable.end('Oxirgi\n', () => {
  console.log('Hammasi yozildi');
});

callback muhim — uni chaqirmasangiz stream to'xtab qoladi. Keyingi bo'lak kelmaydi.

Duplex Stream

Duplex — bir vaqtda ham o'qish, ham yozish mumkin. Lekin ikki tomon mustaqil — o'qilgan narsa yozilgan narsa bilan bog'liq emas.

Eng yaxshi misol — TCP socket:

const net = require('net');

const server = net.createServer((socket) => {
  // socket — Duplex stream
  // O'qish: mijozdan kelgan ma'lumot
  socket.on('data', (data) => {
    console.log('Mijozdan:', data.toString());
  });

  // Yozish: mijozga javob yuborish
  socket.write('Salom, mijoz!\n');
  socket.end();
});

server.listen(3000);

socket dan o'qiladi (mijoz nima yuboryapti) va unga yoziladi (javob) — ikkalasi mustaqil.

Transform Stream

Transform — Duplex ning maxsus turi. Kirgan ma'lumotni o'zgartirib chiqaradi. Kirish va chiqish bog'liq.

Misollari: zlib.createGzip() (zip qilish), crypto.createCipher() (shifrlash), CSV → JSON aylantirish.

const { Transform } = require('stream');

// Matnni katta harfga aylantiruvchi Transform:
const upperCase = new Transform({
  transform(chunk, encoding, callback) {
    const upper = chunk.toString().toUpperCase();
    callback(null, upper); // null — xato yo'q, upper — chiqish
  }
});

// Ishlatish:
process.stdin.pipe(upperCase).pipe(process.stdout);
// Klaviaturadan yozilgan har narsa katta harfda chiqadi

Amaliy misol — CSV ni JSON ga aylantirish:

const { Transform } = require('stream');

class CsvToJson extends Transform {
  constructor() {
    super({ objectMode: true });
    this.headers = null;
    this.buffer  = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop(); // oxirgi to'liqsiz qator keyingiga qoladi

    lines.forEach(line => {
      if (!line.trim()) return;

      if (!this.headers) {
        this.headers = line.split(',');
        return;
      }

      const values = line.split(',');
      const obj    = {};
      this.headers.forEach((h, i) => obj[h.trim()] = values[i]?.trim());
      this.push(obj);
    });

    callback();
  }
}

// Ishlatish:
const fs = require('fs');

fs.createReadStream('users.csv', 'utf8')
  .pipe(new CsvToJson())
  .on('data', (user) => {
    console.log(user); // { name: 'Ali', age: '25', city: 'Toshkent' }
  });

pipe zanjiri

Stream ning kuchi — pipe bilan zanjir hosil qilish. Har bir bo'lak zanjir bo'ylab o'tadi.

Fayl o'qib, zip qilib, diskka yozish:

const fs   = require('fs');
const zlib = require('zlib');

fs.createReadStream('big-file.txt')
  .pipe(zlib.createGzip())          // zip qilish
  .pipe(fs.createWriteStream('big-file.txt.gz'));

// big-file.txt → [bo'lak] → zip → [zip bo'lak] → disk
// Xotira sarfi: faqat bir bo'lak hajmi

Zip qilib, HTTP ga yuborish:

const http = require('http');
const fs   = require('fs');
const zlib = require('zlib');

http.createServer((req, res) => {
  res.setHeader('Content-Encoding', 'gzip');
  fs.createReadStream('big-file.txt')
    .pipe(zlib.createGzip())
    .pipe(res); // res — Writable stream
}).listen(3000);

pipeline — pipe ning xavfsiz versiyasi

pipe ning muammosi — xato bo'lsa stream lar to'liq yopilmaydi, memory leak bo'lishi mumkin. pipeline bu muammoni hal qiladi:

const { pipeline } = require('stream/promises');
const fs   = require('fs');
const zlib = require('zlib');

async function compress() {
  await pipeline(
    fs.createReadStream('input.txt'),
    zlib.createGzip(),
    fs.createWriteStream('output.gz')
  );
  console.log('Tugadi');
}

compress().catch(console.error);
// Xato bo'lsa — barcha stream lar to'g'ri yopiladi

Yangi loyihalarda pipe o'rniga pipeline ishlatiladi.

Tez-tez uchraydigan xatolar

callback ni chaqirmaslik:

const writable = new Writable({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    // callback() unutildi → stream to'xtab qoladi, keyingi bo'lak kelmaydi
  }
});

pipe da xatolarni ushlamaslik:

// Xato — xato ushlanmaydi:
readStream.pipe(writeStream);

// To'g'ri — pipeline ishlatiladi:
await pipeline(readStream, writeStream);

null push qilishni unutish:

class MyReadable extends Readable {
  _read() {
    this.push('ma\'lumot');
    // null push qilinmasa — stream hech qachon tugamaydi
    // this.push(null); ← kerak
  }
}

Xulosa

Stream — ma'lumotni bo'laklarga bo'lib qayta ishlash mexanizmi. To'rt turi:

pipe stream larni ulaydi. pipeline — pipe ning xavfsiz versiyasi, yangi kodda shu ishlatiladi.

Keyingi dars

18-dars: pipe, pipeline — stream larni ulash, Backpressure