Bootcamp backendontwikkeling met Node.js · Les

Internals van readable-, writable-, duplex- en transformstreams

Krijg inzicht in de vier streamtypen en hoe de interne buffer en highWaterMark hun gedrag bepalen.

Les 1 van 413 stappen

Internals van readable-, writable-, duplex- en transformstreams is een gratis Bootcamp backendontwikkeling met Node.js-les op CoddyKit. Dit is les 1 van 4. Je kunt de volledige les hieronder gratis lezen en daarna in de browser praktisch oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject Bootcamp backendontwikkeling met Node.js. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus Bootcamp backendontwikkeling met Node.js bevat in totaal 4 lessen.

Waarom streams bestaan

Met Node.js-streams kun je gegevens stuk voor stuk verwerken in plaats van alles tegelijk in het geheugen te laden. Dit is essentieel voor backendwerk, zoals het aanbieden van grote bestanden, het doorsturen van HTTP-lichamen of het doorsturen van database-exports.

Readable — een bron waaruit je leest (bestand lezen, HTTP-verzoek)
Writable — een bestemming waarnaar je schrijft (bestand schrijven, HTTP-antwoord)
Duplex — zowel leesbaar als schrijfbaar, met onafhankelijke kanalen (TCP-socket)
Transform — een Duplex waarvan de uitvoer een functie van de invoer is (gzip, versleuteling)

Achter elk hiervan zit een interne buffer die wordt beheerd door één getal: highWaterMark.

De interne buffer en highWaterMark

Elke stream houdt een interne buffer bij in zijn _readableState of _writableState. De highWaterMark (HWM) is de drempel, geen harde limiet, waarop de stream aangeeft dat er "genoeg" is gebufferd.

De standaard-HWM voor bytestreams is 16 KB (16384 bytes).
In de objectmodus telt de HWM objecten, standaard 16.

Wanneer de buffer van een Readable tot de HWM is gevuld, stopt deze met ophalen uit de bron. Wanneer de buffer van een Writable de HWM overschrijdt, geeft write() false terug — het signaal dat bekendstaat als terugdruk.

const fs = require('fs');

const rs = fs.createReadStream('/etc/hostname', { highWaterMark: 4 });
console.log('configured HWM:', rs.readableHighWaterMark);

rs.on('data', (chunk) => {
  console.log('chunk of', chunk.length, 'bytes:', JSON.stringify(chunk.toString()));
});
rs.on('end', () => console.log('done'));

Readable-streams: doorstromen of pauzeren

Een Readable werkt in een van twee modi:

Paused (standaard): je moet expliciet read() aanroepen om gegevens op te halen.
Flowing: gegevens worden via 'data'-gebeurtenissen naar je toegestuurd zodra ze binnenkomen.

Door een 'data'-listener toe te voegen of .pipe() aan te roepen, schakelt de stream over naar de doorstromende modus. Met .pause() schakel je deze terug. Dit begrijpen is essentieel om het geheugengebruik te beheersen.

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

const r = Readable.from(['a', 'b', 'c']);

// Paused mode: pull explicitly
r.on('readable', () => {
  let chunk;
  while ((chunk = r.read()) !== null) {
    console.log('pulled:', chunk);
  }
});
r.on('end', () => console.log('stream finished'));

Een aangepaste Readable implementeren

Breid Readable uit en implementeer _read(size) om je eigen bron te bouwen. Roep daarin this.push(chunk) aan om de buffer te vullen, en this.push(null) om het einde van de stream (EOF) aan te geven.

Het belangrijkste detail: wanneer push() false teruggeeft, heeft de interne buffer de HWM bereikt. Een goed opgevoede producent stopt met pushen totdat _read opnieuw wordt aangeroepen.

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

class Counter extends Readable {
  constructor(max) {
    super({ objectMode: true, highWaterMark: 2 });
    this.max = max;
    this.current = 1;
  }
  _read() {
    if (this.current > this.max) {
      this.push(null); // EOF
      return;
    }
    const keepGoing = this.push({ n: this.current++ });
    console.log('pushed, buffer wants more:', keepGoing);
  }
}

Readable.from([]); // noop
const c = new Counter(5);
c.on('data', (obj) => console.log('consumed:', obj.n));
c.on('end', () => console.log('all consumed'));

Writable-streams en de retourwaarde van write()

Een Writable buffert binnenkomende chunks en verwerkt ze via _write(chunk, encoding, callback). Je MOET de callback aanroepen zodra elke chunk is verwerkt — zo weet de stream dat hij zijn buffer kan legen en meer kan accepteren.

