[NODE JS] pipe ile Stream Yönlendirme

Silverhand · Pazartesi saat 10:45'de

[NODE JS] pipe ile Stream Yönlendirme

pipe metodu, stream'leri birbirine bağlamanın en temel ve zarif yoludur. Unix dünyasındaki pipe operatörünün (|) Node.js karşılığıdır. Bir Readable Stream'in çıkışını bir Writable Stream'in girişine bağlayarak, veriyi akıcı şekilde bir noktadan diğerine yönlendirirsiniz. Bu basit ama güçlü mekanizma, Node.js'in stream mimarisinin temelini oluşturur.

pipe kullanmanın en büyük avantajı, backpressure yönetimini otomatik olarak halletmesidir. Manuel olarak write() ve drain yönetimi yapmak yerine, pipe bu karmaşıklığı sizden gizler. Okuma hızı yazma hızından fazla olduğunda, pipe otomatik olarak okumayı durdurur. Yazma kuyruğu boşaldığında okumaya devam eder. Bu otomatik yönetim, bellek güvenli ve performanslı veri aktarımı sağlar.

İlk büyük dosya işleme projemde, pipe kullanmadan önce veriyi manuel olarak Readable'dan okuyup Writable'a yazıyordum. Backpressure yönetimini kendim yapmaya çalışıyordum ve kodun karmaşıklığı inanılmaz artmıştı. pipe'a geçtiğimde, aynı işi yapan kod yarı yarıya kısaldı ve hata oranı sıfıra düştü. O günden sonra her stream bağlantısında pipe kullanmayı alışkanlık edindim.

pipe metodunun zincirleme kullanımı, birden fazla Transform Stream'i seri şekilde bağlamanızı sağlar. readable.pipe(transform1).pipe(transform2).pipe(writable) şeklinde yazabilirsiniz. Her pipe çağrısı, hedef stream'i döndürdüğü için bu zincirleme mümkün olur. Bu yapı, veri işleme pipeline'larını son derece okunabilir hale getirir.

Ancak pipe'ın bir zayıf noktası vardır: hata yönetimi. pipe zincirinde bir stream hata verdiğinde, bu hata otomatik olarak diğer stream'lere yayılmaz. Her stream'e ayrı ayrı error olayı dinleyicisi eklemeniz gerekir. Bu durum, özellikle uzun pipe zincirlerinde kodun tekrarlı ve hata eğilimli olmasına yol açar. Bu sorunu çözmek için Node.js 10'dan itibaren pipeline fonksiyonu eklendi.

Unpipe işlemi de bilinmesi gereken bir konudur. Belirli koşullarda pipe bağlantısını kesmek isteyebilirsiniz. unpipe() metodu ile bağlantıyı sonlandırabilirsiniz. Bu özellik, dinamik stream routing senaryolarında kullanışlıdır. Örneğin, belirli bir koşula göre veriyi farklı hedeflere yönlendirebilirsiniz.

Performans açısından pipe oldukça verimlidir. Dahili tamponlama mekanizması, gereksiz bellek kopyalamasını minimize eder. highWaterMark ayarını doğru yapılandırarak, tampon boyutunu ihtiyacınıza göre optimize edebilirsiniz. Dosya kopyalama gibi basit işlemlerde, pipe yaklaşımı fs.readFile + fs.writeFile yaklaşımından çok daha verimlidir.

JavaScript:

// pipe ile Stream Yönlendirme - Kapsamlı Örnekler
const fs = require('fs');
const path = require('path');
const zlib = require('zlib');
const crypto = require('crypto');
const { Transform, Readable, Writable } = require('stream');
const http = require('http');

// ==========================================
// 1. Basit Dosya Kopyalama (pipe ile)
// ==========================================
function dosyaKopyala(kaynak, hedef) {
  return new Promise((resolve, reject) => {
    console.log('--- Dosya Kopyalama ---');
    console.log('Kaynak:', kaynak);
    console.log('Hedef:', hedef);

    const okumaAkisi = fs.createReadStream(kaynak);
    const yazmaAkisi = fs.createWriteStream(hedef);

    // Her stream'e hata dinleyicisi ekle
    okumaAkisi.on('error', (hata) => {
      console.error('Okuma hatası:', hata.message);
      reject(hata);
    });

    yazmaAkisi.on('error', (hata) => {
      console.error('Yazma hatası:', hata.message);
      reject(hata);
    });

    yazmaAkisi.on('finish', () => {
      console.log('Kopyalama tamamlandı');
      resolve();
    });

    // pipe ile bağla - backpressure otomatik yönetilir
    okumaAkisi.pipe(yazmaAkisi);
  });
}

