[NODE JS] Stream Pipeline Kullanımı

Silverhand · Pazartesi saat 10:45'de

[NODE JS] Stream Pipeline Kullanımı

Stream pipeline, Node.js 10 ile birlikte gelen ve pipe metodunun zayıf yönlerini gideren güçlü bir araçtır. pipe ile yapılan zincirleme bağlantılarda hata yönetimi büyük bir sorundu. Her stream'e ayrı ayrı error dinleyicisi eklemeniz gerekiyordu ve bir hata oluştuğunda diğer stream'lerin düzgün temizlenmesi garanti değildi. pipeline fonksiyonu bu sorunları tek seferde çözer.

pipeline fonksiyonu, stream'leri sıralı şekilde birbirine bağlar ve zincirdeki herhangi bir stream hata verdiğinde, tüm stream'leri otomatik olarak temizler (destroy eder). Ayrıca bir callback ile tüm işlemin başarılı mı yoksa hatalı mı tamamlandığını bildirir. Bu, kaynak sızıntılarını önler ve kodunuzu çok daha güvenilir hale getirir.

util.promisify ile pipeline'ı Promise tabanlı hale getirebilirsiniz. Bu sayede async/await ile kullanabilir ve hata yönetimini try/catch bloklarıyla yapabilirsiniz. Modern Node.js projelerinde bu yaklaşım standart haline gelmiştir.

Bir dosya işleme projesinde, pipe kullanırken aralıklı bellek sızıntıları yaşıyorduk. Hata durumlarında stream'ler düzgün kapatılmıyordu ve açık dosya tanımlayıcıları birikiyordu. pipeline'a geçtiğimizde bu sorunlar tamamen ortadan kalktı. Her hata durumunda tüm stream'ler otomatik olarak temizleniyor, kaynaklar serbest bırakılıyordu.

Node.js 15 ve sonrasında stream.pipeline artık doğrudan Promise döndürebiliyor. Yani util.promisify'a bile gerek kalmıyor. Ayrıca async generator'lar ile pipeline kullanmak da mümkün. Bu özellik, fonksiyonel programlama stilinde veri dönüşüm pipeline'ları oluşturmayı kolaylaştırır.

AbortController ile pipeline'ı iptal edebilmek de önemli bir özelliktir. Uzun süren işlemlerde, kullanıcı isteği iptal ettiğinde veya zaman aşımı oluştuğunda, pipeline'ı güvenli şekilde durdurabilirsiniz. Bu özellik, özellikle HTTP sunucularında istemci bağlantıyı kopardığında kaynakları serbest bırakmak için kritiktir.

pipeline ile çalışırken dikkat edilmesi gereken bir nokta, zincirdeki her stream'in uyumlu olmasıdır. objectMode olan bir stream ile olmayan bir stream'i aynı pipeline'da kullanmak hataya yol açar. Dönüşüm stream'lerinizde readableObjectMode ve writableObjectMode ayarlarını doğru yapılandırmanız gerekir.

İzleme ve loglama açısından, pipeline'ın başlangıç ve bitiş zamanlarını ölçmek, işlenen veri miktarını takip etmek ve hata istatistiklerini toplamak, üretim ortamında sorunları tespit etmenize yardımcı olur. Bu bilgileri bir monitoring sistemine göndererek, veri işleme pipeline'larınızın sağlığını sürekli izleyebilirsiniz.

JavaScript:

// Stream Pipeline Kullanımı - Kapsamlı Örnekler
const { pipeline, Readable, Transform, Writable } = require('stream');
const { promisify } = require('util');
const fs = require('fs');
const path = require('path');
const zlib = require('zlib');
const crypto = require('crypto');

const pipelineAsync = promisify(pipeline);

// ==========================================
// 1. Temel Pipeline Kullanımı
// ==========================================
async function temelPipeline() {
  console.log('--- Temel Pipeline ---');

  // Kaynak: dosya okuma
  // Dönüşüm: büyük harfe çevirme
  // Hedef: dosyaya yazma

  const buyukHarfDonusturucu = new Transform({
    transform(parca, kodlama, geriCagir) {
      this.push(parca.toString().toUpperCase());
      geriCagir();
    }
  });

  const kaynakVeri = Readable.from([
    'merhaba dünya!
',
    'node.js stream pipeline örneği.
',
    'türkçe karakterler: ş, ç, ğ, ü, ö, ı.
',
    'büyük harfe dönüştürme testi.
'
  ]);

  const sonucParcalari = [];
  const hedef = new Writable({
    write(parca, kodlama, geriCagir) {
      sonucParcalari.push(parca.toString());
      geriCagir();
    }
  });

  try {
    await pipelineAsync(
      kaynakVeri,
      buyukHarfDonusturucu,
      hedef
    );
    console.log('Pipeline başarıyla tamamlandı');
    console.log('Sonuç:', sonucParcalari.join(''));
  } catch (hata) {
    console.error('Pipeline hatası:', hata.message);
  }
}

