Sepanjang Section ini kita beruntung: sumber data kecil dan konsumen selalu mengejar. Kenyataannya, kecepatan dua sisi stream sering tidak seimbang — berkas terbaca lebih cepat daripada jaringan mengirim, atau pengguna mengunggah lebih cepat daripada basis data menulis. Tanpa pengendalian, data yang belum sempat diproses menumpuk di memori tepat seperti yang ingin dihindari dengan stream. Node.js menyelesaikan masalah ini lewat protokol bernama backpressure: seperangkat sinyal yang membuat produsen menunggu ketika konsumen kewalahan. Chapter penutup Section ini membongkar protokol tersebut sampai ke dasarnya.

Ketika Produsen Lebih Cepat dari Konsumen #

Bayangkan selang air yang diisi pompa kencang namun ujungnya hanya menetes. Air harus menumpuk di suatu tempat — dan dalam dunia stream, tempat itu adalah memori. Backpressure membalik situasi ini: aliran dari sumber ditekan mundur agar mengikuti tempo tujuan.

Protokolnya bertumpu pada dua sinyal yang telah kita jumpai sepanjang Section ini — pembahasan resminya ada pada bagian Buffering di dokumentasi stream:

  1. Nilai balik writable.write(): false berarti buffer internal tujuan telah mencapai ambang highWaterMark; produsen bijak menahan diri sampai event drain tiba. Keadaan buffer ini juga bisa dipantau lewat properti writable.writableNeedDrain .
  2. Nilai balik readable.push(): false berarti buffer internal sumber penuh; sumber berhenti membaca dari resource aslinya sampai buffer dikosongkan konsumen.

Protokol Backpressure Secara Manual #

Cara terbaik memahami protokol ini adalah menuliskannya sendiri. Contoh berikut menyambungkan Readable dan Writable tanpa pipe(): produsen menulis sendiri setiap potongan sambil mematuhi sinyal dari tujuan:

js 
import { Readable, Writable } from "node:stream";

const produsen = Readable.from([
  "a1", "a2", "a3", "a4", "a5", "a6",
]);

const konsumen = new Writable({
  objectMode: true,
  highWaterMark: 2,
  write(potongan, _encoding, callback) {
    setImmediate(() => {
      console.log("selesai memproses", potongan);
      callback();
    });
  },
});

for await (const potongan of produsen) {
  const lega = konsumen.write(potongan);
  console.log("kirim", potongan, "->", lega);
  if (!lega) {
    await new Promise((resolve) => konsumen.once("drain", resolve));
    console.log("drain: produsen melanjutkan");
  }
}
konsumen.end();
await new Promise((resolve) => konsumen.once("finish", resolve));
console.log("selesai");

Output yang Dihasilkan:

text 
kirim a1 -> true
kirim a2 -> false
selesai memproses a1
selesai memproses a2
drain: produsen melanjutkan
kirim a3 -> true
kirim a4 -> false
selesai memproses a3
selesai memproses a4
drain: produsen melanjutkan
kirim a5 -> true
kirim a6 -> false
selesai memproses a5
selesai memproses a6
drain: produsen melanjutkan
selesai

Mari kita telaah baris-baris kode di atas:

  1. Baris 7-16 (konsumen): Writable dengan highWaterMark: 2 dan objectMode — buffer internalnya hanya menampung dua potongan. Method write() menunda callback() lewat setImmediate untuk mensimulasikan pekerjaan yang memakan waktu satu giliran event loop.
  2. Baris 18 (for await (const potongan of produsen)): produsen mengambil satu potongan per iterasi dari stream sumber.
  3. Baris 19-20 (konsumen.write(potongan)): potongan dikirim ke konsumen; nilai baliknya dicatat. Pengiriman pertama menghasilkan true, pengiriman kedua membuat buffer penuh dan menghasilkan false.
  4. Baris 21-23 (await new Promise(...)): inilah jantung protokolnya — produsen berhenti mengambil potongan berikutnya sampai event drain menyatakan buffer konsumen kosong. Baris kirim a3 baru muncul setelah selesai memproses a1 dan a2.
  5. Baris 26-28 (konsumen.end()): setelah semua potongan terkirim, stream diakhiri dan event finish ditunggu sebagai bukti seluruh data tersampaikan.