// ==========================================
// 2. Sıkıştırma ile Dosya Kopyalama
// ==========================================
function dosyaSikistir(kaynak, hedef) {
  return new Promise((resolve, reject) => {
    console.log('\n--- Dosya Sıkıştırma (gzip) ---');

    const okumaAkisi = fs.createReadStream(kaynak);
    const sikistirici = zlib.createGzip({ level: 9 }); // Maksimum sıkıştırma
    const yazmaAkisi = fs.createWriteStream(hedef + '.gz');

    // Zincirleme pipe: oku -> sıkıştır -> yaz
    okumaAkisi
      .pipe(sikistirici)
      .pipe(yazmaAkisi);

    // Hata yönetimi - her stream için ayrı ayrı
    okumaAkisi.on('error', reject);
    sikistirici.on('error', reject);
    yazmaAkisi.on('error', reject);

    yazmaAkisi.on('finish', () => {
      console.log('Sıkıştırma tamamlandı: ' + hedef + '.gz');
      resolve();
    });
  });
}

// ==========================================
// 3. İlerleme Takipçisi Transform
// ==========================================
class IlerlemeIzleyici extends Transform {
  constructor(toplamBoyut, secenekler = {}) {
    super();
    this.toplamBoyut = toplamBoyut;
    this.islenenBoyut = 0;
    this.sonYuzde = 0;
    this.baslangicZamani = Date.now();
    this.raporAraligi = secenekler.raporAraligi || 10; // Yüzde cinsinden
  }

  _transform(parca, kodlama, geriCagir) {
    this.islenenBoyut += parca.length;

    const yuzde = Math.floor((this.islenenBoyut / this.toplamBoyut) * 100);

    if (yuzde >= this.sonYuzde + this.raporAraligi) {
      this.sonYuzde = yuzde;
      const gecenSure = ((Date.now() - this.baslangicZamani) / 1000).toFixed(1);
      const hiz = (this.islenenBoyut / 1024 / 1024 / parseFloat(gecenSure)).toFixed(2);
      console.log(
        'İlerleme: %' + yuzde + ' - '
        + (this.islenenBoyut / 1024 / 1024).toFixed(2) + 'MB / '
        + (this.toplamBoyut / 1024 / 1024).toFixed(2) + 'MB - '
        + hiz + ' MB/s'
      );
    }

    // Veriyi değiştirmeden aktar
    this.push(parca);
    geriCagir();
  }

  _flush(geriCagir) {
    const toplamSure = ((Date.now() - this.baslangicZamani) / 1000).toFixed(2);
    console.log('İşlem tamamlandı! Süre: ' + toplamSure + ' saniye');
    geriCagir();
  }
}

// ==========================================
// 4. Çoklu Hedef Yönlendirme (Tee)
// ==========================================
class StreamBolucusu extends Transform {
  constructor(hedefler = []) {
    super();
    this.hedefler = hedefler;
  }

  _transform(parca, kodlama, geriCagir) {
    // Veriyi tüm hedeflere gönder
    for (const hedef of this.hedefler) {
      hedef.write(parca);
    }

    // Ana akıştan da devam et
    this.push(parca);
    geriCagir();
  }

  _flush(geriCagir) {
    // Tüm hedefleri sonlandır
    for (const hedef of this.hedefler) {
      hedef.end();
    }
    geriCagir();
  }
}

// ==========================================
// 5. Koşullu Yönlendirme
// ==========================================
class KosulluYonlendirici extends Writable {
  constructor(kosullar) {
    super({ objectMode: true });
    this.kosullar = kosullar; // [{ test: fn, hedef: stream }]
    this.varsayilan = null;
    this.yonlendirmeSayilari = {};
  }

  varsayilanHedefAyarla(hedef) {
    this.varsayilan = hedef;
  }

  _write(kayit, kodlama, geriCagir) {
    let yonlendirildi = false;

    for (const kosul of this.kosullar) {
      if (kosul.test(kayit)) {
        const hedefAdi = kosul.isim || 'isimsiz';
        this.yonlendirmeSayilari[hedefAdi] =
          (this.yonlendirmeSayilari[hedefAdi] || 0) + 1;

        kosul.hedef.write(JSON.stringify(kayit) + '
');
        yonlendirildi = true;
        break; // İlk eşleşen koşula yönlendir
      }
    }

    if (!yonlendirildi && this.varsayilan) {
      this.yonlendirmeSayilari['varsayilan'] =
        (this.yonlendirmeSayilari['varsayilan'] || 0) + 1;
      this.varsayilan.write(JSON.stringify(kayit) + '
');
    }

    geriCagir();
  }

  _final(geriCagir) {
    console.log('\nYönlendirme istatistikleri:', this.yonlendirmeSayilari);

    // Tüm hedefleri kapat
    for (const kosul of this.kosullar) {
      kosul.hedef.end();
    }
    if (this.varsayilan) this.varsayilan.end();

    geriCagir();
  }
}

// ==========================================
// 6. HTTP Sunucu ile pipe Kullanımı
// ==========================================
function httpSunucuOrnegi() {
  const sunucu = http.createServer((req, res) => {
    if (req.method === 'GET' && req.url === '/dosya-indir') {
      const dosyaYolu = path.join(__dirname, 'buyuk-dosya.txt');

      // Dosya boyutunu öğren
      const bilgi = fs.statSync(dosyaYolu);

      res.writeHead(200, {
        'Content-Type': 'application/octet-stream',
        'Content-Disposition': 'attachment; filename="indirilen.txt"',
        'Content-Length': bilgi.size
      });

      // Dosyayı sıkıştırarak gönder
      const okumaAkisi = fs.createReadStream(dosyaYolu);
      const ilerleme = new IlerlemeIzleyici(bilgi.size);

      okumaAkisi
        .pipe(ilerleme)
        .pipe(res);

      okumaAkisi.on('error', (hata) => {
        console.error('Dosya okuma hatası:', hata.message);
        res.statusCode = 500;
        res.end('Sunucu hatası');
      });
    }

    if (req.method === 'POST' && req.url === '/dosya-yukle') {
      // Gelen veriyi dosyaya pipe et
      const hedefYol = path.join(__dirname, 'yuklenen-' + Date.now() + '.dat');
      const yazmaAkisi = fs.createWriteStream(hedefYol);

      req.pipe(yazmaAkisi);

      yazmaAkisi.on('finish', () => {
        res.writeHead(200, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({
          basarili: true,
          mesaj: 'Dosya yüklendi',
          dosya: hedefYol
        }));
      });

      yazmaAkisi.on('error', (hata) => {
        res.writeHead(500, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ basarili: false, hata: hata.message }));
      });
    }
  });

