pipe, pipeline — stream larni ulash, Backpressure
Muammodan boshlaylik
Oldingi darsda stream larni ko'rdik. Endi ularni bir-biriga ulash haqida.
Tasavvur qiling: siz katta faylni o'qib, zip qilib, diskka yozmoqchisiz. Qo'lda yozilsa:
const readStream = fs.createReadStream('input.txt');
const gzip = zlib.createGzip();
const writeStream = fs.createWriteStream('output.gz');
readStream.on('data', (chunk) => {
const compressed = gzip.write(chunk); // zip qilish
if (!compressed) {
readStream.pause(); // yozish ulgurmasdan o'qish to'xtatiladi
}
});
gzip.on('data', (chunk) => {
writeStream.write(chunk);
});
gzip.on('drain', () => {
readStream.resume();
});
// ... va hokazo
Bu juda murakkab. pipe bularning hammasini avtomatik qiladi.
pipe nima?
pipe — Readable stream ning chiqishini Writable stream ning kirishiga ulaydi. Ma'lumot avtomatik oqadi.
const fs = require('fs');
const zlib = require('zlib');
fs.createReadStream('input.txt')
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream('output.gz'));
Uch qator — xuddi yuqoridagi murakkab kod bilan bir xil natija. pipe quyidagilarni avtomatik qiladi:
- Readable dan bo'lak o'qiydi
- Writable ga yozadi
- Writable band bo'lsa — Readable ni to'xtatadi (backpressure)
- Writable bo'shagach — Readable ni davom ettiradi
- Readable tugagach — Writable ni yopadi
Backpressure nima?
Bu pipe ning eng muhim xususiyati. Tushunish uchun misol:
Tasavvur qiling: disk dan o'qish tezligi — 500 MB/s. Tarmoqqa yozish tezligi — 10 MB/s. Agar o'qish to'xtatilmasa:
Disk: [500 MB/s] → → → → → → → → → → →
Tarmoq: [10 MB/s] ← ← ← ← ← ← ← ← ← ←
↓
Xotira to'lib ketadi!
Backpressure — "sekinroq yoz, men ulgura olmayapman" degan signal. pipe bu signalni avtomatik boshqaradi:
Disk o'qiydi → Tarmoqqa yozadi
Tarmoq band bo'lsa → Disk to'xtatiladi (pause)
Tarmoq bo'shagach → Disk davom etadi (resume)
Xotira sarfi — har doim bir bo'lak hajmi. Fayl qanchalik katta bo'lmasin.
pipe — qo'lda ishlatish
const fs = require('fs');
const readStream = fs.createReadStream('source.txt');
const writeStream = fs.createWriteStream('dest.txt');
readStream.pipe(writeStream);
writeStream.on('finish', () => {
console.log('Nusxalash tugadi');
});
Zanjir hosil qilish — Transform stream lar orqali:
const fs = require('fs');
const zlib = require('zlib');
const crypto = require('crypto');
// O'qi → zip qil → shifirla → yoz
fs.createReadStream('secret.txt')
.pipe(zlib.createGzip())
.pipe(crypto.createCipheriv('aes-256-cbc', key, iv))
.pipe(fs.createWriteStream('secret.txt.gz.enc'));
Har bir .pipe() Transform stream qaytaradi — shuning uchun zanjir hosil qilish mumkin.
pipe ning muammosi
pipe qulay, lekin bir kamchiligi bor — xato bo'lsa stream lar to'liq yopilmaydi.
const readStream = fs.createReadStream('input.txt');
const writeStream = fs.createWriteStream('output.txt');
readStream.pipe(writeStream);
readStream.on('error', (err) => {
console.error('O\'qishda xato:', err);
writeStream.destroy(); // qo'lda yopish kerak
});
writeStream.on('error', (err) => {
console.error('Yozishda xato:', err);
readStream.destroy(); // qo'lda yopish kerak
});
Har bir stream uchun alohida xato boshqarish kerak. Zanjir uzun bo'lsa — juda ko'p kod.
pipeline — to'g'ri yechim
pipeline — pipe ning xavfsiz versiyasi. Xato bo'lsa barcha stream larni avtomatik yopadi.
const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');
async function compress(input, output) {
await pipeline(
fs.createReadStream(input),
zlib.createGzip(),
fs.createWriteStream(output)
);
console.log('Tugadi');
}
compress('input.txt', 'output.gz').catch(console.error);
Xato bo'lsa — catch ga tushadi, barcha stream lar yopiladi. Memory leak yo'q.
Yangi loyihalarda har doim pipeline ishlatiladi.
HTTP da stream
HTTP javobini stream bilan yuborish — katta fayllar uchun ideal:
const http = require('http');
const fs = require('fs');
const zlib = require('zlib');
const { pipeline } = require('stream/promises');
http.createServer(async (req, res) => {
if (req.url === '/download') {
res.setHeader('Content-Type', 'application/octet-stream');
res.setHeader('Content-Encoding', 'gzip');
try {
await pipeline(
fs.createReadStream('big-file.txt'),
zlib.createGzip(),
res // res — Writable stream
);
} catch (err) {
if (!res.headersSent) {
res.statusCode = 500;
res.end('Xato');
}
}
}
}).listen(3000);
Fayl to'liq xotiraga yuklanmaydi — bo'laklarga bo'linib, zip qilinib, mijozga yuboriladi.
HTTP so'rovini stream bilan qabul qilish
Mijoz katta fayl yuborayotganda — req ham Readable stream:
const http = require('http');
const fs = require('fs');
const { pipeline } = require('stream/promises');
http.createServer(async (req, res) => {
if (req.method === 'POST' && req.url === '/upload') {
try {
await pipeline(
req, // Readable stream — kelayotgan fayl
fs.createWriteStream('uploaded-file')
);
res.end('Yuklandi');
} catch (err) {
res.statusCode = 500;
res.end('Xato');
}
}
}).listen(3000);
Fayl kelayotganda darhol diskka yoziladi — xotiraga yuklanmaydi.
Stream ni Promise ga aylantirish
Ba'zan stream ni await bilan kutish kerak bo'ladi:
const { Writable } = require('stream');
function streamToBuffer(readable) {
return new Promise((resolve, reject) => {
const chunks = [];
readable.on('data', (chunk) => chunks.push(chunk));
readable.on('end', () => resolve(Buffer.concat(chunks)));
readable.on('error', (err) => reject(err));
});
}
// Ishlatish:
const stream = fs.createReadStream('small-file.txt');
const buffer = await streamToBuffer(stream);
console.log(buffer.toString());
Kichik fayllar uchun yaxshi. Katta fayllar uchun — to'liq xotiraga yuklanadi, ehtiyot bo'ling.
Tez-tez uchraydigan xatolar
pipe da xatolarni ushlamaslik:
// Xato — xato bo'lsa stream lar yopilmaydi:
readStream.pipe(writeStream);
// To'g'ri:
await pipeline(readStream, writeStream);
pipeline ni async funksiyasiz ishlatish:
// Xato:
pipeline(readStream, writeStream); // Promise qaytaradi, kutilmaydi
// To'g'ri:
await pipeline(readStream, writeStream);
// yoki
pipeline(readStream, writeStream).catch(console.error);
Yopilgan stream ga yozish:
writeStream.end();
writeStream.write('yana'); // Error: write after end
end() chaqirilgandan keyin stream yopiladi — yozib bo'lmaydi.
Xulosa
pipe — stream larni ulaydi, backpressure ni avtomatik boshqaradi. Qulay, lekin xato boshqarish qo'lda.
pipeline — pipe ning xavfsiz versiyasi. Xato bo'lsa barcha stream lar yopiladi. Yangi kodda har doim pipeline ishlatiladi.
Backpressure — tez o'quvchi va sekin yozuvchi o'rtasidagi muvozanat. pipe va pipeline buni avtomatik hal qiladi — siz o'ylamasangiz ham bo'ladi.
Keyingi dars
19-dars: Buffer — alloc, from, concat, encoding