You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
210 lines
4.1 KiB
210 lines
4.1 KiB
2 years ago
|
var test = require('tape');
|
||
|
var exhaust = require('./index.js');
|
||
|
var Stream = require('stream');
|
||
|
var Readable = Stream.Readable;
|
||
|
var Writable = Stream.Writable;
|
||
|
var Duplex = Stream.Duplex;
|
||
|
var through = require('through2');
|
||
|
var S2Readable = require('readable-stream').Readable;
|
||
|
|
||
|
test('it should cause a Readable stream to complete if it\'s not piped anywhere', function(assert) {
|
||
|
var rs = new Readable({highWaterMark: 2});
|
||
|
var a = 0;
|
||
|
var ended = false;
|
||
|
rs._read = function() {
|
||
|
if (a++ < 100) {
|
||
|
rs.push(a + "");
|
||
|
} else {
|
||
|
ended = true;
|
||
|
rs.push(null);
|
||
|
}
|
||
|
};
|
||
|
|
||
|
rs.on("end", function() {
|
||
|
assert.ok(a > 99, 'a should be > 99');
|
||
|
assert.ok(ended, 'it should end');
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
exhaust(rs);
|
||
|
});
|
||
|
|
||
|
test('should work with Readable streams in objectMode', function(assert) {
|
||
|
var rs = new Readable({highWaterMark: 2, objectMode: true});
|
||
|
var a = 0;
|
||
|
var ended = false;
|
||
|
rs._read = function() {
|
||
|
if (a++ < 100) {
|
||
|
rs.push(a);
|
||
|
} else {
|
||
|
ended = true;
|
||
|
rs.push(null);
|
||
|
}
|
||
|
};
|
||
|
|
||
|
rs.on("end", function() {
|
||
|
assert.ok(a > 99, 'a > 99');
|
||
|
assert.ok(ended, 'ended is true');
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
exhaust(rs);
|
||
|
});
|
||
|
|
||
|
test('should not interfere with a Readable stream that is piped somewhere', function(assert) {
|
||
|
var rs = new Readable({highWaterMark: 2});
|
||
|
var a = 0;
|
||
|
var ended = false;
|
||
|
rs._read = function() {
|
||
|
if (a++ < 100) {
|
||
|
rs.push(".");
|
||
|
} else {
|
||
|
ended = true;
|
||
|
rs.push(null);
|
||
|
}
|
||
|
};
|
||
|
|
||
|
var sizeRead = 0;
|
||
|
var ws = new Writable({highWaterMark: 2});
|
||
|
ws._write = function(chunk, enc, next) {
|
||
|
sizeRead += chunk.length;
|
||
|
next();
|
||
|
}
|
||
|
|
||
|
ws.on("finish", function() {
|
||
|
assert.ok(a > 99, 'a > 99');
|
||
|
assert.ok(ended, 'ended is true');
|
||
|
assert.equal(sizeRead, 100, 'sizeRead === 100');
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
rs.pipe(ws);
|
||
|
|
||
|
exhaust(rs);
|
||
|
});
|
||
|
|
||
|
test('should not interfere with a Writable stream', function(assert) {
|
||
|
var rs = new Readable({highWaterMark: 2});
|
||
|
var a = 0;
|
||
|
var ended = false;
|
||
|
rs._read = function() {
|
||
|
if (a++ < 100) {
|
||
|
rs.push(".");
|
||
|
} else {
|
||
|
ended = true;
|
||
|
rs.push(null);
|
||
|
}
|
||
|
};
|
||
|
|
||
|
var sizeRead = 0;
|
||
|
var ws = new Writable({highWaterMark: 2});
|
||
|
ws._write = function(chunk, enc, next) {
|
||
|
sizeRead += chunk.length;
|
||
|
next();
|
||
|
}
|
||
|
|
||
|
ws.on("finish", function() {
|
||
|
assert.ok(a > 99, 'a > 99');
|
||
|
assert.ok(ended, 'ended is true');
|
||
|
assert.equal(sizeRead, 100, 'sizeRead === 100');
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
rs.pipe(ws);
|
||
|
|
||
|
exhaust(ws);
|
||
|
});
|
||
|
|
||
|
test('should handle a Transform stream', function(assert) {
|
||
|
var rs = new Readable({highWaterMark: 2});
|
||
|
var a = 0;
|
||
|
var ended = false;
|
||
|
rs._read = function() {
|
||
|
if (a++ < 100) {
|
||
|
rs.push(".");
|
||
|
} else {
|
||
|
ended = true;
|
||
|
rs.push(null);
|
||
|
}
|
||
|
};
|
||
|
|
||
|
var sizeRead = 0;
|
||
|
var flushed = false;
|
||
|
var ts = through({highWaterMark: 2}, function(chunk, enc, cb) {
|
||
|
sizeRead += chunk.length;
|
||
|
this.push(chunk);
|
||
|
cb();
|
||
|
}, function(cb) {
|
||
|
flushed = true;
|
||
|
cb();
|
||
|
});
|
||
|
|
||
|
ts.on("end", function() {
|
||
|
assert.ok(a > 99, 'a > 99');
|
||
|
assert.ok(ended, 'ended is true');
|
||
|
assert.equal(sizeRead, 100, 'sizeRead === 100');
|
||
|
assert.ok(flushed, 'flushed is true');
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
rs.pipe(ts);
|
||
|
|
||
|
exhaust(ts);
|
||
|
});
|
||
|
|
||
|
test('should handle a classic stream', function(assert) {
|
||
|
var rs = new Stream();
|
||
|
var ended = false;
|
||
|
var i;
|
||
|
|
||
|
rs.on("end", function() {
|
||
|
assert.ok(ended, 'ended is true');
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
exhaust(rs);
|
||
|
|
||
|
for (i = 0; i < 100; i++) {
|
||
|
rs.emit("data", i);
|
||
|
}
|
||
|
ended = true;
|
||
|
rs.emit("end");
|
||
|
});
|
||
|
|
||
|
test('should not modify .pipe', function(assert) {
|
||
|
var stream = new S2Readable;
|
||
|
|
||
|
var pipe = stream.pipe;
|
||
|
|
||
|
stream._read = function() {
|
||
|
stream.push('ending');
|
||
|
stream.push(null);
|
||
|
};
|
||
|
|
||
|
exhaust(stream);
|
||
|
|
||
|
assert.equal(stream.pipe, pipe);
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
test('does not error on no resume but readable set to true', function(assert) {
|
||
|
var rs = new Stream();
|
||
|
rs.readable = true;
|
||
|
|
||
|
var ended = false;
|
||
|
var i;
|
||
|
|
||
|
rs.on("end", function() {
|
||
|
assert.ok(ended, 'ended is true');
|
||
|
assert.end();
|
||
|
});
|
||
|
|
||
|
exhaust(rs);
|
||
|
|
||
|
for (i = 0; i < 100; i++) {
|
||
|
rs.emit("data", i);
|
||
|
}
|
||
|
ended = true;
|
||
|
rs.emit("end");
|
||
|
});
|