In the previous lesson you consumed streams that already existed: createReadStream produced chunks and createWriteStream received them. You understood backpressure, you saw that pipe() handles it for you and you discovered its flaw: it neither propagates errors nor cleans up what it leaves behind.

Now comes the other half. You are going to write your own streams — the middle link that turns one thing into another — and to assemble the complete processing pipe of Escena Viva: read data/sales.csv, turn every line into a sale object, filter by venue, aggregate by session and write the result, all with constant memory, whether the file is two kilobytes or eight hundred megabytes. And you will replace pipe() with stream.pipeline, the real tool: it propagates errors, destroys the whole chain when something fails and has a promise-based version. At the end you will see the most readable way of writing transformations, which is not a class but an async generator.

1. objectMode: streams that carry objects
2. The Transform stream: _transform and _flush
3. A Transform that parses CSV lines
4. Custom Readable and Writable streams
5. stream.pipeline versus pipe()
6. The Escena Viva processing pipe
7. Compression with zlib in the pipe
8. Readable.from() and async generators
9. Errors and cancellation with AbortSignal

By default, a stream carries bytes: Buffer objects or strings. That is the right thing for files and sockets, but useless when what flows through the pipe are sales, events or already interpreted records. objectMode: true changes that nature:

The mode can be declared per side: new Transform({ readableObjectMode: true, writableObjectMode: false }) describes exactly what a parser does — it receives text and emits objects — while a plain objectMode: true switches on both sides. And null deserves a warning: in a stream, push(null) means "that's the end", so you cannot send null as data, not even in object mode; to represent the absence of something, use undefined or a marker object.

A Transform is a stream that reads on one side, does something and writes on the other. It is implemented with two methods:

The mechanics are always the same: _transform receives a chunk, does its work, emits zero or more results with this.push(...) and announces it has finished by calling callback(); _flush runs exactly once at the end, when nothing more will come in, and is the last chance to emit.

Three rules that admit no exception:

• callback() must be called exactly once per _transform. If you never call it, the pipe hangs forever with no message at all; if you call it twice, Node throws ERR_MULTIPLE_CALLBACK.
• Errors are signaled with callback(error), not with throw: an exception thrown in an asynchronous context is caught by nobody.
• _flush is where the accumulated state lives. Everything you cannot emit until you see the end — the last line with no break, an aggregate total, the closing of a JSON document — comes out there. That is what tells a Transform apart from a plain map: it can accumulate across chunks.

Here is the real problem: the file's chunks do not line up with the lines. A stateful Transform solves splitting and conversion in a single step.

Notice the role of #leftover: it is the problem readline solved for us in the previous lesson, now solved by hand because we need to emit objects, not lines. And _flush is where that pending leftover gets processed: without that line, the last sale would be silently lost whenever the file does not end in a break. It is a classic bug and hard to spot, because it fails on one record out of a million.

A Readable produces data. It is implemented with _read(), which Node calls when it wants more material, and inside you emit with this.push(value); when there is nothing left to produce, this.push(null) signals the end of the stream. In practice you almost never need to write one: Readable.from() (section 8) covers nearly every case in a single line.

A Writable consumes data. It implements _write(chunk, encoding, callback), and that callback is the key piece: until you call it, the stream considers you still busy. That is where the backpressure you felt in the previous lesson is born.

If the work in _write were asynchronous — saving to a database, calling an API — it would be enough to call the callback when it finished: the whole pipe slows down by itself in the meantime. There is also _writev for processing several items at once, useful when inserting in batches.

You already saw the problem with pipe(). stream.pipeline fixes it, and its promise version — const { pipeline } = require('node:stream/promises') — is the one we will always use.

The practical difference is enormous:

Course rule: never pipe() in production code. pipe is there to explain the concept; pipeline is what you write.

With the pieces above we assemble the complete process: from raw CSV to a JSON report, filtering by venue and never loading the whole file.

The 1811 tickets match the seed, which is the check we were after. And look at the shape of the pipe: four stages, each with a single responsibility, chained in a pipeline that reads top to bottom; changing the filter, adding a validation or writing to another destination means touching one line. A design detail: the SessionAccumulator goes last because an aggregation cannot emit anything until it has seen the final item; if you needed to continue the pipe after aggregating, it would be a Transform emitting its result in _flush.

Archiving the history means slotting in one more stage: zlib.createGzip() is a Transform just like yours.

No encoding was given to the reader here: the data travels as Buffer objects from start to finish, which is right for compression — turning it into text and back into bytes would add nothing and cost CPU. Decompression is the same pipe in reverse: just slot zlib.createGunzip() between the .gz reader and the parser to process the compressed history without decompressing it to disk. That is what makes the model powerful: stages combine without any of them knowing anything about the others, and the parser has no idea its bytes came from a compressed file.

Readable.from() turns any iterable — array, Map, generator — into a stream:

It is invaluable for the tests in Module 9: you feed a pipe with fake data without touching the disk. But the best part comes now: an async generator can be used directly as a stage of a pipeline, and it is the most readable way of writing a transformation.

The comparison is revealing:

