A little confused on this stream readable syntax

Question

A little confused on this stream readable syntax

I read through the stream handbook and they had this example:

var Readable = require('stream').Readable;

var rs = new Readable;
rs.push('beep ');
rs.push('boop\n');
rs.push(null);

rs.pipe(process.stdout);

It looks like exactly what I needed (push objects into a readable stream and pipe it to a writable.)

So this is what I came up with

var Readable = require('stream').Readable;
var rs = new Readable;

app.get('/:foo', function(req, res) {
oboe(fs.createReadStream("/file"))
    .node( "{}", function(data) { 
          rs.push(data) 
     })
rs.pipe(res);

I am using oboe to listen on the fsStream for objects and modifying them. Currently I am pushing the newly found objects into an array and then on "done" stream event I would tell express to res.json the newly created array. This got too memory heavy and i was wondering if I could just empty the stream as the objects are found and modified, push it into a readable and as the readable stream gets data, it would pipe it out to res but still maintain one object.

However I was receiving this error

Error: not implemented at Readable._read

Does this mean the stream handbook is out of date?

node.js
express
stream

Answer 1

You're not supposed to use Readable directly. You're supposed to subclass it and write your own implementation of _read; that's why node is complaining that Readable._read is not implemented. Also, node streams want Strings or Buffers, not JSON Objects, so writing data from oboe's node event to a stream will probably cause problems.

I think there's an easier solution than using a Readable stream here. It sounds like you want to take the JSON Object that oboe is giving you, change it, and stream it out express's res object. res itself is a Writable stream so, instead of going through an intermediary stream, you can simply write to it as many times as you'd like until you're done. Be aware though that you'll also have to call end() on the response stream (res) when when oboe is finished.

app.get('/:foo', function(req, res) {
    oboe(fs.createReadStream("/file"))
        .node( "{}", function(data) { 
            // ... transform data however you need ...
            res.write(JSON.stringify(data)); // Note that data is JSON.stringified here since res.write wants a String or a Buffer, not an Object
        })
        .done(function() {
            res.end();
        })
})

Note: I'm assuming you have something on the other end of the response that can handle getting JSON Objects streamed in like that. You'll probably need something like oboe on that end too, if you don't have it already.