// ==========================================
// 2. Çok Aşamalı Veri İşleme Pipeline'ı
// ==========================================

// Adım 1: Satır Ayırıcı
class SatirAyirici extends Transform {
  constructor() {
    super({ readableObjectMode: true });
    this.kalanVeri = '';
  }

  _transform(parca, kodlama, geriCagir) {
    const veri = this.kalanVeri + parca.toString();
    const satirlar = veri.split('
');
    this.kalanVeri = satirlar.pop() || '';

    for (const satir of satirlar) {
      if (satir.trim()) {
        this.push(satir.trim());
      }
    }
    geriCagir();
  }

  _flush(geriCagir) {
    if (this.kalanVeri.trim()) {
      this.push(this.kalanVeri.trim());
    }
    geriCagir();
  }
}

// Adım 2: JSON Ayrıştırıcı
class JsonAyristirici extends Transform {
  constructor() {
    super({ objectMode: true });
    this.basarisiz = 0;
  }

  _transform(satir, kodlama, geriCagir) {
    try {
      const nesne = JSON.parse(satir);
      this.push(nesne);
    } catch (hata) {
      this.basarisiz++;
      // Geçersiz JSON satırlarını atla, pipeline'ı durdurmak yerine
      console.warn('Geçersiz JSON satırı atlandı:', satir.substring(0, 50));
    }
    geriCagir();
  }

  _flush(geriCagir) {
    if (this.basarisiz > 0) {
      console.log('Atlanan geçersiz satır sayısı:', this.basarisiz);
    }
    geriCagir();
  }
}

// Adım 3: Veri Doğrulayıcı
class VeriDogrulayici extends Transform {
  constructor(sema) {
    super({ objectMode: true });
    this.sema = sema;
    this.gecerli = 0;
    this.gecersiz = 0;
  }

  _transform(kayit, kodlama, geriCagir) {
    const hatalar = [];

    for (const [alan, kural] of Object.entries(this.sema)) {
      if (kural.zorunlu && (kayit[alan] === undefined || kayit[alan] === null)) {
        hatalar.push(alan + ' alanı zorunludur');
      }
      if (kural.tip && typeof kayit[alan] !== kural.tip) {
        hatalar.push(alan + ' alanı ' + kural.tip + ' olmalıdır');
      }
      if (kural.minDeger && kayit[alan] < kural.minDeger) {
        hatalar.push(alan + ' değeri en az ' + kural.minDeger + ' olmalıdır');
      }
    }

    if (hatalar.length === 0) {
      this.gecerli++;
      this.push(kayit);
    } else {
      this.gecersiz++;
      // Hatalı kayıtları loglama (opsiyonel)
    }
    geriCagir();
  }

  _flush(geriCagir) {
    console.log('Doğrulama: Geçerli=' + this.gecerli + ', Geçersiz=' + this.gecersiz);
    geriCagir();
  }
}

// Adım 4: Toplu İşleyici
class TopluIsleyici extends Writable {
  constructor(isleyiciFn, secenekler = {}) {
    super({ objectMode: true });
    this.isleyici = isleyiciFn;
    this.topluBoyut = secenekler.topluBoyut || 50;
    this.tampon = [];
    this.toplamIslenen = 0;
  }

  async _write(kayit, kodlama, geriCagir) {
    this.tampon.push(kayit);

    if (this.tampon.length >= this.topluBoyut) {
      try {
        await this.topluIsle();
        geriCagir();
      } catch (hata) {
        geriCagir(hata);
      }
    } else {
      geriCagir();
    }
  }

  async _final(geriCagir) {
    try {
      if (this.tampon.length > 0) {
        await this.topluIsle();
      }
      console.log('Toplu işleme tamamlandı. Toplam:', this.toplamIslenen);
      geriCagir();
    } catch (hata) {
      geriCagir(hata);
    }
  }