For new transformations, always start with an async generator. Fall back to the Transform class when you need an object with state readable from outside — like our SessionAccumulator, whose result is read when it finishes — or when you have to integrate with an API expecting a specific stream.

A failure in any stage aborts the pipe and rejects the promise; what reaches the catch is the original error, with its code intact:

And pipeline accepts a cancellation signal, just like fetch:

On aborting, pipeline destroys every stream in the chain: descriptors are released and the promise is rejected with an AbortError. It is exactly the work that in the previous lesson had to be written by hand with crossed on('error') handlers and destroy(). In Module 4 you will come back to this: when an HTTP client closes the connection halfway through a download, the right move is to abort the pipe generating the response instead of working on for nobody.

• Forgetting callback() in _transform or _write. The pipe hangs silently, with no error and no trace. It is the number one failure.
• Using throw inside _transform instead of callback(error), or forgetting _flush, so everything accumulated — the last line, the total, the closing — is lost without warning.
• Not declaring objectMode when emitting objects: Node complains with ERR_INVALID_ARG_TYPE because it expects a Buffer or a string. And push(null) is not data: it means end of stream, always.
• Still using pipe(). If an error from the first stage never reaches the last one, you have leaked descriptors and invisible failures.
• Tip: one stage, one responsibility. A Transform that parses, filters and aggregates all at once is impossible to test; three chained stages are tested separately with Readable.from(). And if you hesitate between a class and a generator, start with the generator: turning it into a class later is easy, the other way round less so.

Write an async generator validateSales that slots in between the parser and the filter, and checks every sale: ticketCode with the format EV-<year>-<6 digits>, sessionId shaped as ses-NNN-M, priceCents a positive integer and channel one of web, box-office and app. Valid sales pass through; invalid ones are counted and logged on stderr with their line number. At the end it must report how many were discarded.

Write src/reports/sales-by-channel.js with a pipe that reads data/sales.csv, aggregates by channel (tickets, revenue and average ticket) and writes reports/sales-by-channel.json.gz compressed, without producing the uncompressed JSON on disk. Hint: Readable.from([JSON.stringify(report, null, 2)]) turns the result into the source of a second pipe.

Write a function generateWithRetry(options) that runs the generateSalesBySession pipe with a 10-second AbortSignal and, if it fails with an AbortError or a transient I/O error, retries it up to three times using retry and sleep from Module 2. It must not retry on ENOENT or on corrupt data: those failures are not fixed by repeating.

Solution 1. The validator as a generator turns out almost declarative:

The final block, after the for await, is the exact equivalent of a Transform class's _flush: it runs when the input is exhausted. If you needed to parameterize the validator, wrap it in a function returning the generator, like the filter in section 8.

Solution 2. Two chained pipes, the second one with Readable.from:

The JSON never touches the disk uncompressed: it leaves memory as a stream, goes through the compressor and lands in the file. With a small report it makes no difference; with one of hundreds of megabytes it is the difference between needing twice the free space or not.

Solution 3. The key is to classify the errors before deciding whether to retry:

Retrying an ENOENT is pointless: the file is not going to show up on its own and all you achieve is delaying the error message three times over. An AbortError from a time limit or a transient EBUSY, on the other hand, may well be resolved on the second attempt. Retrying without classifying the error is one of the most expensive ways of hiding a problem.

You no longer only consume streams: you write them. You know that objectMode turns a pipe of bytes into a pipe of objects, that the highWaterMark starts counting items and that readableObjectMode and writableObjectMode are declared separately because a parser receives text and emits objects. You have mastered Transform with _transform — calling callback() exactly once and signaling errors with callback(error) — and _flush, where everything accumulated is emitted, including that last line with no break that gets silently lost when it is missing. You have built a stateful CSV parser, an aggregating Writable and a filter, and you have chained them into the complete Escena Viva pipe: CSV → objects → filter by venue → aggregation → JSON, with the 1811 sales matching the seed and constant memory all the way through. And you have replaced pipe() with stream.pipeline from node:stream/promises, which propagates errors along the whole chain, destroys every stream on failure, integrates with try/catch and accepts an AbortSignal. The rule is firm: pipe to explain, pipeline to work.

You have also seen how zlib.createGzip slots in as one more stage — and how a pipe can read straight from a .gz without decompressing to disk —, how Readable.from() turns any iterable into a stream, and why an async generator is almost always better than a Transform class: less code, implicit objectMode, natural await and the end of input handled without ceremony. And you know how to cancel a pipe with AbortSignal without leaving descriptors alive.

One piece remains at the bottom of all this. Every time we wrote { encoding: 'utf8' } we were asking for a translation from bytes to text, and every time we left it out — when compressing, when copying — the data traveled as a Buffer. In Buffers and Binary Data we will finally look inside that box: what a Buffer is and why it lives outside the V8 heap, why allocUnsafe can show you somebody else's memory, the encodings and their uses, byte ordering, the classic mistake of slice sharing memory, why an emoji breaks a naive split and how StringDecoder avoids it. And we will apply it to Escena Viva: detecting by its magic numbers whether the poster an organizer uploads really is a PNG, and encoding a ticket's QR code in base64url.