Pipe & Pipeline

readable.pipe(writable)

.pipe() เชื่อม Readable เข้ากับ Writable โดยตรง จัดการ flow ของข้อมูลระหว่างกันให้อัตโนมัติ มันจัดการ backpressure ด้วย: เมื่อ writable ส่งสัญญาณว่า buffer เต็ม (โดยคืน false จาก write()) pipe จะ pause readable แล้ว resume มันตอน drain

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

fs.writeFileSync('./input.txt', 'hello world'); // demo fixture

const readable = fs.createReadStream('./input.txt');
const writable = fs.createWriteStream('./output.txt');

readable.pipe(writable);

คุณเชื่อม .pipe() ต่อกันผ่าน Transform stream เพื่อสร้าง processing pipeline ได้:

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

fs.writeFileSync('./file.txt', 'hello world'); // demo fixture

fs.createReadStream('./file.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('./file.txt.gz'));

ปัญหาของ .pipe()

.pipe() มีข้อบกพร่องที่รู้กันดี: มันไม่ส่งต่อ error ถ้า stream ตัวใดใน chain emit event error stream ตัวอื่นจะไม่ถูก destroy หรือปิดอัตโนมัติ คุณจะเจอ file handle ที่เปิดค้างและ memory leak

// @skip-verify illustrative fragment: readable/writable continue the example above
// Dangerous: error on readable does NOT close writable
readable.pipe(writable);

readable.on('error', function(err) {
  console.error('read error — but writable is still open!', err.message);
  writable.destroy();   // you must clean up manually
});

stream.pipeline() — ทางเลือกที่ปลอดภัย

stream.pipeline(...streams, callback) ถูกเพิ่มเข้ามาใน Node 10 เพื่อแก้ปัญหานี้โดยเฉพาะ มัน:

เชื่อม stream ทั้งหมดเข้าด้วยกันตั้งแต่ต้นจนจบ
ส่งต่อ error จาก stream ใดก็ตามไปยัง callback
destroy stream ทั้งหมดใน chain อัตโนมัติเมื่อเกิด error

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

pipeline(
  fs.createReadStream('./input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('./input.txt.gz'),
  function(err) {
    if (err) {
      console.error('pipeline failed:', err.message);
    } else {
      console.log('pipeline complete');
    }
  }
);

Node 15+ ก็มี stream/promises ให้ด้วย ซึ่งให้เวอร์ชันที่เป็นมิตรกับ await — ไม่มี callback และเป็น rejected promise ธรรมดาเมื่อเกิด error:

Node.js

import { pipeline } from 'node:stream/promises';
import { Readable, Writable } from 'node:stream';
import { createGzip, createGunzip } from 'node:zlib';

var source = Readable.from(['hello world, compressed and decompressed via a stream/promises pipeline']);
var chunks = [];
var sink = new Writable({
  write(chunk, encoding, callback) { chunks.push(chunk); callback(); }
});

await pipeline(source, createGzip(), createGunzip(), sink);
console.log('round-tripped:', Buffer.concat(chunks).toString());
console.log('pipeline() resolved -- no callback needed with stream/promises');

Output (verified on Node 24)

round-tripped: hello world, compressed and decompressed via a stream/promises pipeline
pipeline() resolved -- no callback needed with stream/promises

ยกเลิก pipeline ด้วย AbortSignal

pipeline() (ทั้งแบบ callback และ promise) รับ options.signal ได้ การ abort มัน — เพราะ client หลุดการเชื่อมต่อ, request timeout หรือผู้ใช้ยกเลิก — จะ destroy ทุก stream ใน chain ทันทีแทนที่จะปล่อยให้ source ที่ช้าทำงานต่อไปโดยไม่มีประโยชน์:

Node.js

import { pipeline } from 'node:stream/promises';
import { Readable, Writable } from 'node:stream';
import { setTimeout as delay } from 'node:timers/promises';

async function* slowSource() {
  for (let i = 0; i < 20; i++) {
    await delay(10); // simulate a slow upstream (network, DB cursor, ...)
    yield String(i);
  }
}

var controller = new AbortController();
var sink = new Writable({
  write(chunk, enc, cb) { cb(); }
});

// Simulates a client disconnect or a request timeout arriving mid-stream
setTimeout(function() { controller.abort(); }, 30);

try {
  await pipeline(Readable.from(slowSource()), sink, { signal: controller.signal });
  console.log('pipeline finished normally');
} catch (err) {
  console.log('pipeline aborted:', err.name);
}

Output (verified on Node 24)

pipeline aborted: AbortError

Transform stream: การแปลงรูปร่างข้อมูลระหว่างทาง

ตัวอย่าง uppercase ด้านบนคือ Transform ที่ง่ายที่สุดเท่าที่จะเป็นไปได้: chunk เข้าหนึ่งอัน chunk ออกหนึ่งอัน Transform เปลี่ยน รูปร่าง ของข้อมูลได้ง่ายพอ ๆ กัน — เช่น เปลี่ยน stream ของ object ให้เป็น newline-delimited JSON (format ที่นิยมสำหรับ log และ streaming API) โดยรันเป็น object mode ที่ฝั่ง input และ byte mode ที่ฝั่ง output:

Node.js

import { Transform, Readable } from 'node:stream';

var toJsonLines = new Transform({
  objectMode: true, // accepts JS objects as input chunks
  transform(obj, encoding, callback) {
    callback(null, JSON.stringify(obj) + '\n'); // emits strings/bytes as output
  }
});

var source = Readable.from([{ id: 1 }, { id: 2 }, { id: 3 }], { objectMode: true });

var out = '';
source.pipe(toJsonLines);
for await (const chunk of toJsonLines) {
  out += chunk;
}
console.log(out.trim());

Output (verified on Node 24)

{"id":1}
{"id":2}
{"id":3}

.destroy() และ finished() — การเคลียร์ทุกเส้นทาง

ช่องโหว่การส่งต่อ error ของ .pipe() (ด้านบน) หมายความว่า คุณ ต้องรับผิดชอบ destroy ทุก stream ใน chain เมื่อตัวใดตัวหนึ่งเกิด error stream.finished(stream, callback) ให้ callback หนึ่งตัวที่ยิงแค่ครั้งเดียว ไม่ว่า stream จะจบแบบสำเร็จ, error หรือถูก destroy — วิธีที่เชื่อถือได้ในการรู้ว่า “stream นี้จบแล้ว ไม่ทางใดก็ทางหนึ่ง”:

Node.js

import { Readable, Writable, finished } from 'node:stream';

var source = Readable.from((function* () {
  yield 'a'; yield 'b'; throw new Error('source broke');
})());

var sink = new Writable({ write(chunk, enc, cb) { cb(); } });

finished(sink, function(err) {
  console.log('sink finished callback, err:', err ? err.message : null);
});

source.on('error', function(err) {
  console.log('source error:', err.message);
  sink.destroy(err); // propagate manually -- .pipe() alone would not do this
});

source.pipe(sink);

Output (verified on Node 24)

source error: source broke
sink finished callback, err: source broke

Duplex และ compose()

Duplex stream มีฝั่ง readable และ writable ที่เป็นอิสระต่อกัน — TCP socket คือตัวอย่างคลาสสิก: สิ่งที่คุณเขียนกับสิ่งที่คุณอ่านเป็นข้อมูลที่ไม่เกี่ยวข้องกัน stream.compose(...streams) (Node 16.9+) ต่อยอดแนวคิดนี้ไปอีกขั้น: มันเชื่อม stream แบบ Transform หลายตัวเข้าเป็น Duplex เดียว ทำให้ pipeline หลายขั้นตอนถูกส่งต่อและ pipe เข้าได้เหมือน stream ตัวหนึ่ง:

Node.js

import { Transform, Readable, compose } from 'node:stream';

var upper = new Transform({ transform(chunk, enc, cb) { cb(null, chunk.toString().toUpperCase()); } });
var addBang = new Transform({ transform(chunk, enc, cb) { cb(null, chunk.toString() + '!'); } });

// compose() fuses two Transforms into one Duplex -- one unit you can pipe into
var combined = compose(upper, addBang);
var source = Readable.from(['node', 'streams']);

var out = '';
source.pipe(combined);
for await (const chunk of combined) {
  out += chunk;
}
console.log('composed result:', out);

Output (verified on Node 24)

composed result: NODE!STREAMS!

Runnable demo

demo ด้านล่างสร้าง pipeline สั้น ๆ ใน memory: Readable source สร้าง string สามตัว, Transform ทำให้แต่ละ chunk เป็นตัวพิมพ์ใหญ่ และ Writable เก็บผลลัพธ์

Node.js

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

const source = Readable.from(['hello', ' world', ' streams']);

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  }
});

