Readable Streams

Readable stream คืออะไร?

Readable คือ stream ที่ข้อมูลถูกบริโภคได้ ตัวอย่างในตัวของ Node ได้แก่ fs.createReadStream (อ่านไฟล์), process.stdin (input จาก terminal) และ HTTP request body คุณสร้าง readable source ของตัวเองได้ด้วย Readable.from หรือ extend Readable

Readable stream ทำงานในสองโหมด:

Flowing mode — event data จะยิงต่อเนื่องขณะ chunk มาถึง คุณผูก listener data เพื่อเข้าสู่โหมดนี้
Paused mode — chunk จะสะสมไว้จนกว่าคุณจะเรียก .read() เอง นี่คือดีฟอลต์จนกว่าคุณจะผูก listener หรือ pipe

วิธีที่สะอาดที่สุดในปัจจุบันคือ async iteration ด้วย for await...of ซึ่งจัดการ paused mode และ backpressure ให้อัตโนมัติ

event `data`, `end` และ `error`

event-based API แบบดั้งเดิมมีมาก่อน async iteration มันยังคงสำคัญที่ต้องเข้าใจเพราะ Node API จำนวนมากยัง emit event เหล่านี้อยู่:

const fs = require('node:fs');

const stream = fs.createReadStream('./file.txt', { encoding: 'utf8' });

stream.on('data', function(chunk) {
  console.log('received chunk of length', chunk.length);
});

stream.on('end', function() {
  console.log('no more data');
});

stream.on('error', function(err) {
  console.error('stream error', err.message);
});

data — ยิงทุกครั้งที่มี chunk พร้อมใช้ chunk เป็น Buffer โดยดีฟอลต์ หรือ string ถ้าตั้ง encoding ไว้
end — ยิงครั้งเดียวเมื่อไม่มีข้อมูลให้บริโภคแล้ว
error — ยิงถ้ามีอะไรผิดพลาด (ไฟล์หาย, permission ถูกปฏิเสธ ฯลฯ)

Readable.from

Readable.from(iterable) แปลง sync หรือ async iterable ใด ๆ ให้เป็น Readable stream มีประโยชน์สำหรับ testing และห่อ in-memory data source:

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

const stream = Readable.from(['chunk1', 'chunk2', 'chunk3']);

iterable นั้นเป็น array ธรรมดา, generator function หรือ async generator ก็ได้ — ให้วิธีที่สะอาดในการสร้างข้อมูลแบบ lazy

Async iteration ด้วย `for await`

ตั้งแต่ Node 10+ Readable stream ทุกตัว implement async iterator protocol นั่นหมายความว่าคุณบริโภคมันด้วย for await...of ภายใน async function ได้:

async function processStream(stream) {
  for await (const chunk of stream) {
    // handle each chunk
  }
}

วิธีนี้อ่านง่ายกว่า event listener จัดการ backpressure ให้อัตโนมัติ และแสดง error เป็น thrown exception ที่ try/catch ของคุณจับได้

Runnable demo

demo ด้านล่างสร้าง Readable จาก async generator แล้วบริโภคมันด้วย for await แต่ละ yield ของ generator แทนหนึ่ง chunk ที่มาถึงแบบ asynchronous

Node.js

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

async function* generate() {
  const words = ['Node', 'streams', 'are', 'powerful'];
  for (const word of words) {
    yield word;
  }
}

async function main() {
  const stream = Readable.from(generate());
  const parts = [];

  for await (const chunk of stream) {
    parts.push(chunk);
    console.log('chunk: ' + chunk);
  }

  console.log('complete: ' + parts.join(' '));
}

main();

Output (verified on Node 24)

chunk: Node
chunk: streams
chunk: are
chunk: powerful
complete: Node streams are powerful

Object mode

โดยดีฟอลต์ Readable ทำงานใน byte mode: chunk เป็น Buffer (หรือ string ถ้าตั้ง encoding) และ high-water mark นับเป็น byte (ดีฟอลต์ 64 KiB) การส่ง { objectMode: true } จะสลับไปเป็น object mode ที่ chunk เป็นค่า JavaScript อะไรก็ได้ — object, ตัวเลข, array — และ high-water mark นับเป็น object แทน Readable.from(iterable) ดีฟอลต์เป็น object mode ไม่ว่า iterable จะ yield อะไรก็ตาม:

