0Pricing
Node.js Backend Development Bootcamp · Pelajaran

Mengimplementasikan Stream Transform Khusus dengan _transform dan _flush

Bangun stream Transform yang dapat digunakan ulang untuk mengubah, memfilter, dan mengagregasi potongan data saat mengalir.

Mengimplementasikan Stream Transform Khusus dengan _transform dan _flush adalah pelajaran Node.js Backend Development Bootcamp gratis di CoddyKit. Ini adalah pelajaran 2 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar Node.js Backend Development Bootcamp, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus Node.js Backend Development Bootcamp mencakup 4 pelajaran total.

Bagian dari pelajaran ini belum diterjemahkan dan ditampilkan dalam bahasa Inggris.

Why Transform Streams?

A Transform stream is both readable and writable: it consumes input chunks, processes them, and pushes output chunks. It is the right tool whenever data must be mutated as it flows rather than buffered fully in memory.

  • Writable side accepts data via write() / pipe() from upstream.
  • Readable side emits processed data that downstream consumers read.

Typical backend uses: uppercasing/normalizing a payload, gzip-style encoding, CSV-to-JSON conversion, redacting secrets in a log pipeline, or counting bytes — all without loading the whole file or HTTP body into RAM.

The Two Hooks: _transform and _flush

A custom Transform stream is defined by implementing two internal methods. Node calls them for you — you never call them directly.

  • _transform(chunk, encoding, callback) — invoked once per incoming chunk. Do your work, push() any output, then signal completion with callback().
  • _flush(callback) — invoked once, after the last chunk, just before the stream ends. Use it to emit any trailing/aggregated data.

The leading underscore marks them as the framework-facing implementation. Consumers still use the public write, read, and pipe API.

A Minimal Uppercase Transform

The classic starting point: extend the Transform class and override _transform. Each chunk is a Buffer (unless object mode), so convert to a string, transform it, and push the result.

Calling callback() with no error tells Node this chunk is fully processed and it may deliver the next one. Passing the value as the second arg to callback is shorthand for push + callback().

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

class Upper extends Transform {
  _transform(chunk, encoding, callback) {
    const out = chunk.toString().toUpperCase();
    callback(null, out); // shorthand for this.push(out); callback();
  }
}

const up = new Upper();
up.on('data', (d) => process.stdout.write(d));
up.write('hello ');
up.write('streams\n');
up.end();

callback() Is a Contract

The callback in _transform is mandatory. Until you call it, Node assumes the chunk is still in progress and will not hand you the next one. This is how backpressure flows through your transform.

  • callback() — success, ready for next chunk.
  • callback(err) — emits an 'error' event and destroys the stream.
  • callback(null, data) — pushes data and signals success.

Forgetting to call callback is the #1 bug: the pipeline silently stalls forever with no error.

push() Multiple Times Per Chunk

One input chunk can produce zero, one, or many output chunks. Call this.push() as many times as needed before invoking callback(). This is what makes splitting (e.g. line-by-line) possible.

Below, a single write containing several lines is fanned out into one push per line.

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

class LineSplitter extends Transform {
  _transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    for (const line of lines) {
      if (line.length) this.push(line + ' <<\n');
    }
    callback();
  }
}

const s = new LineSplitter();
s.on('data', (d) => process.stdout.write(d));
s.end('alpha\nbeta\ngamma\n');

Filtering: Drop Chunks by Pushing Nothing

To filter, simply decide not to push. If a chunk should be discarded, call callback() without pushing anything — the data never reaches the readable side.

This pattern is ideal for redacting or dropping records mid-pipeline, e.g. removing log lines that contain a secret token.

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

class DropSecrets extends Transform {
  _transform(chunk, encoding, callback) {
    const line = chunk.toString();
    if (line.includes('SECRET')) {
      return callback(); // filtered out, nothing pushed
    }
    callback(null, line);
  }
}

const f = new DropSecrets();
f.on('data', (d) => process.stdout.write(d));
f.write('ok line 1\n');
f.write('this has a SECRET token\n');
f.write('ok line 2\n');
f.end();

Object Mode for Structured Records

By default chunks are Buffer/string. Set objectMode: true to push and receive JavaScript objects instead — essential for record-oriented pipelines (JSON rows, DB results, parsed events).

  • writableObjectMode / readableObjectMode can be set independently if input and output types differ.
  • In object mode, each push emits exactly one object regardless of size.
const { Transform } = require('node:stream');

class AddTax extends Transform {
  constructor() { super({ objectMode: true }); }
  _transform(order, encoding, callback) {
    callback(null, { ...order, total: order.price * 1.2 });
  }
}

const t = new AddTax();
t.on('data', (o) => console.log(o));
t.write({ id: 1, price: 100 });
t.write({ id: 2, price: 250 });
t.end();

_flush: Emit Trailing/Aggregated Data

_flush(callback) runs once after the final chunk, before 'end'. It is where you push anything you have been accumulating — a running total, a buffered partial line, or a closing delimiter.

You can push inside _flush just like in _transform. You must call its callback() so the stream can finish.

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

