Duplex and Transform Streams

A Duplex has a readable and a writable side with no required relationship between them; a TCP socket is the canonical example. A Transform is the case where the readable side is the writable side after processing: zlib.createGzip() and every parser you write.

Implement _transform(chunk, encoding, callback) and call callback(err, outputChunk), or this.push() repeatedly and then callback(). _flush(callback) emits what is left when the input ends — the half-line every chunked parser accumulates.

A line splitter and an object mapper as transformsJavaScript

class LineSplitter extends Transform {
  constructor() { super({ readableObjectMode: true }); this.tail = ''; }
  _transform(chunk, enc, cb) {
    const parts = (this.tail + chunk).split('\n');
    this.tail = parts.pop();                    // keep the incomplete tail
    for (const line of parts) if (line) this.push(line);
    cb();
  }
  _flush(cb) { if (this.tail) this.push(this.tail); cb(); }
}
const toRecord = new Transform({ objectMode: true, transform(line, enc, cb) {
  const [id, region, amount] = line.split(',');
  cb(null, { id: Number(id), region, amount: Number(amount) });
} });
const source = Readable.from(['1,emea,10.5\n2,ap', 'ac,3.25\n']);
console.log(JSON.stringify(await source.pipe(new LineSplitter()).pipe(toRecord).toArray()));

Output

[{"id":1,"region":"emea","amount":10.5},{"id":2,"region":"apac","amount":3.25}]

The record with id: 2 proves the point: its line was split across two source chunks and LineSplitter rejoined it. readableObjectMode: true with a byte-mode writable side is the normal shape for a parser — bytes in, objects out; a serializer sets writableObjectMode instead.

MENU

Duplex and Transform Streams