Node.js

import { Readable } from 'node:stream';

// Byte mode (the default for a plain Readable): highWaterMark counts bytes
var byteStream = new Readable({ read() {} });
console.log('byte mode highWaterMark:', byteStream.readableHighWaterMark);

// Readable.from() defaults objectMode to true regardless of what it yields --
// even for Buffers -- unless you pass { objectMode: false } yourself.
var fromDefault = Readable.from([Buffer.from('a'), Buffer.from('b')]);
console.log('Readable.from() default objectMode:', fromDefault.readableObjectMode);
console.log('Readable.from() default highWaterMark:', fromDefault.readableHighWaterMark);

// Explicit object mode with plain objects as chunks
var objStream = Readable.from([{ id: 1 }, { id: 2 }], { objectMode: true });
console.log('explicit object mode highWaterMark:', objStream.readableHighWaterMark);

var received = [];
for await (const chunk of objStream) {
  received.push(chunk);
}
console.log('objects received:', JSON.stringify(received));

Output (verified on Node 24)

byte mode highWaterMark: 65536
Readable.from() default objectMode: true
Readable.from() default highWaterMark: 1
explicit object mode highWaterMark: 1
objects received: [{"id":1},{"id":2}]

Interop กับ web streams (WHATWG)

Browser และ fetch ใช้ stream API ที่ต่างออกไป ซึ่งเป็นมาตรฐาน (ReadableStream จาก WHATWG Streams spec) Readable.toWeb() และ Readable.fromWeb() ของ Node แปลงไปมาระหว่างสองแบบนี้ ซึ่งสำคัญเมื่อไหร่ก็ตามที่ code ของ Node ส่งข้อมูลให้ (หรือรับจาก) fetch, Response หรือ web-standard API ใด ๆ:

Node.js

import { Readable } from 'node:stream';

// Node Readable -> web ReadableStream -> Response body
var nodeReadable = Readable.from([Buffer.from('alpha'), Buffer.from('beta'), Buffer.from('gamma')]);
var webReadable = Readable.toWeb(nodeReadable);
console.log('is a WHATWG ReadableStream:', webReadable instanceof ReadableStream);

var response = new Response(webReadable);
var text = await response.text();
console.log('Response.text():', text);

// The reverse: a web ReadableStream (like fetch's Response.body) -> Node Readable
var webBody = new Response(JSON.stringify({ ok: true })).body;
var back = Readable.fromWeb(webBody);
var collected = '';
for await (const chunk of back) collected += chunk;
console.log('fromWeb collected:', collected);

Output (verified on Node 24)

is a WHATWG ReadableStream: true
Response.text(): alphabetagamma
fromWeb collected: {"ok":true}

Mental Model

flowchart TB
  Source["Readable source"] -->|"attach 'data' listener"| Flowing["flowing mode -- pushed at you continuously"]
  Source -->|"no listener / no pipe yet"| Paused["paused mode -- call .read() yourself"]
  Source -->|"for await...of"| Iter["async iteration -- paused mode + backpressure, handled for you"]
  Flowing -->|"consumer too slow, no pause()"| Leak["buffered chunks pile up in the consumer"]

Readable ถูกดันข้อมูลใส่คุณ (flowing/data) หรือคุณดึงมันเอง (paused/read หรือ for await...of) — for await คือ paused mode ที่จัดการ backpressure ให้อัตโนมัติ

Trade-offs

ทางเลือก	ข้อดี	ต้นทุน
event `data` (flowing mode)	style callback ที่คุ้นเคย ใช้ได้ทั้ง code เก่าและใหม่	ไม่มี backpressure อัตโนมัติ — handler ที่ช้าและไม่เรียก `.pause()` จะปล่อยให้ chunk สะสมในสิ่งที่มันเขียนไป
`for await...of` (async iteration)	จัดการ backpressure ให้อัตโนมัติ; อ่านจากบนลงล่างเหมือน code synchronous; error แสดงเป็น exception ปกติ	ต้องการ `async` function; overhead ต่อ chunk มากกว่า `data` listener ธรรมดาเล็กน้อย
Byte mode	ตรงกับสิ่งที่ต้นทางส่วนใหญ่ (ไฟล์, socket) สร้างขึ้นจริง; high-water mark เป็นไบต์มีความหมาย	ขอบเขต chunk เป็นไปตามอำเภอใจ — chunk อาจตัดตัวอักษร multi-byte หรือ record เชิงตรรกะขาด
Object mode	chunk เป็นค่าที่มีโครงสร้างอะไรก็ได้ (แถวข้อมูล, JSON ที่ parse แล้ว, custom object)	high-water mark นับ object ไม่ใช่ byte — stream ของ object ขนาดใหญ่ยังทำให้ memory บวมได้แม้อยู่ใต้ limit ดีฟอลต์

