NODE.JS STREAM KULLANMAK PERFORMANS

Node.js stream, büyük dosya ve veriyi parça parça (chunk) işleyerek bellek tüketimini binlerce kat azaltan API'dir; bu yazıda Readable, Writable ve Transform stream'lerini, pipe ve modern pipeline mekanizmasını pratik örneklerle öğreneceksiniz.

1. Stream nedir?

Veri akışı; tek seferde okumak yerine parça parça (chunk) okuma/yazma. Node.js'te 4 ana stream türü var; Readable, Writable ve Transform sınıflarının tüm imzaları için resmi Node.js dokümantasyonu tek kaynaktır:

• Readable: Veri okuma (fs.createReadStream, http response)
• Writable: Veri yazma (fs.createWriteStream, http response)
• Duplex: Hem okur hem yazar (TCP socket)
• Transform: Veri dönüştürür (gzip, encryption)

2. Klasik vs stream yaklaşımı

Klasik (tüm dosya belleğe)

const fs = require('fs');

// 10 GB dosyayı belleğe yükler
const data = fs.readFileSync('big-file.txt');
const processed = data.toString().toUpperCase();
fs.writeFileSync('output.txt', processed);

// Sonuç: 20+ GB RAM kullanır; muhtemelen çöker

Stream ile (parça parça)

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

const upperCase = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

fs.createReadStream('big-file.txt')
  .pipe(upperCase)
  .pipe(fs.createWriteStream('output.txt'));

// Sonuç: birkaç MB RAM; sorunsuz çalışır

3. Readable stream

Veri kaynağından okuyan stream.

const fs = require('fs');

const readable = fs.createReadStream('data.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024 // 64 KB chunk size
});

readable.on('data', (chunk) => {
  console.log('Chunk:', chunk.length, 'bytes');
});

readable.on('end', () => {
  console.log('Dosya okuma bitti');
});

readable.on('error', (error) => {
  console.error('Hata:', error);
});

Event-based çalışır; data event'i her chunk için tetiklenir.

4. Writable stream

Hedefe veri yazan stream.

const fs = require('fs');

const writable = fs.createWriteStream('output.txt');

writable.write('Birinci satır\n');
writable.write('İkinci satır\n');
writable.end('Son satır\n');

writable.on('finish', () => {
  console.log('Yazma bitti');
});

writable.on('error', (error) => {
  console.error('Hata:', error);
});

5. pipe() metodu

Readable'dan Writable'a doğrudan veri aktarımı. Backpressure otomatik yönetilir.

const fs = require('fs');

// Basit dosya kopyalama
fs.createReadStream('source.txt')
  .pipe(fs.createWriteStream('destination.txt'));

pipe() arka planda chunk yönetimini ve hız kontrolünü yapar. Readable hızlı, Writable yavaşsa otomatik durdurulur (backpressure).

6. Transform stream

Veriyi okurken/yazarken dönüştüren stream. Şifreleme, sıkıştırma, parse etme gibi işlemler için.

Örnek: gzip sıkıştırma

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

fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'));

Örnek: özel transform

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

const removeEmptyLines = new Transform({
  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    const filtered = lines.filter(line => line.trim().length > 0);
    this.push(filtered.join('\n'));
    callback();
  }
});

fs.createReadStream('messy.txt')
  .pipe(removeEmptyLines)
  .pipe(fs.createWriteStream('clean.txt'));

7. Modern pipeline API

pipe() chain'inde hata yönetimi karmaşıktı. Modern pipeline() helper'ı daha güvenli:

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

async function compressFile(input, output) {
  await pipeline(
    fs.createReadStream(input),
    zlib.createGzip(),
    fs.createWriteStream(output)
  );
  console.log('Sıkıştırma bitti');
}

compressFile('input.txt', 'input.txt.gz')
  .catch(error => console.error('Hata:', error));

pipeline tüm stream'leri otomatik bağlar; biri hata verirse hepsini temizler. Modern Node.js'te pipe() yerine pipeline() tercih edilir.

8. HTTP stream

HTTP request ve response Node.js'te stream'dir.

Response olarak dosya stream

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

const app = express();

app.get('/download/big-video', (req, res) => {
  const stream = fs.createReadStream('big-video.mp4');
  stream.pipe(res); // res Writable stream
});