Pola yang tampak pada output — dua kiriman, dua pemrosesan, satu drain, berulang tiga kali — adalah protokol backpressure yang bekerja penuh tanpa pipe().

pipe Menjalankan Protokol Ini Otomatis #

Menuliskan protokol manual pada setiap kode jelas tidak praktis. Method pipe() dan function pipeline() menjalankan protokol yang sama secara otomatis. Contoh berikut memakai pipe() dengan Readable yang dibangun lewat read(), sehingga nilai balik push() ikut terlihat:

js 
import { Readable, Writable } from "node:stream";

const TOTAL = 8;
let dibuat = 0;

const produsen = new Readable({
  objectMode: true,
  highWaterMark: 2,
  read() {
    while (dibuat < TOTAL) {
      dibuat++;
      const lega = this.push(`potongan-${dibuat}`);
      console.log("push potongan-" + dibuat, "->", lega);
      if (!lega) return;
    }
    this.push(null);
  },
});

let diproses = 0;
const konsumenLambat = new Writable({
  objectMode: true,
  highWaterMark: 2,
  write(potongan, _encoding, callback) {
    diproses++;
    setImmediate(() => {
      console.log("konsumen selesai:", potongan);
      callback();
    });
  },
});

produsen.pipe(konsumenLambat).on("finish", () => {
  console.log("selesai, diproses:", diproses);
});

Output yang Dihasilkan:

text 
push potongan-1 -> true
push potongan-2 -> false
push potongan-3 -> false
push potongan-4 -> false
konsumen selesai: potongan-1
konsumen selesai: potongan-2
push potongan-5 -> false
push potongan-6 -> false
konsumen selesai: potongan-3
konsumen selesai: potongan-4
push potongan-7 -> false
push potongan-8 -> false
konsumen selesai: potongan-5
konsumen selesai: potongan-6
konsumen selesai: potongan-7
konsumen selesai: potongan-8
selesai, diproses: 8

Mari kita telaah bagian inti contoh di atas:

  1. Baris 9-16 (read()): method read() dipanggil stream ketika buffer internalnya siap menerima data. Loop mendorong potongan satu per satu dan berhenti seketika push() mengembalikan false — baris 14 (if (!lega) return;) adalah titik di mana produsen menurunkan tekanannya.
  2. Baris 19-29 (konsumenLambat): konsumen memproses satu potongan per giliran event loop, sama seperti versi manual.
  3. Baris 31 (produsen.pipe(konsumenLambat)): tanpa satu baris pun menunggu drain, pipe() menjahit protokolnya sendiri. Urutan pada output membuktikannya: push potongan-5 dan push potongan-6 baru terjadi setelah konsumen menyelesaikan potongan-1 dan potongan-2; delapan potongan diproduksi, delapan diproses, dan tidak ada yang menumpuk.

Peran highWaterMark #

highWaterMark adalah ambang yang menentukan kapan sinyal “penuh” dinyalakan — untuk stream byte ukurannya dalam byte, untuk objectMode dalam jumlah object:

js 
import { Readable, Writable } from "node:stream";

const tulis = new Writable({ write(_c, _e, cb) { cb(); } });
const baca = new Readable({ read() {} });
const tulisObjek = new Writable({ objectMode: true, write(_c, _e, cb) { cb(); } });

console.log("Writable byte       :", tulis.writableHighWaterMark);
console.log("Readable byte       :", baca.readableHighWaterMark);
console.log("Writable objectMode :", tulisObjek.writableHighWaterMark);

Output yang Dihasilkan:

text 
Writable byte       : 65536
Readable byte       : 65536
Writable objectMode : 16

Pada Node.js 24, ambang bawaan stream byte adalah 65536 byte (64 KiB), sedangkan objectMode memakai 16 object. Nilai ambang ini bisa dibaca kembali lewat properti readableHighWaterMark dan writableHighWaterMark.

Backpressure menutup petualangan kita dengan data bertahap: Buffer menyediakan wadah byte, stream mengalirkannya potongan demi potongan, dan protokol ini menjaga kedua sisi tetap seimbang. Seluruh mekanisme yang kita pelajari sepanjang Section ini berdiri di atas satu fondasi yang sama — event loop yang menjadwalkan drain, setImmediate, dan seluruh event lain secara adil. Fondasi itulah pokok Section Asynchronous Runtime.