2 · Data and boundaries · 30 MIN

Streams and backpressure

A producer should not outrun its downstream consumer indefinitely.

pipeline connects streams, propagates errors and finishes when the whole chain completes. An asynchronous generator can transform one chunk at a time without collecting the entire input. The writable callback signals that a chunk has been processed; forgetting it stalls the pipeline. Chunk boundaries are transport boundaries, not guaranteed application records. For JSON Lines or CSV, retain partial records between chunks and enforce a maximum record size instead of splitting each chunk as if it were a complete file.

Split a record across two chunks and keep it intact.

Read the example

import {Readable,Writable} from 'node:stream';
import {pipeline} from 'node:stream/promises';
const sink=new Writable({write(chunk,encoding,done){process.stdout.write(chunk);done()}});
await pipeline(
  Readable.from(['one\n','two\n']),
  async function* (source){for await(const chunk of source)yield String(chunk).toUpperCase()},
  sink
);

ONE
TWO

Your challenge

Build a bounded line splitter that handles a newline split across chunks and rejects records above your chosen size limit.

Solution cost: O(b) transformation for b input bytes. time · Bounded buffers plus the maximum partial record. space

Common trap

A stream chunk is not necessarily a whole line, JSON object or Unicode character.

import {Readable,Writable} from 'node:stream';
import {pipeline} from 'node:stream/promises';
const sink=new Writable({write(chunk,encoding,done){process.stdout.write(chunk);done()}});
await pipeline(
  Readable.from(['one\n','two\n']),
  async function* (source){for await(const chunk of source)yield String(chunk).toUpperCase()},
  sink
);

Further reading: Official documentation