Dasar stream

Stream adalah antarmuka Node untuk memproses data secara bertahap alih-alih sekaligus. Sebuah stream membaca atau menulis potongan demi potongan, sehingga berkas berukuran besar atau data dari jaringan dapat diolah tanpa dimuat seluruhnya ke memori. Node menyediakan empat jenis stream: Readable yang menghasilkan data, Writable yang menyerap data, Duplex yang keduanya, dan Transform yang mengubah data yang lewat. Setelah membaca topik ini, pembaca dapat membuat stream dasar, memahami peristiwa yang dipancarkan, serta memahami mekanisme pipe dan backpressure yang menjaga keseimbangan antara sumber dan tujuan.

Readable dan peristiwanya#

Stream Readable memancarkan peristiwa saat data tersedia dan saat data habis. Peristiwa data dipancarkan setiap kali potongan data keluar dari stream, sedangkan peristiwa end dipancarkan ketika seluruh data sudah dibaca. Peristiwa close dipancarkan ketika sumber daya di balik stream ditutup. Konstruktor Readable menerima opsi read berupa fungsi yang dipanggil ketika stream meminta data. Di dalam fungsi tersebut, this.push(chunk) mengirimkan potongan data ke stream, sedangkan this.push(null) menandai bahwa seluruh data sudah terkirim dan akhir stream telah tercapai.

const { Readable } = require('node:stream');

const r = new Readable({
  read() {
    this.push('A');
    this.push('B');
    this.push(null);
  }
});

const potongan = [];
r.on('data', (chunk) => potongan.push(chunk.toString()));
r.on('end', () => console.log('selesai, potongan:', potongan.join(',')));
r.on('close', () => console.log('stream ditutup'));

selesai, potongan: A,B
stream ditutup

Stream berjalan dalam mode flowing begitu penerima peristiwa data dipasang. Sebelum penerima dipasang, stream berada dalam mode paused dan data menumpuk di buffer internal hingga dibaca secara eksplisit dengan read(). Untuk mengubah mode secara manual, metode pause() dan resume() tersedia, tetapi mayoritas pemakaian cukup memasang penerima data dan membiarkan stream mengalir.

Writable dan operasi tulis#

Stream Writable menerima data melalui metode write(chunk) dan menandai selesai melalui end(). Konstruktor menerima opsi write berupa fungsi yang dipanggil untuk setiap potongan, dengan tiga argumen: chunk, encoding, dan callback. Argumen callback wajib dipanggil setelah potongan selesai diproses agar stream tahu kapan siap menerima potongan berikutnya. Pemanggilan callback dengan argumen Error menyebabkan stream memancarkan peristiwa error.

const { Writable } = require('node:stream');

const tujuan = new Writable({
  write(chunk, encoding, callback) {
    console.log('tulis:', chunk.toString().trim());
    callback();
  }
});

tujuan.write('satu\n');
tujuan.write('dua\n');
tujuan.end('akhir\n');

tulis: satu
tulis: dua
tulis: akhir

Metode end() menandai bahwa tidak ada data lagi yang akan ditulis. Argumen opsional pada end() ditulis sebagai potongan terakhir sebelum stream ditutup. Peristiwa finish dipancarkan ketika seluruh data telah ditulis ke sumber daya di bawahnya, sedangkan peristiwa close dipancarkan ketika sumber daya ditutup. Pemakaian write() tanpa memeriksa nilai kembalian sering membengkakkan memori ketika sumber menghasilkan data lebih cepat dari yang dapat ditulis tujuan.

pipe dan backpressure#

Metode pipe() menghubungkan stream Readable ke stream Writable, mengalirkan data secara otomatis dan menghentikan sumber ketika tujuan belum siap. Penghentian sementara ini disebut backpressure, dan mekanismenya bekerja lewat nilai kembalian writable.write(chunk) yang berubah menjadi false ketika buffer internal sudah penuh. Stream Readable yang dipasang dengan pipe() akan menjeda pemancaran data sampai tujuan memberi sinyal siap kembali. Pola ini mencegah memori membengkak ketika kecepatan sumber dan tujuan berbeda.