  async topluIsle() {
    const kayitlar = this.tampon.splice(0);
    await this.isleyici(kayitlar);
    this.toplamIslenen += kayitlar.length;
    console.log('Toplu işlendi: ' + kayitlar.length + ' kayıt (Toplam: ' + this.toplamIslenen + ')');
  }
}

// ==========================================
// 3. AbortController ile Pipeline İptali
// ==========================================
async function iptalEdilebilirPipeline() {
  console.log('\n--- İptal Edilebilir Pipeline ---');

  const controller = new AbortController();
  const { signal } = controller;

  // 2 saniye sonra iptal et
  const zamanAsimi = setTimeout(() => {
    console.log('Zaman aşımı! Pipeline iptal ediliyor...');
    controller.abort();
  }, 2000);

  // Yavaş veri kaynağı
  async function* yavasKaynak() {
    let sayac = 0;
    while (true) {
      sayac++;
      yield JSON.stringify({ id: sayac, veri: 'Kayıt ' + sayac }) + '
';
      await new Promise(resolve => setTimeout(resolve, 100));
    }
  }

  const sonuclar = [];
  const toplayici = new Writable({
    write(parca, kodlama, geriCagir) {
      sonuclar.push(parca.toString().trim());
      geriCagir();
    }
  });

  try {
    await pipelineAsync(
      Readable.from(yavasKaynak()),
      toplayici,
      { signal }
    );
  } catch (hata) {
    if (hata.code === 'ABORT_ERR') {
      console.log('Pipeline iptal edildi. İşlenen kayıt:', sonuclar.length);
    } else {
      throw hata;
    }
  } finally {
    clearTimeout(zamanAsimi);
  }
}

// ==========================================
// 4. Async Generator ile Pipeline
// ==========================================
async function asyncGeneratorPipeline() {
  console.log('\n--- Async Generator Pipeline ---');

  // Veri kaynağı
  const veriler = [
    { isim: 'Laptop', fiyat: 25000, kategori: 'Elektronik' },
    { isim: 'Tişört', fiyat: 150, kategori: 'Giyim' },
    { isim: 'Roman', fiyat: 45, kategori: 'Kitap' },
    { isim: 'Telefon', fiyat: 15000, kategori: 'Elektronik' },
    { isim: 'Pantolon', fiyat: 350, kategori: 'Giyim' },
    { isim: 'Kulaklık', fiyat: 899, kategori: 'Elektronik' },
    { isim: 'Ansiklopedi', fiyat: 200, kategori: 'Kitap' },
    { isim: 'Tablet', fiyat: 12000, kategori: 'Elektronik' }
  ];

  // Async generator fonksiyonlar ile dönüşüm adımları
  async function* fiyatFiltresi(kaynak) {
    for await (const kayit of kaynak) {
      if (kayit.fiyat > 100) {
        yield kayit;
      }
    }
  }

  async function* kdvHesaplayici(kaynak) {
    for await (const kayit of kaynak) {
      yield {
        ...kayit,
        kdv: (kayit.fiyat * 0.20).toFixed(2),
        kdvliFiyat: (kayit.fiyat * 1.20).toFixed(2)
      };
    }
  }

  async function* formatci(kaynak) {
    for await (const kayit of kaynak) {
      yield kayit.isim + ' | ' + kayit.kategori
        + ' | Fiyat: ' + kayit.fiyat + ' TL'
        + ' | KDV: ' + kayit.kdv + ' TL'
        + ' | Toplam: ' + kayit.kdvliFiyat + ' TL
';
    }
  }

  const sonuclar = [];
  const toplayici = new Writable({
    write(parca, kodlama, geriCagir) {
      sonuclar.push(parca.toString().trim());
      process.stdout.write(parca);
      geriCagir();
    }
  });

  await pipelineAsync(
    Readable.from(veriler),
    fiyatFiltresi,
    kdvHesaplayici,
    formatci,
    toplayici
  );

  console.log('\nToplam işlenen ürün: ' + sonuclar.length);
}

// ==========================================
// 5. Hata Yönetimi ve İzleme ile Pipeline
// ==========================================
class PipelineIzleyici {
  constructor(isim) {
    this.isim = isim;
    this.baslangic = null;
    this.bitis = null;
    this.islenenParcaSayisi = 0;
    this.toplamBoyut = 0;
    this.hatalar = [];
  }

