node/test/simple/test-stream2-basic.js

// Copyright Joyent, Inc. and other Node contributors.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to permit
// persons to whom the Software is furnished to do so, subject to the
// following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE.


var common = require('../common.js');
var R = require('_stream_readable');
var assert = require('assert');

var util = require('util');
var EE = require('events').EventEmitter;

function TestReader(n) {
  R.apply(this);
  this._buffer = new Buffer(n || 100);
  this._buffer.fill('x');
  this._pos = 0;
  this._bufs = 10;
}

util.inherits(TestReader, R);

TestReader.prototype.read = function(n) {
  if (n === 0) return null;
  var max = this._buffer.length - this._pos;
  n = n || max;
  n = Math.max(n, 0);
  var toRead = Math.min(n, max);
  if (toRead === 0) {
    // simulate the read buffer filling up with some more bytes some time
    // in the future.
    setTimeout(function() {
      this._pos = 0;
      this._bufs -= 1;
      if (this._bufs <= 0) {
        // read them all!
        if (!this.ended) {
          this.emit('end');
          this.ended = true;
        }
      } else {
        this.emit('readable');
      }
    }.bind(this), 10);
    return null;
  }

  var ret = this._buffer.slice(this._pos, this._pos + toRead);
  this._pos += toRead;
  return ret;
};

/////

function TestWriter() {
  EE.apply(this);
  this.received = [];
  this.flush = false;
}

util.inherits(TestWriter, EE);

TestWriter.prototype.write = function(c) {
  this.received.push(c.toString());
  this.emit('write', c);
  return true;
};

TestWriter.prototype.end = function(c) {
  if (c) this.write(c);
  this.emit('end', this.received);
};

////////

// tiny node-tap lookalike.
var tests = [];
var count = 0;

function test(name, fn) {
  count++;
  tests.push([name, fn]);
}

function run() {
  var next = tests.shift();
  if (!next)
    return console.error('ok');

  var name = next[0];
  var fn = next[1];
  console.log('# %s', name);
  fn({
    same: assert.deepEqual,
    ok: assert,
    equal: assert.equal,
    end: function () {
      count--;
      run();
    }
  });
}

// ensure all tests have run
process.on("exit", function () {
  assert.equal(count, 0);
});

process.nextTick(run);


test('a most basic test', function(t) {
  var r = new TestReader(20);

  var reads = [];
  var expect = [ 'x',
                 'xx',
                 'xxx',
                 'xxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxxxxx',
                 'xxxxxxxxx',
                 'xxx',
                 'xxxxxxxxxxxx',
                 'xxxxxxxx',
                 'xxxxxxxxxxxxxxx',
                 'xxxxx',
                 'xxxxxxxxxxxxxxxxxx',
                 'xx',
                 'xxxxxxxxxxxxxxxxxxxx',
                 'xxxxxxxxxxxxxxxxxxxx',
                 'xxxxxxxxxxxxxxxxxxxx',
                 'xxxxxxxxxxxxxxxxxxxx',
                 'xxxxxxxxxxxxxxxxxxxx' ];

  r.on('end', function() {
    t.same(reads, expect);
    t.end();
  });

  var readSize = 1;
  function flow() {
    var res;
    while (null !== (res = r.read(readSize++))) {
      reads.push(res.toString());
    }
    r.once('readable', flow);
  }

  flow();
});