const { Readable, Writable } = require('node:stream');

const sumber = Readable.from(['satu\n', 'dua\n', 'tiga\n']);
const tujuan = new Writable({
  write(chunk, encoding, callback) {
    console.log('terima:', JSON.stringify(chunk.toString()));
    callback();
  }
});

sumber.pipe(tujuan).on('finish', () => console.log('selesai'));

terima: "satu\n"
terima: "dua\n"
terima: "tiga\n"
selesai

Fungsi Readable.from() menerima iterable atau fungsi generator dan menghasilkan stream Readable yang memancarkan tiap nilai. Kombinasi dengan pipe() menghasilkan alur singkat dari sumber ke tujuan tanpa penerima data manual. Peristiwa finish dipancarkan pada stream Writable ketika seluruh data dari pipe() sudah ditulis, sehingga penerima pada pipe(tujuan) memberi sinyal penyelesaian yang andal.

Diagram berikut memvisualkan aliran data dari Readable ke Writable melalui pipe, serta arah balik backpressure yang menjeda sumber ketika buffer tujuan penuh.

Object mode dan encoding#

Stream Readable dan Writable secara default beroperasi pada potongan berbentuk Buffer atau Uint8Array. Sebuah stream dapat dibuat dengan opsi objectMode: true agar potongan dapat berupa nilai JavaScript apa pun, termasuk objek, angka, atau string yang sudah terurai. Stream berbasis objek berguna untuk pemrosesan terstruktur seperti membaca rekaman JSON baris-per-baris atau mengubah objek menjadi baris CSV. Stream Readable.from(iterable) otomatis berjalan dalam object mode ketika iterablenya memproduksi nilai non-Buffer.

Opsi encoding pada stream Readable membuat potongan yang diterima berbentuk string alih-alih Buffer. Ketika encoding diatur ke 'utf8', setiap potongan Buffer dikonversi menjadi string sebelum dipancarkan pada peristiwa data. Pengaturan ini berguna ketika sumber data sudah pasti teks dan kode yang memproses ingin menghindari konversi eksplisit di setiap penerima. Pada stream Writable, opsi decodeStrings mengontrol apakah string yang ditulis melalui write() dikonversi menjadi Buffer sebelum diteruskan ke fungsi write internal.

Kesalahan umum#

Memasang penerima data pada stream yang besar tanpa pipe atau pipeline dapat membanjiri memori karena stream Readable tidak memberi backpressure ke hulu secara otomatis pada mode flowing. Pengganti yang aman adalah stream.pipeline(sumber, tujuan, callback) yang menangani backpressure sekaligus merambatkan error dan menutup kedua stream. Pemakaian readable.pipe(writable) tanpa penerima error pada masing-masing stream juga berbahaya karena error pada satu stream tidak otomatis menutup yang lain.

Pemanggilan writable.write() berturut-turut tanpa memeriksa nilai kembaliannya adalah kesalahan lain. Ketika write() mengembalikan false, pemanggil harus menunggu peristiwa drain sebelum menulis lagi. Pemakaian pipe() sudah menangani detail ini secara otomatis, sehingga lebih disarankan daripada penulisan manual dengan write() pada alur yang berkelanjutan.

Ringkasan#

• Readable memancarkan data, end, dan close, sedangkan Writable memancarkan finish dan close. Konstruktor menerima opsi read atau write yang dipanggil sesuai permintaan stream.
• Mode flowing dimulai ketika penerima data dipasang, dan mode paused memakai read() secara eksplisit.
• pipe() menghubungkan Readable ke Writable sambil menjaga keseimbangan kecepatan lewat backpressure yang menjeda sumber ketika tujuan belum siap.
• stream.pipeline lebih disarankan pada produksi karena merambatkan error dan menutup kedua stream secara otomatis.

Referensi#

• Stream — Node.js — Referensi resmi modul stream dengan rincian Readable, Writable, Duplex, dan Transform.
• Backpressure — Node.js — Penjelasan resmi tentang mekanisme backpressure dan nilai kembalian writable.write.