De retourwaarde van write() is je signaal voor terugdruk:

true — de buffer staat onder de HWM, blijf schrijven.
false — de buffer staat op of boven de HWM, je MOET stoppen en wachten op de gebeurtenis 'drain'.

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

class SlowSink extends Writable {
  constructor() {
    super({ highWaterMark: 8 });
  }
  _write(chunk, enc, cb) {
    console.log('writing', chunk.length, 'bytes');
    setTimeout(cb, 50); // simulate slow I/O
  }
}

const sink = new SlowSink();
const ok = sink.write(Buffer.alloc(16));
console.log('write returned:', ok); // false -> over HWM
sink.once('drain', () => console.log('drained, safe to write again'));
sink.end(() => console.log('finished'));

Terugdruk handmatig respecteren

Als je een false van write() negeert en blijft schrijven, groeit de interne buffer onbeperkt en kan je proces zonder geheugen komen te zitten. Het juiste handmatige patroon is de productie pauzeren totdat 'drain' wordt geactiveerd.

In de praktijk schrijf je dit zelden zelf — .pipe() en pipeline() doen dit voor je — maar als je de werking kent, begrijp je waarom doorsturen veilig is.

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

const sink = new Writable({
  highWaterMark: 4,
  write(chunk, enc, cb) { setTimeout(cb, 20); }
});

let i = 0;
function writeMore() {
  let ok = true;
  while (i < 10 && ok) {
    ok = sink.write(String(i++));
  }
  if (i < 10) {
    console.log('backpressure at i =', i, '-> wait for drain');
    sink.once('drain', writeMore);
  } else {
    sink.end(() => console.log('done'));
  }
}
writeMore();

pipe(): automatische stroomregeling

readable.pipe(writable) verbindt een bron met een bestemming en houdt automatisch rekening met terugdruk: wanneer de bestemming false teruggeeft, roept pipe source.pause() aan; bij 'drain' roept het source.resume() aan.

Het nadeel van een losse .pipe() is foutafhandeling: als de bron een fout geeft, wordt de bestemming NIET automatisch gesloten, wat bestandsdescriptors kan laten lekken. Gebruik in productie liever stream.pipeline().

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

// gzip a file: Readable -> Transform -> Writable
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'))
  .on('finish', () => console.log('compressed'));

Duplex-streams: twee onafhankelijke kanalen

Een Duplex-stream is zowel Readable als Writable, maar de twee kanten zijn onafhankelijk — wat je schrijft, verschijnt niet automatisch aan de leeskant. Een TCP-socket is het standaardvoorbeeld: bytes die je schrijft, gaan naar de andere partij; bytes die je leest, komen van de andere partij.

Geef zowel _read als _write op om er een te implementeren. Elke kant heeft een eigen buffer en een eigen highWaterMark.

const { Duplex } = require('stream');

class Echo extends Duplex {
  constructor() {
    super();
    this.queue = [];
  }
  _write(chunk, enc, cb) {
    this.queue.push(chunk.toString().toUpperCase());
    cb();
  }
  _read() {
    const item = this.queue.shift();
    this.push(item !== undefined ? item : null);
  }
}

const d = new Echo();
d.on('data', (c) => console.log('read side:', c.toString()));
d.write('hello');
d.write('world');
d.end();

Transform-streams: uitvoer afgeleid van invoer

Een Transform is een speciale Duplex waarbij de leesbare kant wordt berekend uit de schrijfbare kant. In plaats van afzonderlijke _read/_write-methoden implementeer je één _transform(chunk, encoding, callback) en geef je resultaten door via this.push() of het tweede argument van de callback.

Een optionele _flush(callback) wordt eenmaal aan het einde uitgevoerd — ideaal om afsluitende gegevens uit te sturen, zoals een laatste controlesom of sluithaakje.

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

class UpperCase extends Transform {
  _transform(chunk, enc, cb) {
    cb(null, chunk.toString().toUpperCase());
  }
  _flush(cb) {
    this.push('\n-- END --\n');
    cb();
  }
}

const t = new UpperCase();
t.on('data', (c) => process.stdout.write(c.toString()));
t.write('node ');
t.write('streams');
t.end();

Objectmodus en HWM-telling

Standaard verplaatsen streams Buffers/strings en telt de HWM bytes. Geef { objectMode: true } door en de stream verplaatst willekeurige JS-waarden, waarbij de HWM in plaats daarvan objecten telt.

Standaard-HWM in bytemodus: 16384 bytes
Standaard-HWM in objectmodus: 16 objecten