const collected = [];
const sink = new Writable({
  write(chunk, encoding, callback) {
    collected.push(chunk.toString());
    callback();
  }
});

pipeline(source, upper, sink, function(err) {
  if (err) {
    console.error('pipeline error:', err.message);
  } else {
    console.log('result: ' + collected.join(''));
  }
});

Output (verified on Node 24)

result: HELLO WORLD STREAMS

Mental Model

flowchart TB
  Pipe[".pipe(writable)"] -->|"error on any stream"| Gap["NOT propagated -- you must listen + destroy() manually"]
  Pipeline["pipeline(...streams, cb)"] -->|"error on any stream"| Auto["all streams destroyed automatically, cb(err) called"]
  PipelinePromise["stream/promises pipeline(...)"] -->|"error on any stream"| Reject["all streams destroyed, promise rejects"]
  Pipeline -->|"options.signal"| Abort["AbortSignal cancels the whole chain early"]
  Compose["compose(t1, t2, ...)"] -->|"fuses"| OneDuplex["a single Duplex you can pipe like any stream"]

pipe() ปล่อยการส่งต่อ error ให้เป็นหน้าที่คุณ; pipeline() (callback หรือ promise) ทำให้อัตโนมัติและรับ AbortSignal เพื่อยกเลิกก่อนได้

