Bootcamp backendontwikkeling met Node.js · Les

Aangepaste transformstreams implementeren met _transform en _flush

Bouw herbruikbare Transform-streams die chunks wijzigen, filteren en aggregeren terwijl gegevens erdoorheen stromen.

Les 2 van 413 stappen

Aangepaste transformstreams implementeren met _transform en _flush is een gratis Bootcamp backendontwikkeling met Node.js-les op CoddyKit. Dit is les 2 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 Transform-streams?

Een Transform-stream is zowel leesbaar als schrijfbaar: hij verbruikt invoerchunks, verwerkt ze en pusht uitvoerchunks. Dit is het juiste hulpmiddel wanneer gegevens tijdens het doorstromen moeten worden gewijzigd, in plaats van volledig in het geheugen te worden gebufferd.

Schrijfkant accepteert gegevens via write() / pipe() van de upstream.
Leeskant geeft verwerkte gegevens uit die downstream-verbruikers lezen.

Typische backendtoepassingen zijn: een payload omzetten naar hoofdletters of normaliseren, codering in gzip-stijl, conversie van CSV naar JSON, geheimen onleesbaar maken in een logging-pipeline of bytes tellen — allemaal zonder het volledige bestand of de HTTP-body in het RAM te laden.

De twee hooks: _transform en _flush

Een aangepaste Transform-stream wordt gedefinieerd door twee interne methoden te implementeren. Node roept ze voor je aan — je roept ze nooit rechtstreeks aan.

_transform(chunk, encoding, callback) — wordt één keer per binnenkomende chunk aangeroepen. Voer je bewerking uit, push() eventuele uitvoer en geef daarna aan dat je klaar bent met callback().
_flush(callback) — wordt eenmalig aangeroepen na de laatste chunk, vlak voordat de stream eindigt. Gebruik deze methode om resterende of samengevoegde gegevens uit te geven.

De voorloopunderscore geeft aan dat dit de implementatie voor het framework is. Verbruikers gebruiken nog steeds de openbare API met write, read en pipe.

Een minimale Transform voor hoofdletters

Het klassieke startpunt: breid de klasse Transform uit en overschrijf _transform. Elke chunk is een Buffer (tenzij de objectmodus actief is), dus zet deze om naar een string, transformeer de string en push het resultaat.

Als je callback() zonder fout aanroept, laat je Node weten dat deze chunk volledig is verwerkt en dat de volgende kan worden geleverd. De waarde als tweede argument aan callback doorgeven is een verkorte vorm van 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 een contract

De callback in _transform is verplicht. Totdat je deze aanroept, gaat Node ervan uit dat de chunk nog wordt verwerkt en geeft Node je niet de volgende chunk. Zo stroomt backpressure door je transform heen.

callback() — geslaagd, klaar voor de volgende chunk.
callback(err) — genereert een 'error'-event en vernietigt de stream.
callback(null, data) — pusht data en geeft aan dat de bewerking is geslaagd.

Vergeten om callback aan te roepen is de nummer-éénfout: de pipeline loopt stilletjes voor altijd vast zonder fout.

push() meerdere keren per chunk

Eén invoerchunk kan nul, één of meerdere uitvoerchunks opleveren. Roep this.push() zo vaak aan als nodig is voordat je callback() aanroept. Hierdoor kun je gegevens splitsen, bijvoorbeeld regel voor regel.

Hieronder wordt één write met meerdere regels opgesplitst in één push per regel.

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');

Filteren: chunks verwijderen door niets te pushen

Om te filteren hoef je eenvoudigweg niets te pushen. Als een chunk moet worden verwijderd, roep je callback() aan zonder iets te pushen — de gegevens bereiken de leeskant nooit.

Dit patroon is ideaal om records midden in een pipeline onleesbaar te maken of te verwijderen, bijvoorbeeld logregels met een geheim 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();

Objectmodus voor gestructureerde records

