Setelah memahami peta besar stream, kita mulai dari sisi yang menghasilkan data: Readable. Hampir semua sumber data di Node.js — berkas, koneksi jaringan, stdin — hadir dalam bentuk stream ini, dan perilakunya dirinci pada dokumentasi class stream.Readable . Mengonsumsinya dengan benar artinya kita bisa mulai memproses data sejak potongan pertama tiba, bukan menunggu seluruh data terkumpul. Chapter ini membahas dua cara konsumsi utama (event data dan async iterator), pengendalian aliran dengan pause() dan resume(), serta pola praktis membaca berkas per potongan dan per baris.

Dua Mode Kerja Readable #

Sebuah Readable stream berada dalam salah satu dari dua mode kerja. Mode jeda (paused) adalah keadaan awal: data tersedia di buffer internal, tetapi tidak mengalir ke mana pun sampai kita bertanya. Mode mengalir (flowing) menghidupkan aliran: setiap potongan data didorong keluar begitu tiba. Tiga cara memasuki mode mengalir yang paling umum:

  1. Mendaftarkan listener event data.
  2. Memanggil method resume().
  3. Menyambungkan stream ke tujuan dengan pipe().

Function Readable.from() yang akan sering kita pakai di chapter ini membuat stream dari iterable apa pun — array, generator, hingga async generator — sehingga kita bisa berlatih tanpa berkas atau jaringan.

Konsumsi dengan Event data #

Cara paling mendasar menerima data adalah mendengarkan event data, yang terpicu sekali per potongan, diikuti event end sebagai penanda sumber habis:

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

const kota = Readable.from(["Jakarta", "Bandung", "Surabaya"]);

kota.on("data", (potongan) => {
  console.log("menerima:", potongan);
});

kota.on("end", () => {
  console.log("data habis");
});

Output yang Dihasilkan:

text 
menerima: Jakarta
menerima: Bandung
menerima: Surabaya
data habis

Mendaftarkan listener data pada baris 5 langsung memindahkan stream ke mode mengalir. Setiap elemen array dikirim sebagai satu potongan, sehingga console.log dijalankan tiga kali sebelum event end tiba. Model event seperti ini bekerja baik untuk alur sederhana, tetapi callback bersarang membuat alur kompleks — misalnya memproses data lalu menulis hasilnya — sulit dibaca.

Konsumsi dengan Async Iterator #

Sejak Node.js 10, seluruh Readable stream adalah async iterable. Artinya kita boleh memakai for await...of dan memperlakukan stream seperti array yang datang secara asynchronous:

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

const kota = Readable.from(["Jakarta", "Bandung", "Surabaya"]);

let nomor = 0;
for await (const nama of kota) {
  nomor++;
  console.log(`${nomor}. ${nama}`);
}
console.log("total:", nomor);

Output yang Dihasilkan:

text 
1. Jakarta
2. Bandung
3. Surabaya
total: 3

Perulangan pada baris 6 menunggu potongan berikutnya secara otomatis, mengeksekusi body untuk setiap potongan, lalu selesai dengan sendirinya ketika stream berakhir — tidak perlu listener end terpisah. Karena body bisa memakai await, pemrosesan asynchronous per potongan menjadi lugas tanpa callback bersarang. Pola lengkapnya dijelaskan pada Consuming readable streams with async iterators .

Menahan Aliran dengan pause dan resume #

Pengendalian aliran menjadi penting ketika pemrosesan satu potongan butuh waktu. Method pause() menghentikan aliran data, dan resume() menyalakannya kembali:

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

const angka = Readable.from([1, 2, 3, 4, 5]);
let jumlah = 0;

angka.on("data", (nilai) => {
  jumlah += nilai;
  console.log("terima:", nilai);
  angka.pause();
  setImmediate(() => angka.resume());
});

angka.on("end", () => {
  console.log("end, jumlah:", jumlah);
});

Output yang Dihasilkan:

text 
terima: 1
terima: 2
terima: 3
terima: 4
terima: 5
end, jumlah: 15

Di dalam listener data, setiap nilai dicatat lalu stream langsung dijeda — mensimulasikan pemrosesan yang belum selesai. Pemanggilan resume() ditunda lewat setImmediate agar berjalan pada giliran event loop berikutnya. Urutan terima: 1 sampai terima: 5 tetap terjaga karena setiap nilai hanya mengalir setelah resume() dipanggil. Pola jeda-putar-putar ini adalah inti dari mekanisme backpressure yang akan kita bedah pada chapter akhir Section ini.

Membaca Berkas Secara Bertahap #

fs.createReadStream() membungkus berkas menjadi Readable stream; seluruh opsinya — termasuk start dan end untuk membaca sebagian berkas — dirinci pada dokumentasi fs.createReadStream() . Demo berikut memakai berkas daftar-kota.txt yang berada di folder yang sama dengan contoh kode, dibatasi highWaterMark: 12 agar potongan kecil dan banyak:

js 
import { createReadStream } from "node:fs";

const berkas = createReadStream("daftar-kota.txt", {
  encoding: "utf8",
  highWaterMark: 12,
});

let nomor = 0;
for await (const potongan of berkas) {
  nomor++;
  console.log(`potongan ${nomor}:`, JSON.stringify(potongan));
}

Output yang Dihasilkan:

text 
potongan 1: "Jakarta\nBand"
potongan 2: "ung\nSurabaya"
potongan 3: "\nMedan\nDenpa"
potongan 4: "sar\n"

Option encoding: "utf8" membuat tiap potongan langsung berupa string, dan highWaterMark: 12 membatasi buffer internal sekitar 12 byte. Hasilnya, berkas 43 byte terbagi menjadi empat potongan. JSON.stringify() dipakai untuk menampilkan batas potongan secara jujur: perhatikan bahwa potongan memotong nama kota di tengah — Band dan ung terpisah antara potongan 1 dan 2.

Membaca per Baris dengan readline #

Kebutuhan paling umum justru per baris. Daripada menulis sendiri logika penyambung potongan, module readline menyediakan createInterface() yang menyulap stream byte menjadi stream baris — opsinya dirinci pada dokumentasi readline.createInterface() :

js 
import { createReadStream } from "node:fs";
import { createInterface } from "node:readline";

const barisStream = createInterface({
  input: createReadStream("daftar-kota.txt"),
  crlfDelay: Infinity,
});

for await (const baris of barisStream) {
  console.log("kota:", baris);
}

Output yang Dihasilkan:

text 
kota: Jakarta
kota: Bandung
kota: Surabaya
kota: Medan
kota: Denpasar

Option input menerima Readable stream apa pun — createInterface() menyambungkannya secara otomatis, menyatukan potongan demi potongan, lalu memancarkan satu baris setiap kali penanda baris baru ditemukan. Option crlfDelay: Infinity menstabilkan perlakuan terhadap kombinasi \r\n pada berkas dari sistem Windows. Hasilnya lima baris bersih, terlepas dari bagaimana byte terbagi di balik layar.

Kita kini menguasai sisi penghasil data beserta dua gaya konsumsinya. Belum lengkap tanpa sisi penerima: bagaimana menulis data secara bertahap, menandai selesainya penulisan, dan menangani buffer internal yang penuh — itu pokok Chapter Writable Stream.