ordered-read-streams/test/main.js

var should = require('should');
var through = require('through2');
var OrderedStreams = require('../');

describe('ordered-read-streams', function () {
  it('should end if no streams are given', function (done) {
    var streams = new OrderedStreams();
    streams.on('data', function () {
      done('error');
    });
    streams.on('end', done);
  });

  it('should throw error if one or more streams are not readable', function (done) {
    var writable = { readable: false };

    try {
      new OrderedStreams(writable);
    } catch (e) {
      e.message.should.equal('All input streams must be readable');
      done();
    }
  });

  it('should emit data from all streams', function(done) {
    var s1 = through.obj(function (data, enc, next) {
      this.push(data);
      next();
    });
    var s2 = through.obj(function (data, enc, next) {
      this.push(data);
      next();
    });
    var s3 = through.obj(function (data, enc, next) {
      this.push(data);
      next();
    });

    var streams = new OrderedStreams([s1, s2, s3]);
    var results = [];
    streams.on('data', function (data) {
      results.push(data);
    });
    streams.on('end', function () {
      results.length.should.be.exactly(3);
      results[0].should.equal('stream 1');
      results[1].should.equal('stream 2');
      results[2].should.equal('stream 3');
      done();
    });

    s1.write('stream 1');
    s1.end();

    s2.write('stream 2');
    s2.end();

    s3.write('stream 3');
    s3.end();
  });

  it('should emit all data event from each stream', function (done) {
    var s = through.obj(function (data, enc, next) {
      this.push(data);
      next();
    });

    var streams = new OrderedStreams(s);
    var results = [];
    streams.on('data', function (data) {
      results.push(data);
    });
    streams.on('end', function () {
      results.length.should.be.exactly(3);
      done();
    });

    s.write('data1');
    s.write('data2');
    s.write('data3');
    s.end();
  });

  it('should preserve streams order', function(done) {
    var s1 = through.obj(function (data, enc, next) {
      var self = this;
      setTimeout(function () {
        self.push(data);
        next();
      }, 200);
    });
    var s2 = through.obj(function (data, enc, next) {
      var self = this;
      setTimeout(function () {
        self.push(data);
        next();
      }, 30);
    });
    var s3 = through.obj(function (data, enc, next) {
      var self = this;
      setTimeout(function () {
        self.push(data);
        next();
      }, 100);
    });

    var streams = new OrderedStreams([s1, s2, s3]);
    var results = [];
    streams.on('data', function (data) {
      results.push(data);
    });
    streams.on('end', function () {
      results.length.should.equal(3);
      results[0].should.equal('stream 1');
      results[1].should.equal('stream 2');
      results[2].should.equal('stream 3');
      done();
    });

    s1.write('stream 1');
    s1.end();

    s2.write('stream 2');
    s2.end();

    s3.write('stream 3');
    s3.end();
  });

  it('should emit stream errors downstream', function (done) {
    var s = through.obj(function (data, enc, next) {
      this.emit('error', new Error('stahp!'));
      next();
    });
    var s2 = through.obj(function (data, enc, next) {
      this.push(data);
      next();
    });

    var errMsg;
    var streamData;
    var streams = new OrderedStreams([s, s2]);
    streams.on('data', function (data) {
      streamData = data;
    });
    streams.on('error', function (err) {
      errMsg = err.message;
    });
    streams.on('end', function () {
      errMsg.should.equal('stahp!');
      streamData.should.equal('Im ok!');
      done();
    });

    s.write('go');
    s.end();
    s2.write('Im ok!');
    s2.end();
  });
});