Dit is belangrijk voor backendwerkstromen: een Transform die NDJSON parseert, kan onbewerkte bytes lezen (schrijfbare kant, bytemodus), maar geparseerde objecten uitsturen (leesbare kant, objectmodus) met readableObjectMode.

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

// Bytes in, objects out
class NdjsonParse extends Transform {
  constructor() {
    super({ writableObjectMode: false, readableObjectMode: true });
    this.buf = '';
  }
  _transform(chunk, enc, cb) {
    this.buf += chunk.toString();
    const lines = this.buf.split('\n');
    this.buf = lines.pop();
    for (const line of lines) {
      if (line.trim()) this.push(JSON.parse(line));
    }
    cb();
  }
}

const p = new NdjsonParse();
p.on('data', (o) => console.log('parsed object:', o));
p.write('{"id":1}\n{"id":2}\n');
p.end();

pipeline(): samenstelling voor productiegebruik

stream.pipeline() schakelt een willekeurig aantal streams aaneen en geeft, anders dan .pipe(), fouten door en ruimt elke stream op (door ze te vernietigen) wanneer een ervan mislukt of klaar is. Zo voorkom je gelekte bestandsdescriptors en vastgelopen sockets.

De promise-gebaseerde vorm (require('stream/promises')) integreert netjes met async/await in routehandlers.

const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');

async function gzipFile(src, dest) {
  await pipeline(
    fs.createReadStream(src),
    zlib.createGzip(),
    fs.createWriteStream(dest)
  );
  console.log('pipeline complete:', dest);
}

gzipFile('access.log', 'access.log.gz').catch((err) => {
  console.error('pipeline failed, all streams destroyed:', err.message);
});

Korte controle: signaal voor terugdruk

Je schrijft in een lus een grote gegevensset naar een aangepaste Writable-stream. Je wilt onbeperkte geheugengroei voorkomen door terugdruk te respecteren. Welk signaal vertelt je dat je moet stoppen met schrijven en wachten?

Beantwoord dit in de interactieve les

Samenvatting

Je begrijpt nu de vier typen streams en de buffermechanismen erachter:

Readable — bron; implementeer _read, push gegevens en push(null) voor EOF; modi voor doorstromen en pauzeren.
Writable — bestemming; implementeer _write en roep de callback aan; als write() false retourneert, betekent dit backpressure: wacht op 'drain'.
Duplex — onafhankelijke lees- en schrijfkanalen, elk met een eigen buffer en HWM (bijvoorbeeld een TCP-socket).
Transform — uitvoer die via _transform uit de invoer wordt afgeleid, met optioneel _flush.

De highWaterMark (standaard 16 KB bytes / 16 objecten) is een drempel, geen harde limiet. Deze bepaalt wanneer buffers aangeven dat ze "vol" zijn. Gebruik in productie altijd liever pipeline() dan een losse .pipe() voor correcte foutdoorgifte en opruiming.

Gratis beginnen

Leer JavaScript met een AI-tutor — gratis

Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.

Cursussen: 22
Lessen: 92

Gratis beginnen met leren Bekijk de volledige cursus

Veelgestelde vragen

Is de les “Internals van readable-, writable-, duplex- en transformstreams” gratis?

Ja — de volledige tekst van “Internals van readable-, writable-, duplex- en transformstreams” kun je hier gratis op het web lezen. Als je interactief wilt oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is, en de rest van de cursus Bootcamp backendontwikkeling met Node.js wilt ontgrendelen, kun je upgraden naar CoddyKit PRO. De cursus Bootcamp backendontwikkeling met Node.js bevat in totaal 4 lessen.

Wat leer ik in “Internals van readable-, writable-, duplex- en transformstreams”?

Krijg inzicht in de vier streamtypen en hoe de interne buffer en highWaterMark hun gedrag bepalen. Je oefent met Bootcamp backendontwikkeling met Node.js door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.

Heb ik ervaring nodig om met Bootcamp backendontwikkeling met Node.js te beginnen?

Ervaring vooraf is niet nodig. Bootcamp backendontwikkeling met Node.js op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 1 van 4.

Hoe lang duurt de les “Internals van readable-, writable-, duplex- en transformstreams”?

De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.

Kan ik code schrijven en uitvoeren in deze les over Bootcamp backendontwikkeling met Node.js?

Ja. Elke les over Bootcamp backendontwikkeling met Node.js bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.

Alle lessen in deze cursus

Internals van readable-, writable-, duplex- en transformstreams
Aangepaste transformstreams implementeren met _transform en _flush
Backpressure, pipe() en het pipeline()-hulpprogramma
Async-iterators en for-await-of over streams

← Terug naar Bootcamp backendontwikkeling met Node.js