// Tüm 1 GB videoyu belleğe yüklemez; parça parça gönderir

Request body stream

app.post('/upload', (req, res) => {
  // req Readable stream
  const writeStream = fs.createWriteStream('uploaded.bin');
  req.pipe(writeStream);

  req.on('end', () => res.send('Upload tamamlandı'));
});

9. Backpressure yönetimi

Readable hızlı, Writable yavaşsa veri birikir; bellek doluyor. Backpressure otomatik kontrol:

const writable = fs.createWriteStream('output.txt');

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);

  if (!canContinue) {
    readable.pause(); // Writable yavaş; oku duraklat
    writable.once('drain', () => {
      readable.resume(); // Tekrar yaz
    });
  }
});

pipe() ve pipeline() bunu otomatik yapar; manuel yönetim gerekmez. Sadece custom stream yazarken bilmek gerekir.

10. Object mode

Stream'ler varsayılan olarak Buffer/string aktarır. Object mode ile JavaScript nesnesi de aktarılabilir:

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

const parseJson = new Transform({
  readableObjectMode: true,
  writableObjectMode: false,
  transform(chunk, encoding, callback) {
    try {
      const obj = JSON.parse(chunk.toString());
      this.push(obj);
    } catch (error) {
      callback(error);
      return;
    }
    callback();
  }
});

MongoDB query result'ları object mode stream olarak gelir; her döküman bir chunk.

11. Pratik kullanım senaryoları

CSV parse

const fs = require('fs');
const csv = require('csv-parser');

fs.createReadStream('data.csv')
  .pipe(csv())
  .on('data', (row) => {
    console.log(row); // Her satır object olarak
  })
  .on('end', () => {
    console.log('CSV bitti');
  });

Büyük JSON parse

const fs = require('fs');
const StreamArray = require('stream-json/streamers/StreamArray');

fs.createReadStream('huge.json')
  .pipe(StreamArray.withParser())
  .on('data', ({ key, value }) => {
    console.log(value); // Her array elemanı
  });

Database query'leri stream olarak

// Mongoose
User.find().cursor()
  .on('data', (doc) => {
    console.log(doc);
  })
  .on('end', () => console.log('Bitti'));

// 1 milyon doc'u belleğe almadan işler

API endpoint streaming response

app.get('/api/large-data', async (req, res) => {
  res.setHeader('Content-Type', 'application/json');
  res.write('[');

  const cursor = User.find().cursor();
  let first = true;

  for await (const doc of cursor) {
    if (!first) res.write(',');
    res.write(JSON.stringify(doc));
    first = false;
  }

  res.write(']');
  res.end();
});

12. Tipik Hatalar

1. Hata yönetimini atlamak. Stream'lerde error event'i yakalanmalı
2. pipe() kullanırken hata yakalama. Her stream'e ayrı error handler gerekiyor; pipeline() bunu çözer
3. Buffer mode'da object göndermek. Object mode aktif olmadan obje aktarılamaz
4. Backpressure'ı görmezden gelmek. Manuel stream'de pause/resume yönetimi şart
5. writable.end() çağırmamak. Yazma tamamlanmaz; finish event tetiklenmez
6. Stream'i destroy etmemek. Erken sonlandırılan stream resource leak yaratır

13. Performans karşılaştırması

10-15 bin kat bellek tasarrufu. Üretim sistemleri için stream zorunlu değil ama mantıklı.

14. Atılması Gereken Adım

Stream Node.js'in en güçlü ve en az anlaşılan özelliklerinden. Yapıyı sağlam temellere oturtmak için Node.js eğitimi programları stream, async ve performance konularını birlikte aktarır.

15. Kilit Maddeler

Node.js stream API; büyük veriyi parça parça işleyerek bellek tüketimini binlerce kat azaltır. Readable, Writable, Transform ve Duplex olmak üzere dört tip. pipe() basit zincir; modern pipeline() async hata yönetimi ekler. HTTP request/response zaten stream; dosya işleme, CSV parse, DB cursor'ları doğal kullanım alanları. Backpressure pipe() ile otomatik; manuel yazarken pause/resume yönetimi şart. Stream öğrenmek modern backend için büyük yatırım getirisi.