test('pipe', function(t) {
  var r = new TestReader(5);

  var expect = [ 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx' ]

  var w = new TestWriter;
  var flush = true;

  w.on('end', function(received) {
    t.same(received, expect);
    t.end();
  });

  r.pipe(w);
});



[1,2,3,4,5,6,7,8,9].forEach(function(SPLIT) {
  test('unpipe', function(t) {
    var r = new TestReader(5);

    // unpipe after 3 writes, then write to another stream instead.
    var expect = [ 'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx' ];
    expect = [ expect.slice(0, SPLIT), expect.slice(SPLIT) ];

    var w = [ new TestWriter(), new TestWriter() ];

    var writes = SPLIT;
    w[0].on('write', function() {
      if (--writes === 0) {
        r.unpipe();
        t.equal(r._readableState.pipes, null);
        w[0].end();
        r.pipe(w[1]);
        t.equal(r._readableState.pipes, w[1]);
      }
    });

    var ended = 0;

    var ended0 = false;
    var ended1 = false;
    w[0].on('end', function(results) {
      t.equal(ended0, false);
      ended0 = true;
      ended++;
      t.same(results, expect[0]);
    });

    w[1].on('end', function(results) {
      t.equal(ended1, false);
      ended1 = true;
      ended++;
      t.equal(ended, 2);
      t.same(results, expect[1]);
      t.end();
    });

    r.pipe(w[0]);
  });
});


// both writers should get the same exact data.
test('multipipe', function(t) {
  var r = new TestReader(5);
  var w = [ new TestWriter, new TestWriter ];

  var expect = [ 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx',
                 'xxxxx' ];

  var c = 2;
  w[0].on('end', function(received) {
    t.same(received, expect, 'first');
    if (--c === 0) t.end();
  });
  w[1].on('end', function(received) {
    t.same(received, expect, 'second');
    if (--c === 0) t.end();
  });

  r.pipe(w[0]);
  r.pipe(w[1]);
});


[1,2,3,4,5,6,7,8,9].forEach(function(SPLIT) {
  test('multi-unpipe', function(t) {
    var r = new TestReader(5);

    // unpipe after 3 writes, then write to another stream instead.
    var expect = [ 'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx' ];
    expect = [ expect.slice(0, SPLIT), expect.slice(SPLIT) ];

    var w = [ new TestWriter(), new TestWriter(), new TestWriter() ];

    var writes = SPLIT;
    w[0].on('write', function() {
      if (--writes === 0) {
        r.unpipe();
        w[0].end();
        r.pipe(w[1]);
      }
    });

    var ended = 0;

    w[0].on('end', function(results) {
      ended++;
      t.same(results, expect[0]);
    });

    w[1].on('end', function(results) {
      ended++;
      t.equal(ended, 2);
      t.same(results, expect[1]);
      t.end();
    });

    r.pipe(w[0]);
    r.pipe(w[2]);
  });
});

test('back pressure respected', function (t) {
  function noop() {}

  var r = new R({ objectMode: true });
  r._read = noop;
  var counter = 0;
  r.push(["one"]);
  r.push(["two"]);
  r.push(["three"]);
  r.push(["four"]);
  r.push(null);

  var w1 = new R();
  w1.write = function (chunk) {
    assert.equal(chunk[0], "one");
    w1.emit("close");
    process.nextTick(function () {
      r.pipe(w2);
      r.pipe(w3);
    })
  };
  w1.end = noop;

  r.pipe(w1);

  var expected = ["two", "two", "three", "three", "four", "four"];

  var w2 = new R();
  w2.write = function (chunk) {
    assert.equal(chunk[0], expected.shift());
    assert.equal(counter, 0);

    counter++;

    if (chunk[0] === "four") {
      return true;
    }

    setTimeout(function () {
      counter--;
      w2.emit("drain");
    }, 10);

    return false;
  }
  w2.end = noop;

  var w3 = new R();
  w3.write = function (chunk) {
    assert.equal(chunk[0], expected.shift());
    assert.equal(counter, 1);

    counter++;

    if (chunk[0] === "four") {
      return true;
    }

    setTimeout(function () {
      counter--;
      w3.emit("drain");
    }, 50);

    return false;
  };
  w3.end = function () {
    assert.equal(counter, 2);
    assert.equal(expected.length, 0);
    t.end();
  };
});

test('read(0) for ended streams', function (t) {
  var r = new R();
  var written = false;
  var ended = false;
  r._read = function (n) {};

  r.push(new Buffer("foo"));
  r.push(null);

  var v = r.read(0);

  assert.equal(v, null);

  var w = new R();

  w.write = function (buffer) {
    written = true;
    assert.equal(ended, false);
    assert.equal(buffer.toString(), "foo")
  };

  w.end = function () {
    ended = true;
    assert.equal(written, true);
    t.end();
  };

  r.pipe(w);
})

test('sync _read ending', function (t) {
  var r = new R();
  var called = false;
  r._read = function (n) {
    r.push(null);
  };

  r.once('end', function () {
    called = true;
  })

  r.read();

  process.nextTick(function () {
    assert.equal(called, true);
    t.end();
  })
});

test('adding readable triggers data flow', function(t) {
  var r = new R({ highWaterMark: 5 });
  var onReadable = false;
  var readCalled = 0;

  r._read = function(n) {
    if (readCalled++ === 2)
      r.push(null);
    else
      r.push(new Buffer('asdf'));
  };

  var called = false;
  r.on('readable', function() {
    onReadable = true;
    r.read();
  });

  r.on('end', function() {
    t.equal(readCalled, 3);
    t.ok(onReadable);
    t.end();
  });
});