_read() is not implemented on Readable stream

Question

This question is how to really implement the read method of a readable stream.

I have this implementation of a Readable stream:

import {Readable} from "stream";
this.readableStream = new Readable();

I am getting this error

events.js:136 throw er; // Unhandled 'error' event ^

Error [ERR_STREAM_READ_NOT_IMPLEMENTED]: _read() is not implemented at Readable._read (_stream_readable.js:554:22) at Readable.read (_stream_readable.js:445:10) at resume_ (_stream_readable.js:825:12) at _combinedTickCallback (internal/process/next_tick.js:138:11) at process._tickCallback (internal/process/next_tick.js:180:9) at Function.Module.runMain (module.js:684:11) at startup (bootstrap_node.js:191:16) at bootstrap_node.js:613:3

The reason the error occurs is obvious, we need to do this:

  this.readableStream = new Readable({
      read(size) {
        return true;
      }
    });

I don't really understand how to implement the read method though.

The only thing that works is just calling

this.readableStream.push('some string or buffer');

if I try to do something like this:

   this.readableStream = new Readable({
          read(size) {
            this.push('foo');   // call push here!
            return true;
          }
     });

then nothing happens - nothing comes out of the readable!

Furthermore, these articles says you don't need to implement the read method:

https://github.com/substack/stream-handbook#creating-a-readable-stream

https://medium.freecodecamp.org/node-js-streams-everything-you-need-to-know-c9141306be93

My question is - why does calling push inside the read method do nothing? The only thing that works for me is just calling readable.push() elsewhere.

Answer 1

why does calling push inside the read method do nothing? The only thing that works for me is just calling readable.push() elsewhere.

I think it's because you are not consuming it, you need to pipe it to an writable stream (e.g. stdout) or just consume it through a data event:

const { Readable } = require("stream");

let count = 0;
const readableStream = new Readable({
    read(size) {
        this.push('foo');
        if (count === 5) this.push(null);
        count++;
    }
});

// piping
readableStream.pipe(process.stdout)

// through the data event
readableStream.on('data', (chunk) => {
  console.log(chunk.toString());
});

Both of them should print 5 times foo (they are slightly different though). Which one you should use depends on what you are trying to accomplish.

Furthermore, these articles says you don't need to implement the read method:

You might not need it, this should work:

const { Readable } = require("stream");

const readableStream = new Readable();

for (let i = 0; i <= 5; i++) {
    readableStream.push('foo');
}
readableStream.push(null);

readableStream.pipe(process.stdout)

In this case you can't capture it through the data event. Also, this way is not very useful and not efficient I'd say, we are just pushing all the data in the stream at once (if it's large everything is going to be in memory), and then consuming it.

Answer 2

From documentation:

readable._read:

"When readable._read() is called, if data is available from the resource, the implementation should begin pushing that data into the read queue using the this.push(dataChunk) method. link"

readable.push:

"The readable.push() method is intended be called only by Readable implementers, and only from within the readable._read() method. link"