Working with Streams
1. Creating Readable Streams (fs.createReadStream)
Example
import { createReadStream } from "node:fs";
const rs = createReadStream("big.csv", { encoding: "utf8", highWaterMark: 64 * 1024 });
for await (const chunk of rs) process(chunk);
| Mode | Behavior |
|---|
| Flowing | Pushed via data event |
| Paused | Pulled via read() |
2. Creating Writable Streams (fs.createWriteStream)
| API | Use |
|---|
ws.write(chunk) | Returns false when buffer full |
ws.end(chunk?) | Finish stream |
Event drain | Resume writing |
Event finish | All data flushed |
Example: gzip on the fly
import { pipeline } from "node:stream/promises";
import { createReadStream, createWriteStream } from "node:fs";
import { createGzip } from "node:zlib";
await pipeline(createReadStream("file.txt"), createGzip(), createWriteStream("file.txt.gz"));
4. Using Duplex Streams
| Type | Examples |
|---|
| Duplex | TCP socket, ZLib stream |
| Transform (subclass) | Hash, Compression, Crypto cipher |
5. Handling Stream Events
| Event | Stream Type |
|---|
data | Readable (flowing) |
end | Readable |
readable | Readable (paused) |
drain | Writable |
finish | Writable |
close | Both |
error | Both |
6. Piping Streams
Example: pipeline (preferred)
import { pipeline } from "node:stream/promises";
await pipeline(src, transform, dest);
Warning: Avoid raw .pipe() chains — they don't propagate errors. Use pipeline.
7. Controlling Stream Flow
| Method | Effect |
|---|
stream.pause() | Stop emitting data |
stream.resume() | Resume |
stream.read(n) | Pull bytes (paused mode) |
8. Using Backpressure
Backpressure
write() returns false → producer should pause
↓
wait for "drain" event
↓
resume writing
| Tool | Handles BP |
|---|
pipeline | ✅ automatically |
Manual pipe | Yes, but no error propagation |
| Manual write loop | Must check return value |
9. Working with Object Mode
Example
import { Transform } from "node:stream";
const toUpper = new Transform({
objectMode: true,
transform(obj, _, cb) { cb(null, { ...obj, name: obj.name.toUpperCase() }); }
});
10. Using stream.Readable.from() Method
| Source | Result |
|---|
| Iterable | Readable.from([1,2,3]) |
| Async iterable | Readable.from(asyncGen()) |
| String / Buffer | Single chunk stream |
11. Consuming Streams with Async Iterators
Example
for await (const chunk of readable) {
process(chunk);
}
12. Understanding Highwater Mark
| Default | Value |
|---|
| Byte streams | 16 KB |
| Object mode | 16 objects |
| Tuning up | Fewer reads, more memory |