Standaard zijn chunks Buffers of strings. Stel objectMode: true in om in plaats daarvan JavaScript-objecten te pushen en te ontvangen — essentieel voor recordgerichte pipelines (JSON-rijen, databaseresultaten, geparseerde events).

writableObjectMode / readableObjectMode kunnen onafhankelijk van elkaar worden ingesteld als invoer- en uitvoertypen verschillen.
In de objectmodus levert elke push precies één object op, ongeacht de grootte.

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: resterende of samengevoegde gegevens uitgeven

_flush(callback) wordt één keer uitgevoerd na de laatste chunk, vóór 'end'. Hier push je alles wat je hebt verzameld — een lopend totaal, een gebufferde gedeeltelijke regel of een afsluitend scheidingsteken.

Je kunt binnen _flush net als in _transform push gebruiken. Je moet de callback() ervan aanroepen zodat de stream kan afronden.

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();

Gedeeltelijke regels bufferen over chunkgrenzen heen

Chunks vallen niet altijd samen met logische records. Een regel kan over twee chunks worden gesplitst, dus een robuuste regelparser bewaart tussen aanroepen een restbuffer en geeft de rest uit in _flush.

Dit patroon — combineren in _transform en leegmaken in _flush — vormt de basis van echte 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();

De functionele verkorte vorm: opties voor stream.Transform

Je hebt niet altijd een klasse nodig. De constructor van Transform accepteert rechtstreeks transform- en flush-functies — handig voor kleine, eenmalige transforms.

Binnen deze functies verwijst this nog steeds naar de stream, dus this.push() werkt. De klassevorm heeft de voorkeur voor herbruikbare, benoemde en instantieerbare componenten; de inlinevorm is ideaal voor snelle koppelingen.

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');

Samenstellen in een pipeline

Transform-streams komen goed tot hun recht wanneer je ze aan elkaar koppelt. Gebruik stream.pipeline() (in callback- of promisevorm) in plaats van een kale .pipe() — deze geeft fouten door en vernietigt elke stream bij een fout, zodat lekken worden voorkomen.

Hier voert een bron gegevens aan een omzetter naar hoofdletters, daarna aan een toevoeging van een achtervoegsel en vervolgens aan stdout. Elke transform blijft klein en herbruikbaar.

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'); }
);

Korte controle: waar komen resterende aggregaten terecht?

Je bouwt een Transform die het totale aantal ontvangen bytes telt en één samenvattingsobject moet uitgeven nadat alle invoer is verwerkt. Welke methode moet die samenvatting pushen?

Beantwoord dit in de interactieve les

Samenvatting: de essentie van Transform-streams

Je kunt nu herbruikbare Transform-streams bouwen die gegevens tijdens het doorstromen wijzigen, filteren en aggregeren:

_transform(chunk, enc, cb) — verwerk elke chunk; push nul of meer uitvoerwaarden; roep altijd cb() aan (of cb(null, data)).
_flush(cb) — wordt aan het einde één keer uitgevoerd om resterende of samengevoegde gegevens uit te geven; cb() moet worden aangeroepen.
Filter door niets te pushen; splits door meerdere keren te pushen; aggregeer door de status te verzamelen en te flushen.
objectMode (en de onafhankelijke leesbare/schrijfbare varianten) vervoert gestructureerde records.
Buffer gedeeltelijke records in _transform en maak ze leeg in _flush.
Stel streams samen met stream.pipeline() voor veilige foutafhandeling en opruiming.

Vergeet de callback nooit — een ontbrekende cb() laat de hele pipeline stilletjes vastlopen.

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 “Aangepaste transformstreams implementeren met _transform en _flush” gratis?

Ja — de volledige tekst van “Aangepaste transformstreams implementeren met _transform en _flush” 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 “Aangepaste transformstreams implementeren met _transform en _flush”?

Bouw herbruikbare Transform-streams die chunks wijzigen, filteren en aggregeren terwijl gegevens erdoorheen stromen. 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 2 van 4.

Hoe lang duurt de les “Aangepaste transformstreams implementeren met _transform en _flush”?

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