text: // 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.
  
  'use strict';
  
  const common = require('../common');
  const R = require('_stream_readable');
  const assert = require('assert');
  
  const util = require('util');
  const EE = require('events').EventEmitter;
  
  function TestReader(n) {
    R.apply(this);
    this._buffer = Buffer.alloc(n || 100, 'x');
    this._pos = 0;
    this._bufs = 10;
  }
  
  util.inherits(TestReader, R);
  
  TestReader.prototype._read = function(n) {
    const max = this._buffer.length - this._pos;
    n = Math.max(n, 0);
    const 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.push(null);
        } else {
          // now we have more.
          // kinda cheating by calling _read, but whatever,
          // it's just fake anyway.
          this._read(n);
        }
      }.bind(this), 10);
      return;
    }
  
    const ret = this._buffer.slice(this._pos, this._pos + toRead);
    this._pos += toRead;
    this.push(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);
  };
  
  {
    // Test basic functionality
    const r = new TestReader(20);
  
    const reads = [];
    const expect = [ 'x',
                     'xx',
                     'xxx',
                     'xxxx',
                     'xxxxx',
                     'xxxxxxxxx',
                     'xxxxxxxxxx',
                     'xxxxxxxxxxxx',
                     'xxxxxxxxxxxxx',
                     'xxxxxxxxxxxxxxx',
                     'xxxxxxxxxxxxxxxxx',
                     'xxxxxxxxxxxxxxxxxxx',
                     'xxxxxxxxxxxxxxxxxxxxx',
                     'xxxxxxxxxxxxxxxxxxxxxxx',
                     'xxxxxxxxxxxxxxxxxxxxxxxxx',
                     'xxxxxxxxxxxxxxxxxxxxx' ];
  
    r.on('end', common.mustCall(function() {
      assert.deepStrictEqual(reads, expect);
    }));
  
    let readSize = 1;
    function flow() {
      let res;
      while (null !== (res = r.read(readSize++))) {
        reads.push(res.toString());
      }
      r.once('readable', flow);
    }
  
    flow();
  }
  
  {
    // Verify pipe
    const r = new TestReader(5);
  
    const expect = [ 'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx' ];
  
    const w = new TestWriter();
  
    w.on('end', common.mustCall(function(received) {
      assert.deepStrictEqual(received, expect);
    }));
  
    r.pipe(w);
  }
  
  
  [1, 2, 3, 4, 5, 6, 7, 8, 9].forEach(function(SPLIT) {
    // Verify unpipe
    const r = new TestReader(5);
  
    // unpipe after 3 writes, then write to another stream instead.
    let expect = [ 'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx' ];
    expect = [ expect.slice(0, SPLIT), expect.slice(SPLIT) ];
  
    const w = [ new TestWriter(), new TestWriter() ];
  
    let writes = SPLIT;
    w[0].on('write', function() {
      if (--writes === 0) {
        r.unpipe();
        assert.strictEqual(r._readableState.pipes, null);
        w[0].end();
        r.pipe(w[1]);
        assert.strictEqual(r._readableState.pipes, w[1]);
      }
    });
  
    let ended = 0;
  
    w[0].on('end', common.mustCall(function(results) {
      ended++;
      assert.strictEqual(ended, 1);
      assert.deepStrictEqual(results, expect[0]);
    }));
  
    w[1].on('end', common.mustCall(function(results) {
      ended++;
      assert.strictEqual(ended, 2);
      assert.deepStrictEqual(results, expect[1]);
    }));
  
    r.pipe(w[0]);
  });
  
  
  {
    // Verify both writers get the same data when piping to destinations
    const r = new TestReader(5);
    const w = [ new TestWriter(), new TestWriter() ];
  
    const expect = [ 'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx',
                     'xxxxx' ];
  
    w[0].on('end', common.mustCall(function(received) {
      assert.deepStrictEqual(received, expect, 'first');
    }));
    w[1].on('end', common.mustCall(function(received) {
      assert.deepStrictEqual(received, expect, 'second');
    }));
  
    r.pipe(w[0]);
    r.pipe(w[1]);
  }
  
  
  [1, 2, 3, 4, 5, 6, 7, 8, 9].forEach(function(SPLIT) {
    // Verify multi-unpipe
    const r = new TestReader(5);
  
    // unpipe after 3 writes, then write to another stream instead.
    let expect = [ 'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx',
                   'xxxxx' ];
    expect = [ expect.slice(0, SPLIT), expect.slice(SPLIT) ];
  
    const w = [ new TestWriter(), new TestWriter(), new TestWriter() ];
  
    let writes = SPLIT;
    w[0].on('write', function() {
      if (--writes === 0) {
        r.unpipe();
        w[0].end();
        r.pipe(w[1]);
      }
    });
  
    let ended = 0;
  
    w[0].on('end', common.mustCall(function(results) {
      ended++;
      assert.strictEqual(ended, 1);
      assert.deepStrictEqual(results, expect[0]);
    }));
  
    w[1].on('end', common.mustCall(function(results) {
      ended++;
      assert.strictEqual(ended, 2);
      assert.deepStrictEqual(results, expect[1]);
    }));
  
    r.pipe(w[0]);
    r.pipe(w[2]);
  });
  
  {
    // Verify that back pressure is respected
    const r = new R({ objectMode: true });
    r._read = common.mustNotCall();
    let counter = 0;
    r.push(['one']);
    r.push(['two']);
    r.push(['three']);
    r.push(['four']);
    r.push(null);
  
    const w1 = new R();
    w1.write = function(chunk) {
      assert.strictEqual(chunk[0], 'one');
      w1.emit('close');
      process.nextTick(function() {
        r.pipe(w2);
        r.pipe(w3);
      });
    };
    w1.end = common.mustNotCall();
  
    r.pipe(w1);
  
    const expected = ['two', 'two', 'three', 'three', 'four', 'four'];
  
    const w2 = new R();
    w2.write = function(chunk) {
      assert.strictEqual(chunk[0], expected.shift());
      assert.strictEqual(counter, 0);
  
      counter++;
  
      if (chunk[0] === 'four') {
        return true;
      }
  
      setTimeout(function() {
        counter--;
        w2.emit('drain');
      }, 10);
  
      return false;
    };
    w2.end = common.mustCall();
  
    const w3 = new R();
    w3.write = function(chunk) {
      assert.strictEqual(chunk[0], expected.shift());
      assert.strictEqual(counter, 1);
  
      counter++;
  
      if (chunk[0] === 'four') {
        return true;
      }
  
      setTimeout(function() {
        counter--;
        w3.emit('drain');
      }, 50);
  
      return false;
    };
    w3.end = common.mustCall(function() {
      assert.strictEqual(counter, 2);
      assert.strictEqual(expected.length, 0);
    });
  }
  
  {
    // Verify read(0) behavior for ended streams
    const r = new R();
    let written = false;
    let ended = false;
    r._read = common.mustNotCall();
  
    r.push(Buffer.from('foo'));
    r.push(null);
  
    const v = r.read(0);
  
    assert.strictEqual(v, null);
  
    const w = new R();
  
    w.write = function(buffer) {
      written = true;
      assert.strictEqual(ended, false);
      assert.strictEqual(buffer.toString(), 'foo');
    };
  
    w.end = common.mustCall(function() {
      ended = true;
      assert.strictEqual(written, true);
    });
  
    r.pipe(w);
  }
  
  {
    // Verify synchronous _read ending
    const r = new R();
    let called = false;
    r._read = function(n) {
      r.push(null);
    };
  
    r.once('end', function() {
      // Verify that this is called before the next tick
      called = true;
    });
  
    r.read();
  
    process.nextTick(function() {
      assert.strictEqual(called, true);
    });
  }
  
  {
    // Verify that adding readable listeners trigger data flow
    const r = new R({ highWaterMark: 5 });
    let onReadable = false;
    let readCalled = 0;
  
    r._read = function(n) {
      if (readCalled++ === 2)
        r.push(null);
      else
        r.push(Buffer.from('asdf'));
    };
  
    r.on('readable', function() {
      onReadable = true;
      r.read();
    });
  
    r.on('end', common.mustCall(function() {
      assert.strictEqual(readCalled, 3);
      assert.ok(onReadable);
    }));
  }
  
  {
    // Verify that streams are chainable
    const r = new R();
    r._read = common.mustCall();
    const r2 = r.setEncoding('utf8').pause().resume().pause();
    assert.strictEqual(r, r2);
  }
  