1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-12-27 01:29:14 -05:00
denoland-deno/cli/tests/node_compat/test/parallel/test-stream2-basic.js
2023-06-09 01:13:20 +02:00

452 lines
10 KiB
JavaScript

// deno-fmt-ignore-file
// deno-lint-ignore-file
// Copyright Joyent and Node contributors. All rights reserved. MIT license.
// Taken from Node 18.12.1
// This file is automatically generated by `tools/node_compat/setup.ts`. Do not modify this file manually.
// 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 { Readable: R, Writable: W } = require('stream');
const assert = require('assert');
const EE = require('events').EventEmitter;
class TestReader extends R {
constructor(n) {
super();
this._buffer = Buffer.alloc(n || 100, 'x');
this._pos = 0;
this._bufs = 10;
}
_read(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(() => {
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);
}
}, 10);
return;
}
const ret = this._buffer.slice(this._pos, this._pos + toRead);
this._pos += toRead;
this.push(ret);
}
}
class TestWriter extends EE {
constructor() {
super();
this.received = [];
this.flush = false;
}
write(c) {
this.received.push(c.toString());
this.emit('write', c);
return true;
}
end(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.deepStrictEqual(r._readableState.pipes, []);
w[0].end();
r.pipe(w[1]);
assert.deepStrictEqual(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);
}));
w[1].on('end', common.mustCall(function(received) {
assert.deepStrictEqual(received, expect);
}));
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);
}
{
// Verify readableEncoding property
assert(Object.hasOwn(R.prototype, 'readableEncoding'));
const r = new R({ encoding: 'utf8' });
assert.strictEqual(r.readableEncoding, 'utf8');
}
{
// Verify readableObjectMode property
assert(Object.hasOwn(R.prototype, 'readableObjectMode'));
const r = new R({ objectMode: true });
assert.strictEqual(r.readableObjectMode, true);
}
{
// Verify writableObjectMode property
assert(Object.hasOwn(W.prototype, 'writableObjectMode'));
const w = new W({ objectMode: true });
assert.strictEqual(w.writableObjectMode, true);
}