Implementing Flow Control manually

In this lesson, you will learn how to manually throttle data intake by listening to your writable destination’s readiness, ensuring you never overwhelm your system's memory.

The Problem of Over-production

When you pipe() streams in Node.js, the runtime handles flow control for you. However, when you manually read() from a source and write() to a destination, you risk "flooding" the destination. If your source (like a high-speed disk) is faster than your destination (like a slow network socket), the destination’s internal buffer will keep growing until it exhausts your system's memory.

Implementing Manual Throttling

To take control, you must watch the boolean return value of the writable.write(chunk) method.

1. If write() returns true: The destination buffer still has space. You can continue reading and writing immediately.
2. If write() returns false: The destination is full (it has hit its High Water Mark). You must stop calling read() and wait.

You "pause" the readable stream using readable.pause(). To resume, you wait for the writable stream to emit the 'drain' event, which signals that its internal buffer has emptied and it is ready for more data.

Concrete Example

Here is how you coordinate the two streams manually:

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);

  if (!canContinue) {
    // Stop reading until the destination is ready
    readable.pause();

    writable.once('drain', () => {
      // Resume reading once the buffer clears
      readable.resume();
    });
  }
});

By adding this check, you create a feedback loop. The writable stream effectively "pushes back" against the readable stream. This prevents the memory spikes that occur when a producer ignores the limitations of the consumer. Even if you are dealing with millions of records, your memory usage will remain constant because you only hold one chunk in transit at a time.

Key Takeaway: Always check the return value of writable.write() and use the 'drain' event to gate your readable.resume() calls to prevent memory overflow.