Node.js Promise Stream

promise-stream.js

'use strict';
var stream = require('stream');
var util = require('util');
var Promise = require('bluebird');

function PStream() {
  stream.PassThrough.call(this);
  var _promise = new Promise(function(resolve, reject) {
    this.on('end', resolve);
    this.on('error', reject);
  }.bind(this));
  this.then = _promise.then.bind(_promise);
}

module.exports = PStream;

util.inherits(PStream, stream.PassThrough);

PStream.through = function(readable) {
  var pstream = new PStream();
  readable.on('data', function(chunk) {
    pstream.write(chunk);
  });
  readable.on('end', function(chunk) {
    pstream.end(chunk);
  });
  readable.on('error', pstream.emit.bind(pstream, 'error'));
  return pstream;
};

if (require.main === module) {
  var fs = require('fs');
  var assert = require('assert');
  var tests = [];
  var test = function(fn) {
    tests.push(fn());
  };

  console.log('Running self tests...');

  test(function() {
    var content1 = '';
    var content2 = '';
    var fsStream, pStream;

    fsStream = fs.createReadStream(__filename);
    pStream = PStream.through(fsStream);

    fsStream.on('data', function(chunk) {
      content1 += chunk;
    });
    pStream.on('data', function(chunk) {
      content2 += chunk;
    });

    return pStream.then(function() {
      assert.equal(content1, content2);
    }, function(err) {
      throw new Error('Normal read stream rejected.');
    });
  });

  test(function() {
    var fsStream = fs.createReadStream('/');
    var pStream = PStream.through(fsStream);

    return pStream.then(function() {
      throw new Error('Expected error not propagated.');
    }, function(err) {
      assert.ok(err);
    });
  });

  Promise.all(tests).then(function() {
    console.log('All ' + tests.length + ' tests passed.');
  }, function(err) {
    console.error('Tests failed.', err);
    process.exit(1);
  });
}