Common Misconceptions

“การผูก data listener คือวิธีบริโภค stream ที่ทันสมัยและแนะนำ” — ความจริง: for await...of เป็นที่นิยมกว่าในปัจจุบัน — มันจัดการ backpressure ให้อัตโนมัติ ซึ่ง data listener ธรรมดาไม่ทำ
“Readable ที่สร้างด้วย Readable.from() เป็น byte mode เว้นแต่คุณจะระบุอย่างอื่น” — ความจริง: ตรงกันข้าม — Readable.from() ดีฟอลต์เป็น object mode แม้ว่า iterable จะ yield Buffer หรือ string ก็ตาม เว้นแต่คุณจะส่ง { objectMode: false } อย่างชัดเจน
“web stream (ReadableStream) กับ Node stream (Readable) คือ API เดียวกัน” — ความจริง: มันเป็นสอง API ที่แตกต่างกัน เข้ากันได้แค่บางส่วน Readable.toWeb()/Readable.fromWeb() มีอยู่ก็เพราะคุณส่งตัวหนึ่งตรง ๆ ไปในที่ที่ต้องการอีกตัวไม่ได้

อ่านโค้ดที่ AI เขียน

data listener ทำอะไรที่อาจช้า (เขียนไปยัง socket, database, stream อื่น) โดยไม่ตรวจ return value หรือใช้ backpressure เองหรือเปล่า? flowing mode ไม่เคยหยุดส่ง chunk ด้วยตัวเอง
code ส่ง Node Readable ตรง ๆ ไปยัง API ที่เกี่ยวกับ fetch หรือฝั่ง browser ที่คาดหวัง WHATWG ReadableStream (หรือกลับกัน) โดยไม่ใช้ Readable.toWeb()/fromWeb() หรือเปล่า?
snippet สมมติว่า Readable.from(iterable) สร้าง chunk แบบ byte mode (เรียก .toString() บนแต่ละตัวโดยคิดว่าเป็น Buffer) ทั้งที่ iterable ให้ plain object จริง ๆ หรือไม่เคยตรวจโหมดเลยหรือเปล่า?

🐛 หา bug ในโค้ดนี้

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

var dest = fs.createWriteStream('./out.txt');
var source = Readable.from(['a very large amount of data, chunk after chunk after chunk']);

source.on('data', function(chunk) {
dest.write(chunk); // return value ignored -- source is never paused
});

เฉลย

data listener เรียก dest.write(chunk) แต่ไม่เคยตรวจ boolean ที่มัน return และไม่เคยเรียก source.pause() เมื่อมันเป็น false flowing mode จะส่ง chunk จาก source ต่อไปด้วยความเร็วเต็มที่ไม่ว่า dest จะตามทันหรือไม่ ดังนั้นถ้า dest ช้ากว่า source (network socket, disk ที่ช้า, API ที่ถูกจำกัด rate) chunk จะสะสมใน buffer ภายในของ dest โดยไม่มีขอบเขต วิธีแก้คือตรวจ return value ของ dest.write(chunk) แล้ว pause()/resume() ตามนั้น (ตามที่แสดงในบท Writable และ Backpressure) หรือ — ง่ายกว่ามาก — แทนที่คู่ data/write ทั้งหมดด้วย source.pipe(dest) ซึ่งทำสิ่งนั้นให้อัตโนมัติ

Readable Streams

Readable stream คืออะไร?

event data, end และ error

Readable.from

Async iteration ด้วย for await

Runnable demo

Object mode

Interop กับ web streams (WHATWG)

Mental Model

Trade-offs

Common Misconceptions

อ่านโค้ดที่ AI เขียน

event `data`, `end` และ `error`

Async iteration ด้วย `for await`