  return sunucu;
}

// ==========================================
// 7. Unpipe Örneği - Dinamik Yönlendirme
// ==========================================
function unpipeOrnegi() {
  console.log('\n--- Unpipe Örneği ---');

  const kaynak = Readable.from(['parça1
', 'parça2
', 'parça3
', 'parça4
']);

  const hedef1 = new Writable({
    write(parca, kodlama, geriCagir) {
      console.log('Hedef 1:', parca.toString().trim());
      geriCagir();
    }
  });

  const hedef2 = new Writable({
    write(parca, kodlama, geriCagir) {
      console.log('Hedef 2:', parca.toString().trim());
      geriCagir();
    }
  });

  // Başlangıçta hedef 1'e yönlendir
  kaynak.pipe(hedef1);

  // 2 saniye sonra hedef 1'den ayır, hedef 2'ye bağla
  setTimeout(() => {
    console.log('Yönlendirme değiştiriliyor...');
    kaynak.unpipe(hedef1);
    kaynak.pipe(hedef2);
  }, 100);
}

// ==========================================
// Ana Çalıştırıcı
// ==========================================
async function basla() {
  // Koşullu yönlendirme örneği
  console.log('--- Koşullu Yönlendirme ---');

  const logKaynagi = Readable.from([
    { seviye: 'HATA', mesaj: 'Veritabanı bağlantısı koptu', zaman: Date.now() },
    { seviye: 'BİLGİ', mesaj: 'Kullanıcı giriş yaptı', zaman: Date.now() },
    { seviye: 'UYARI', mesaj: 'Bellek kullanımı yüksek', zaman: Date.now() },
    { seviye: 'HATA', mesaj: 'Dosya bulunamadı', zaman: Date.now() },
    { seviye: 'DEBUG', mesaj: 'Sorgu çalıştırıldı', zaman: Date.now() },
    { seviye: 'BİLGİ', mesaj: 'Rapor oluşturuldu', zaman: Date.now() }
  ], { objectMode: true });

  // Hedef stream'ler (konsola yazan basit stream'ler)
  const hataHedefi = new Writable({
    write(p, k, cb) { console.log('[HATA LOG]', p.toString().trim()); cb(); }
  });
  const bilgiHedefi = new Writable({
    write(p, k, cb) { console.log('[BİLGİ LOG]', p.toString().trim()); cb(); }
  });
  const digerHedefi = new Writable({
    write(p, k, cb) { console.log('[DİĞER LOG]', p.toString().trim()); cb(); }
  });

  const yonlendirici = new KosulluYonlendirici([
    { isim: 'hatalar', test: (k) => k.seviye === 'HATA', hedef: hataHedefi },
    { isim: 'bilgiler', test: (k) => k.seviye === 'BİLGİ', hedef: bilgiHedefi }
  ]);
  yonlendirici.varsayilanHedefAyarla(digerHedefi);

  logKaynagi.pipe(yonlendirici);

  await new Promise(resolve => yonlendirici.on('finish', resolve));

  // Unpipe örneği
  unpipeOrnegi();
}

basla().catch(console.error);

pipe, stream'leri bağlamanın en doğal ve güvenli yoludur. Backpressure yönetimini otomatikleştirir, kodu basitleştirir ve hata olasılığını azaltır. Ancak hata yönetimi konusundaki kısıtlamaları nedeniyle, özellikle üretim kodunda pipeline fonksiyonunu tercih etmenizi öneririm. Bir sonraki yazıda stream pipeline kullanımını derinlemesine inceleyeceğiz.

Best Studio · Çarşamba saat 16:17'de

Emeğiniz ve katkınız için teşekkür ederiz!
Konunuz gerekli incelemelerden geçerek onaylandı. Topluluğumuza katkılarınızın devamını bekliyoruz.
İyi forumlar!

[NODE JS] pipe ile Stream Yönlendirme

Silverhand