Writable Streams

Writable stream คืออะไร?

Writable คือ stream ที่ข้อมูลถูกเขียนเข้าไปได้ ตัวอย่างในตัวของ Node ได้แก่ fs.createWriteStream (เขียนไฟล์), process.stdout (output ของ terminal) และ HTTP response object คุณสร้าง writable กำหนดเองได้ด้วยการ extend Writable และ implement method _write

write() และ end()

method หลักสองตัวบน Writable คือ:

writable.write(chunk[, encoding][, callback]) — เขียน chunk ของข้อมูล คืนค่า true ถ้า internal buffer อยู่ใต้ high-water mark, false ถ้าปลายทางตามไม่ทัน (สัญญาณ backpressure)
writable.end([chunk][, callback]) — ส่งสัญญาณว่าจะไม่มีการเขียนข้อมูลอีก อาจเขียน chunk สุดท้ายหนึ่งอันได้ด้วย จะยิง event finish เมื่อข้อมูลทั้งหมดถูก flush แล้ว

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

const dest = fs.createWriteStream('./output.txt');

dest.write('first line\n');
dest.write('second line\n');
dest.end('last line\n', function() {
  console.log('all data flushed');
});

event `finish`

หลังจาก end() ถูกเรียกและข้อมูลที่ค้างอยู่ทั้งหมดถูก flush ไปยังระบบข้างล่างแล้ว stream จะ emit finish นี่คือจังหวะที่ถูกต้องในการรู้ว่าการเขียนเสร็จสมบูรณ์จริง ๆ:

const fs = require('node:fs');
const dest = fs.createWriteStream('./output.txt');

dest.on('finish', function() {
  console.log('write complete');
});

dest.end('done');

event `drain` และการควบคุม flow

write() คืนค่า boolean เมื่อมันคืน false แปลว่า internal buffer เต็มแล้ว — คุณต้อง หยุดเขียน และรอ event drain ก่อนที่จะเขียนต่อ การเพิกเฉยต่อสิ่งนี้ทำให้ memory โตขึ้นโดยไม่มีขอบเขตเมื่อ chunk ที่ buffer ไว้สะสมมากขึ้นเรื่อย ๆ

function writeWithDrain(writable, chunks) {
  var i = 0;

  function next() {
    while (i < chunks.length) {
      var ok = writable.write(chunks[i]);
      i++;
      if (!ok) {
        // buffer full — wait for drain before continuing
        writable.once('drain', next);
        return;
      }
    }
    writable.end();
  }

  next();
}

นี่คือรูปแบบที่ต้องทำเองซึ่ง pipe และ pipeline จัดการให้อัตโนมัติ การเข้าใจมันช่วยอธิบายว่าทำไม API ระดับสูงเหล่านั้นถึงมีอยู่

การวัด highWaterMark, write() และ writableLength

writable.writableLength รายงานอย่างแม่นยำว่ามีกี่ byte (หรือ object ใน object mode) ที่ถูก buffer อยู่ตอนนี้และยังไม่ถูก flush ไปยังปลายทาง การดูมันคู่กับ return value ของ write() ทำให้ high-water mark เป็นรูปธรรมแทนที่จะเป็นแค่แนวคิดนามธรรม:

Node.js

import { Writable } from 'node:stream';

var sink = new Writable({
  highWaterMark: 10, // bytes -- tiny on purpose so backpressure triggers fast
  write(chunk, encoding, callback) {
    setImmediate(callback); // simulate an async destination
  }
});

console.log('highWaterMark:', sink.writableHighWaterMark);

var okFirst = sink.write('12345'); // 5 bytes, still under 10
console.log('write() #1 returned:', okFirst, '| writableLength:', sink.writableLength);

var okSecond = sink.write('1234567890'); // pushes past the high-water mark
console.log('write() #2 returned:', okSecond, '| writableLength:', sink.writableLength);

sink.once('drain', function() {
  console.log('drain fired -- writableLength now:', sink.writableLength);
  sink.end();
});

Output (verified on Node 24)

highWaterMark: 10
write() #1 returned: true | writableLength: 5
write() #2 returned: false | writableLength: 15
drain fired -- writableLength now: 0

สังเกตว่า writableLength ขึ้นไปถึง 15 — เกิน high-water mark ที่ 10 high-water mark คือ threshold ที่พลิก return value ของ write() เป็น false ไม่ใช่ hard cap: Node ยังรับและ buffer chunk ที่คุณเพิ่งเขียนไป และเป็นความรับผิดชอบของคุณที่จะหยุดเรียก write() จนกว่า drain จะยิง

Runnable demo

demo สร้าง custom Writable ที่ log แต่ละ chunk ไปที่ console จำลอง sink ที่ช้า จากนั้นเขียนสาม chunk และปิด stream

Node.js

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