Trade-offs

ทางเลือก	ข้อดี	ต้นทุน
`.pipe()`	เชื่อม stream สองตัวด้วย code บรรทัดเดียวที่น้อยที่สุด	ไม่มีการส่งต่อ error หรือ cleanup อัตโนมัติ — leak หรือค้างถ้าคุณลืม wire เอง
`pipeline()` (callback)	ส่งต่อ error และ cleanup อัตโนมัติทั้ง chain	style callback — ซ้อนเพิ่มอีกชั้นถ้าคุณอยู่ใน async/await code อยู่แล้ว
`pipeline()` ของ `stream/promises`	รับประกันเหมือน callback form แต่ `await` ได้และประกอบกับ `try/catch`/`Promise.all` ได้	ต้องการ `async` context
`pipeline()` พร้อม `AbortSignal`	ยกเลิกทั้ง chain ก่อนได้เมื่อ timeout หรือ client หลุด ปล่อย resource ทันที	การ abort แสดงเป็น rejected promise/callback error — ผู้เรียกทุกคนต้องจัดการเส้นทางนั้น ไม่ใช่แค่เส้นทาง error ปกติ
`compose()`	เปลี่ยน pipeline หลายขั้นตอนให้เป็น `Duplex` ที่ใช้ซ้ำได้หนึ่งตัว export, test และ pipe ได้เหมือน stream ทั่วไป	เพิ่มชั้นของความอ้อม — error และ backpressure ยังไหลผ่านทุก stream ภายในอยู่ดี แค่ถูกห่อไว้

Common Misconceptions

“.pipe() กับ pipeline() ใช้แทนกันได้ — pipeline แค่ชื่อใหม่กว่า” — ความจริง: .pipe() ไม่เคยส่งต่อ error ระหว่าง stream; pipeline() ถูกเพิ่มเข้ามาเพื่อแก้ปัญหานั้นโดยเฉพาะ บวกกับ cleanup อัตโนมัติและ (ภายหลัง) การรองรับ AbortSignal
“pipeline() ที่ถูก abort จะแค่หยุดเงียบ ๆ” — ความจริง: มัน destroy ทุก stream ใน chain และ callback จะยิง (หรือ promise reject) ด้วย AbortError — คุณยังต้องจัดการเส้นทางนั้นอยู่ดี
“compose() สร้าง stream ชนิดใหม่” — ความจริง: มันคืนค่า Duplex ธรรมดา — interface เดียวกับ TCP socket — ที่บังเอิญถูกสร้างขึ้นโดยเชื่อม Transform เข้าด้วยกันภายใน

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

code ใช้ .pipe() เชื่อม stream มากกว่าหนึ่งตัวโดยไม่มี error listener บนตัวไหนเลยหรือเปล่า? นั่นคือ leak ที่รับประกันได้ทันทีที่ stream ใด ๆ ใน chain fail
snippet สร้าง pipeline() (หรือเวอร์ชัน stream/promises) ที่รับ input จากผู้ใช้โดยไม่มี AbortSignal/timeout เลยหรือเปล่า ทำให้ client ที่ช้าหรือค้างยึด resource ไว้ไม่มีกำหนด?
code implement การแปลงข้อมูลหลายขั้นตอนเป็นการเรียก .pipe() แยกกันหลายครั้งพร้อม error handling ที่ wire แยกกันทีละอัน ทั้งที่ compose() จะแสดงแนวคิดเดียวกันเป็นหน่วยเดียวที่ test ได้และใช้ซ้ำได้หรือเปล่า?

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

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

function compressFile(src, dest) {
fs.createReadStream(src)
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream(dest));
}

compressFile('./input.txt', './input.txt.gz');

เฉลย

ไม่มี stream ตัวไหนในสามตัวของ .pipe() chain นี้ที่มี error listener ถ้า input.txt ไม่มีอยู่ หรือดิสก์เต็มระหว่างเขียน หรือ gzip เจอ input ที่เสียหาย stream นั้นจะ emit 'error' — และเพราะ .pipe() ไม่ส่งต่อ error stream ตัวอื่นใน chain จะไม่ถูก destroy เลย file descriptor ของ read stream (หรือไฟล์ปลายทางที่เขียนไม่เสร็จ) อาจรั่ว วิธีแก้คือ stream.pipeline(fs.createReadStream(src), zlib.createGzip(), fs.createWriteStream(dest), function(err) { if (err) { /* handle it */ } }) ซึ่ง destroy ทุก stream ใน chain อัตโนมัติเมื่อเกิด error ใด ๆ