Stream Example

example.js

var stream = require('stream');
var messageStream = require('./message-stream');

var jsonify = new stream.Transform({
  writableObjectMode: true,
  transform: function(chunk, enc, cb) {
    this.push(JSON.stringify(chunk, null, 2) + '\n');
    cb();
  }
});

messageStream({interval: 50}).pipe(jsonify).pipe(process.stdout);

message-stream.js

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

var objectAssign = require('object-assign')

util.inherits(MessageStream, stream.Readable);
function MessageStream(options) {
  if (!(this instanceof MessageStream)) return new MessageStream(options);

  this.options = objectAssign({
    interval: 2000,
    string: 'Hello World'
  }, options, {
    objectMode: true,
    highWaterMark: 0
  });

  stream.Readable.call(this, this.options);
}

MessageStream.prototype._read = function() {
  var self = this;

  setTimeout(function() {
    self.push({message: self.options.string});
  }, this.options.interval);
}

module.exports = MessageStream;