  izlemeTransformuOlustur(adimAdi) {
    const izleyici = this;
    return new Transform({
      objectMode: true,
      transform(parca, kodlama, geriCagir) {
        izleyici.islenenParcaSayisi++;
        if (typeof parca === 'string') {
          izleyici.toplamBoyut += Buffer.byteLength(parca);
        } else if (Buffer.isBuffer(parca)) {
          izleyici.toplamBoyut += parca.length;
        }
        this.push(parca);
        geriCagir();
      }
    });
  }

  basla() {
    this.baslangic = Date.now();
    console.log('[' + this.isim + '] Pipeline başladı');
  }

  bitir(hata) {
    this.bitis = Date.now();
    const sure = ((this.bitis - this.baslangic) / 1000).toFixed(3);

    if (hata) {
      this.hatalar.push(hata.message);
      console.error('[' + this.isim + '] Pipeline hata ile bitti: ' + hata.message);
    } else {
      console.log('[' + this.isim + '] Pipeline başarıyla tamamlandı');
    }

    console.log('[' + this.isim + '] Süre: ' + sure + 's, '
      + 'Parça: ' + this.islenenParcaSayisi + ', '
      + 'Boyut: ' + (this.toplamBoyut / 1024).toFixed(2) + 'KB');
  }
}

async function izlemePipelineOrnegi() {
  console.log('\n--- İzlemeli Pipeline ---');

  const izleyici = new PipelineIzleyici('VeriIslemePipeline');

  const kaynak = Readable.from([
    '{"isim":"Ahmet","puan":85}
',
    '{"isim":"Ayşe","puan":92}
',
    '{"isim":"Can","puan":78}
',
    '{"isim":"Deniz","puan":95}
',
    '{"isim":"Elif","puan":88}
'
  ]);

  const ayristirici = new Transform({
    objectMode: true,
    transform(parca, kodlama, geriCagir) {
      try {
        const satir = parca.toString().trim();
        if (satir) {
          this.push(JSON.parse(satir));
        }
        geriCagir();
      } catch (h) {
        geriCagir(h);
      }
    }
  });

  const notHesaplayici = new Transform({
    objectMode: true,
    transform(kayit, kodlama, geriCagir) {
      let harf;
      if (kayit.puan >= 90) harf = 'AA';
      else if (kayit.puan >= 85) harf = 'BA';
      else if (kayit.puan >= 80) harf = 'BB';
      else if (kayit.puan >= 75) harf = 'CB';
      else harf = 'CC';

      this.push({ ...kayit, harfNotu: harf });
      geriCagir();
    }
  });

  const sonucYazici = new Writable({
    objectMode: true,
    write(kayit, kodlama, geriCagir) {
      console.log('  ' + kayit.isim + ': ' + kayit.puan + ' puan -> ' + kayit.harfNotu);
      geriCagir();
    }
  });

  izleyici.basla();

  try {
    await pipelineAsync(
      kaynak,
      ayristirici,
      izleyici.izlemeTransformuOlustur('ayrıştırma'),
      notHesaplayici,
      izleyici.izlemeTransformuOlustur('not_hesaplama'),
      sonucYazici
    );
    izleyici.bitir(null);
  } catch (hata) {
    izleyici.bitir(hata);
  }
}

// ==========================================
// Ana Çalıştırıcı
// ==========================================
async function basla() {
  try {
    await temelPipeline();
    await asyncGeneratorPipeline();
    await iptalEdilebilirPipeline();
    await izlemePipelineOrnegi();

    console.log('\nTüm pipeline örnekleri tamamlandı!');
  } catch (hata) {
    console.error('Beklenmeyen hata:', hata);
  }
}

basla();

Stream pipeline, modern Node.js'te veri işleme pipeline'ları oluşturmanın en güvenilir yoludur. pipe'ın hata yönetimi sorunlarını çözer, kaynak temizliğini otomatikleştirir ve async/await ile uyumlu çalışır. Yeni projelerinizde pipe yerine pipeline kullanmanızı kesinlikle öneririm. Bu şekilde hem kodunuz daha güvenilir olur hem de hata ayıklama süreci kolaylaşır.

Best Studio · Çarşamba saat 16:16'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] Stream Pipeline Kullanımı

Silverhand