class Summer extends Transform {
  constructor() { super({ objectMode: true }); this.sum = 0; }
  _transform(num, encoding, callback) {
    this.sum += num;
    callback(); // aggregate, emit nothing yet
  }
  _flush(callback) {
    this.push({ total: this.sum }); // emit once at the end
    callback();
  }
}

const agg = new Summer();
agg.on('data', (o) => console.log(o));
[10, 20, 30, 40].forEach((n) => agg.write(n));
agg.end();

Buffering Partial Lines Across Chunk Boundaries

Chunks do not align with logical records. A line may be split across two chunks, so a robust line-parser keeps a leftover buffer between calls and flushes the remainder in _flush.

This combine-in-_transform, drain-in-_flush pattern is the backbone of real CSV/NDJSON parsers.

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

class LineParser extends Transform {
  constructor() { super({ readableObjectMode: true }); this.tail = ''; }
  _transform(chunk, encoding, callback) {
    const data = this.tail + chunk.toString();
    const parts = data.split('\n');
    this.tail = parts.pop(); // keep incomplete last segment
    for (const line of parts) this.push(line);
    callback();
  }
  _flush(callback) {
    if (this.tail) this.push(this.tail);
    callback();
  }
}

const p = new LineParser();
p.on('data', (l) => console.log('LINE:', l));
p.write('he');
p.write('llo\nwor');
p.write('ld\nlast');
p.end();

The Functional Shorthand: stream.Transform options

You don't always need a class. The Transform constructor accepts transform and flush functions directly — handy for small, one-off transforms.

Inside these functions, this is still the stream, so this.push() works. The class form is preferred when you want reusable, named, instantiable components; the inline form is great for quick glue.

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

const csvToRows = new Transform({
  readableObjectMode: true,
  transform(chunk, encoding, callback) {
    for (const line of chunk.toString().trim().split('\n')) {
      const [id, name] = line.split(',');
      this.push({ id: Number(id), name });
    }
    callback();
  },
});

csvToRows.on('data', (row) => console.log(row));
csvToRows.end('1,Ada\n2,Linus\n3,Grace');

Composing in a Pipeline

Transform streams shine when chained. Use stream.pipeline() (callback or promise form) instead of raw .pipe() — it propagates errors and destroys every stream on failure, preventing leaks.

Here a source feeds an uppercaser, then a suffix-adder, then stdout. Each transform stays small and reusable.

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

const make = (fn) => new Transform({
  transform(chunk, enc, cb) { cb(null, fn(chunk.toString())); },
});

const upper = make((s) => s.toUpperCase());
const bang = make((s) => s + '!\n');

pipeline(
  Readable.from(['log a\n', 'log b\n']),
  upper,
  bang,
  process.stdout,
  (err) => { if (err) console.error('failed', err); else console.error('done'); }
);

Quick Check: Where Do Trailing Aggregates Go?

You are building a Transform that counts total bytes seen and must emit a single summary object after all input is processed. Which method should push that summary?

Recap: Transform Stream Essentials

You can now build reusable Transform streams that mutate, filter, and aggregate flowing data:

  • _transform(chunk, enc, cb) — process each chunk; push zero or more outputs; always call cb() (or cb(null, data)).
  • _flush(cb) — runs once at the end to emit trailing or aggregated data; must call cb().
  • Filter by pushing nothing; split by pushing many times; aggregate by accumulating state and flushing.
  • objectMode (and the independent readable/writable variants) carries structured records.
  • Buffer partial records in _transform and drain them in _flush.
  • Compose with stream.pipeline() for safe error handling and cleanup.

Never forget the callback — a missing cb() silently stalls the entire pipeline.

Pertanyaan yang Sering Diajukan

Apakah pelajaran “Mengimplementasikan Stream Transform Khusus dengan _transform dan _flush” gratis?

Ya — teks lengkap “Mengimplementasikan Stream Transform Khusus dengan _transform dan _flush” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus Node.js Backend Development Bootcamp, upgrade ke CoddyKit PRO. Kursus Node.js Backend Development Bootcamp mencakup 4 pelajaran total.

Apa yang akan aku pelajari di “Mengimplementasikan Stream Transform Khusus dengan _transform dan _flush”?

Bangun stream Transform yang dapat digunakan ulang untuk mengubah, memfilter, dan mengagregasi potongan data saat mengalir. Kamu berlatih Node.js Backend Development Bootcamp dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.

Apakah aku perlu pengalaman untuk memulai Node.js Backend Development Bootcamp?

Tidak diperlukan pengalaman sebelumnya. Node.js Backend Development Bootcamp di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 2 dari 4.

Berapa lama pelajaran “Mengimplementasikan Stream Transform Khusus dengan _transform dan _flush” memakan waktu?

Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.

Bisakah aku menulis dan menjalankan kode dalam pelajaran Node.js Backend Development Bootcamp ini?

Ya. Setiap pelajaran Node.js Backend Development Bootcamp menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.

Semua pelajaran dalam kursus ini

  1. Internal Stream Readable, Writable, Duplex, dan Transform
  2. Mengimplementasikan Stream Transform Khusus dengan _transform dan _flush
  3. Backpressure, pipe(), dan Utilitas pipeline()
  4. Iterator Asinkron dan for-await-of pada Stream
← Kembali ke Node.js Backend Development Bootcamp