Filtering Stream object in node.js

Question

It seems to me that an elegant way to process certain kinds of data in Node.js would be to chain processing objects, like UNIX pipes.

For example, grep:

function Grep(pattern) {
    ...
}
util.inherits(Grep, stream.Stream);

Grep.prototype.???? = ???????  // What goes here?

grep = new Grep(/foo/);

process.stdin.pipe(grep);
myStream.pipe(process.stdout);

However it's not at all clear to me how the various Stream methods need to be overridden in order for this to work.

How can I create a Stream object that simply copies from its input to its output? Presumably with that answered, more sophisticated filtering streams become trivial.

Update: it feels as if the following should work (expressed in CoffeeScript, so I don't fill this box with JS syntax!):

class Forwarder extends stream.Stream
    write: (chunk, encoding) ->
        @emit 'data', chunk
    end: (chunk, encoding) =>
        if chunk?
            @emit 'data', chunk
        @emit 'end'

fwd = new Forwarder()
fwd.pipe(process.stdout);
process.stdin.pipe(fwd);
process.stdin.resume();

However catting something to this script doesn't output anything. Calling fwd.write() explicitly in the script does cause output on stdout.

loganfsmyth · Accepted Answer · 2013-12-13 23:19:39Z

9

You are so very close.

Because you are using the very low-level stream class, you need to set the stream writable property to make it a writable stream. If you were reading from the stream, you'd need to set the readable property. Also the end event doesn't have any arguments.

class Forwarder extends stream.Stream
  constructor: ->
    @writable = true
  write: (chunk, encoding) ->
    @emit 'data', chunk
  end: ->
    @emit 'end'

fwd = new Forwarder()
fwd.pipe(process.stdout);
process.stdin.pipe(fwd);
process.stdin.resume();

Update

The answer above applied to V1 streams in Node <= 0.8. If you are using > 0.8, Node has added more specific classes that are designed to be extended, so you would use something more like this:

class Forwarder extends stream.Transform
    _transform: (chunk, encoding, callback) ->
      this.push(chunk);
      callback();

Processing chunk and pushing the pieces you actually want.

edited Dec 13, 2013 at 23:19

answered Aug 5, 2011 at 1:13

loganfsmyth

162k31 gold badges349 silver badges259 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

slim Over a year ago

Bingo! I had tried using '@writable: true' as a static member; that doesn't work. But putting it in the constructor does.

Pre101 Over a year ago

Augh! I spent quite a while debugging this because I put 'writeable' instead of 'writable'. Be sure you get the word correct!

loganfsmyth Over a year ago

@BT True enough, I probably just never considered it at the time since the question only used write and end, though pause` on V1 streams was a bit weird. The V2 stream version in my update should work properly anyway, so I'm not sure it's worth worrying about now.

RJo · Accepted Answer · 2015-09-07 10:15:26Z

3

Although the answer that exists is nice, it still requires a bit of digging around on behalf of those looking for answers.

The following code completes the example the OP gave, using the Node 0.10 stream API.

var stream = require('stream')
var util = require('util')

function Grep(pattern) {
  stream.Transform.call(this)

  this.pattern = pattern
}

util.inherits(Grep, stream.Transform)

Grep.prototype._transform = function(chunk, encoding, callback) {
  var string = chunk.toString()

  if (string.match(this.pattern)) {
    this.push(chunk)
  }

  callback()
}

var grep = new Grep(/foo/)
process.stdin.pipe(grep)
grep.pipe(process.stdout)

edited Sep 7, 2015 at 10:15

RJo

16.4k5 gold badges36 silver badges63 bronze badges

answered Sep 20, 2014 at 3:36

user431528

1 Comment

corwin.amber Over a year ago

Looks like this searches for the given pattern on every chunk separately. If one chunk ends in "fo" and the next starts with "o", it would have missed it.

Collectives™ on Stack Overflow

Filtering Stream object in node.js

2 Answers 2

Update

3 Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Update

3 Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related