const sink = new Writable({
  write(chunk, encoding, callback) {
    console.log('sink received: ' + chunk.toString());
    callback(); // signal ready for next chunk
  }
});

sink.on('finish', function() {
  console.log('stream finished');
});

sink.write('alpha');
sink.write('beta');
sink.end('gamma');

Output (verified on Node 24)

sink received: alpha
sink received: beta
sink received: gamma
stream finished

Mental Model

flowchart TB
  Write["writable.write(chunk)"] -->|"below high-water mark"| True["returns true -- keep writing"]
  Write -->|"at or above high-water mark"| False["returns false -- buffer is full"]
  False -->|"you must"| Pause["stop writing"]
  Pause -->|"wait for"| Drain["'drain' event"]
  Drain -->|"safe to"| Write
  End["writable.end()"] -->|"once all buffered data is flushed"| Finish["'finish' event"]

return value แบบ boolean ของ write() คือสัญญา backpressure ทั้งหมด: false แปลว่าหยุดและรอ drain ก่อนเขียนอีกครั้ง

Trade-offs

ทางเลือก	ข้อดี	ต้นทุน
จัดการ `write()`/`drain` เอง	ควบคุมได้เต็มที่ว่าจะ buffer เมื่อไหร่และเท่าไหร่	ยาวและพลาดง่าย — ลืมตรวจแค่จุดเดียว memory ก็โตไม่มีขอบเขต
`.pipe()` / `pipeline()`	จัดการขั้นตอน write()/drain ให้อัตโนมัติ	ควบคุมโดยตรงน้อยลง; คุณใช้ contract ของ error/completion ของ API ระดับสูงแทนที่จะสร้างเอง
ตรวจ `writableLength` โดยตรง	ให้จำนวน byte (หรือ object) ที่ buffer อยู่แบบ real-time ที่แม่นยำสำหรับ monitoring/logging	ต้องอ่านให้ถูกอีกเรื่องหนึ่ง — มันไม่มีความหมายถ้าไม่รู้ `writableHighWaterMark` ปัจจุบันด้วย

Common Misconceptions

“high-water mark เป็น hard limit ว่า writable buffer ได้เท่าไหร่” — ความจริง: มันเป็น soft threshold ที่พลิก return value ของ write() เป็น false; Node ยังรับและ buffer chunk ที่ทำให้เกินมันอยู่ดี
“write() ที่คืน false แปลว่าการเขียนล้มเหลว” — ความจริง: chunk ยังถูกรับไว้และจะถูก flush; false เป็นแค่สัญญาณให้ช้าลง ไม่ใช่ error
“finish กับ close หมายถึงสิ่งเดียวกัน” — ความจริง: finish ยิงเมื่อข้อมูลที่ buffer ไว้ทั้งหมดถูก flush แล้วหลัง end(); close ยิงเมื่อ resource ข้างล่าง (file descriptor, socket) ถูกปล่อยจริง ๆ ซึ่งอาจเกิดคนละเวลากัน

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

code เรียก writable.write() ใน loop โดยไม่เคยตรวจ return value แบบ boolean หรือ listen drain เลยหรือเปล่า?
implementation ของ write(chunk, encoding, callback) ใน custom Writable ลืมเรียก callback() ในบาง code path (รวมถึง error path) ทำให้ stream ค้างตลอดกาลรอ callback ที่ไม่มีวันมาหรือเปล่า?
code สมมติว่า end() เสร็จแปลว่าไฟล์/socket ปลายทางถูกปิดสมบูรณ์แล้ว ทั้งที่ควรรอ finish (ข้อมูล flush แล้ว) หรือ close (resource ถูกปล่อยแล้ว) ขึ้นอยู่กับว่าจริง ๆ ต้องการรู้อะไรหรือเปล่า?

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

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

var dest = fs.createWriteStream('./output.log');

function writeAll(lines) {
for (var i = 0; i < lines.length; i++) {
  dest.write(lines[i] + '\n'); // return value never checked
}
dest.end();
}

writeAll(new Array(1000000).fill('a log line'));

เฉลย

writeAll เรียก dest.write() ใน loop แน่นโดยไม่เคยตรวจ return value หรือหยุดเมื่อมันคืน false ถ้าปลายทาง (ดิสก์, network socket) ตามการเรียก write() แบบ synchronous เป็นล้านครั้งไม่ทัน Node ก็ยังคง buffer ทุก chunk ใน memory อยู่ดี และ process อาจ OOM ก่อนที่จะถึง end() ด้วยซ้ำ วิธีแก้คือรูปแบบที่ทำเองจากก่อนหน้านี้ในบทนี้ — ตรวจ return value ของ write() แล้วหยุดผ่าน writable.once('drain', next) — หรือง่ายกว่านั้นคือแทนที่ loop ทั้งหมดด้วย Readable ที่ห่อ lines แล้ว pipe เข้า dest