pipe dan pipeline

Kita kini memiliki seluruh komponennya: Readable menghasilkan data, Writable menerimanya, dan Transform mengubahnya di tengah. Komponen-komponen tersebut menjadi benar-benar berguna ketika disambungkan menjadi satu rangkaian — itulah makna harfiah kata pipeline pada dunia stream. Node.js menyediakan dua alat penyambung: method pipe() yang telah kita pakai beberapa kali, dan function pipeline() yang lebih mumpuni dalam menangani keadaan darurat. Chapter ini membandingkan keduanya dan menunjukkan mengapa pipeline() menjadi pilihan utama pada kode modern.

Menghubungkan Stream dengan pipe #

Method pipe() pada Readable mengalirkan setiap potongan data ke stream tujuan secara otomatis, lengkap dengan pengaturan tempo antar keduanya — perilaku lengkapnya dirinci pada dokumentasi readable.pipe() . Contoh berikut menyalin isi sumber.txt ke salinan.txt tanpa pernah memuat seluruh berkas ke satu variabel:

import { createReadStream, createWriteStream } from "node:fs";
import { readFile, rm } from "node:fs/promises";

const sumber = createReadStream("sumber.txt");
const salinan = createWriteStream("salinan.txt");

sumber.pipe(salinan);

salinan.on("finish", async () => {
  const isi = await readFile("salinan.txt", "utf8");
  console.log("isi salinan:");
  process.stdout.write(isi);
  await rm("salinan.txt", { force: true });
});

Output yang Dihasilkan:

isi salinan:
Halo dari berkas sumber.
Baris kedua di sini.

Baris 7 (sumber.pipe(salinan)) adalah inti contoh ini: sejak panggilan tersebut, setiap potongan yang terbaca dari sumber langsung dituliskan ke salinan, dan sumber otomatis berhenti membaca ketika tujuan sedang kewalahan — mekanisme backpressure yang akan kita bedah pada chapter berikutnya. pipe() juga mengembalikan stream tujuan, sehingga rangkaian bisa ditulis berantai: a.pipe(b).pipe(c) mengalirkan data dari a melalui b ke c. Event finish didaftarkan pada tujuan untuk menjalankan verifikasi dan pembersihan setelah seluruh data tergelak.

Kelemahan pipe dalam Penanganan Error #

Kelemahan pipe() terletak pada keadaan darurat. Method ini tidak meneruskan error antar stream: jika sumber gagal di tengah jalan, salinan tidak pernah diberi tahu dan menunggu data yang tidak akan pernah datang; sebaliknya, kegagalan di tujuan tidak menghentikan sumber yang terus membaca untuk tujuan yang sudah tidak berfungsi. Menangani kedua kemungkinan itu secara manual membutuhkan pendengar event yang banyak:

sumber.on("error", salinDanBerhenti);
salinan.on("error", salinDanBerhenti);

function salinDanBerhenti(galat) {
  console.error("gagal:", galat.message);
  sumber.destroy();
  salinan.destroy();
}

Potongan di atas hanyalah kerangka: salinDanBerhenti masih harus dipastikan tidak berjalan dua kali, berkas tujuan masih harus ditutup, dan sisa data masih harus dibersihkan. Di sinilah letak godaya menulis kode yang mengabaikan penanganan error sama sekali — dan kebocoran sumber daya yang mengikutinya.

pipeline: Perangkai yang Lebih Aman #

Function pipeline() dari module node:stream/promises merangkai stream dengan disiplin penuh: satu error di titik mana pun menghentikan seluruh rangkaian, semua stream terkait ditutup, dan error tersebut dilaporkan pada satu tempat. Versi promise dari seluruh function utilitas stream dirangkum pada bagian Stream promises API . Contoh berikut mensimulasikan kegagalan di tengah pipa:

import { Readable, Writable, Transform } from "node:stream";
import { pipeline } from "node:stream/promises";

const pemeriksa = new Transform({
  transform(potongan, _encoding, callback) {
    if (potongan.toString().includes("rusak")) {
      callback(new Error("data rusak terdeteksi"));
      return;
    }
    callback(null, potongan);
  },
});

const penampung = new Writable({
  write(potongan, _encoding, callback) {
    console.log("diterima:", potongan.toString());
    callback();
  },
});

try {
  await pipeline(
    Readable.from(["data-satu", "data-dua", "data-rusak", "data-empat"]),
    pemeriksa,
    penampung,
  );
  console.log("pipeline selesai");
} catch (galat) {
  console.error("pipeline berhenti:", galat.message);
  process.exitCode = 1;
}

Output yang Dihasilkan:

diterima: data-satu
diterima: data-dua
pipeline berhenti: data rusak terdeteksi

Mari kita telaah bagian inti contoh di atas:

1. Baris 5-12 (pemeriksa): Transform melewatkan potongan apa adanya, kecuali potongan itu memuat kata rusak — saat itulah callback dipanggil dengan new Error(...) untuk mensimulasikan kegagalan nyata seperti berkas korup.
2. Baris 21-27 (await pipeline(...)): versi promise dari pipeline menerima stream apa saja secara beruntun dan selesai ketika seluruh data telah mengalir. Rangkaian berhenti tepat pada data-rusak — data-empat tidak pernah mencapai penampung.
3. Baris 28-31 (catch): error dari titik mana pun di rangkaian muncul di satu tempat ini saja. Menyimpan process.exitCode = 1 memberi tahu pemanggil skrip bahwa proses tidak selesai sempurna, tanpa menghentikan proses secara paksa di tengah pembersihan.

Memilih dengan Tepat #

Bentuk callback dari pipeline() punya kegunaan tersendiri pada kode non-async — keduanya didokumentasikan berdampingan pada rujukan stream.pipeline() .

Menyambung stream memangkas pekerjaan manual, tetapi satu pertanyaan masih menggantung: apa yang sebenarnya terjadi ketika sumber data jauh lebih cepat daripada tujuan? pipe() dan pipeline() menanganinya untuk kita, dan Chapter Backpressure membongkar mekanisme di baliknya — highWaterMark, nilai balik write(), serta jalinan pause() dan resume().