375 lines
11 KiB
JavaScript
375 lines
11 KiB
JavaScript
"use strict";
|
|
|
|
require("leaked-handles");
|
|
|
|
var _fs = _interopRequireDefault(require("fs"));
|
|
|
|
var _stream = _interopRequireDefault(require("stream"));
|
|
|
|
var _tap = _interopRequireDefault(require("tap"));
|
|
|
|
var _ = _interopRequireDefault(require("."));
|
|
|
|
function _interopRequireDefault(obj) {
|
|
return obj && obj.__esModule ? obj : { default: obj };
|
|
}
|
|
|
|
const streamToString = stream =>
|
|
new Promise((resolve, reject) => {
|
|
let ended = false;
|
|
let data = "";
|
|
stream
|
|
.on("error", reject)
|
|
.on("data", chunk => {
|
|
if (ended) throw new Error("`data` emitted after `end`");
|
|
data += chunk;
|
|
})
|
|
.on("end", () => {
|
|
ended = true;
|
|
resolve(data);
|
|
});
|
|
});
|
|
|
|
const waitForBytesWritten = (stream, bytes, resolve) => {
|
|
if (stream.bytesWritten >= bytes) {
|
|
setImmediate(resolve);
|
|
return;
|
|
}
|
|
|
|
setImmediate(() => waitForBytesWritten(stream, bytes, resolve));
|
|
};
|
|
|
|
_tap.default.test("Data from a complete stream.", async t => {
|
|
let data = "";
|
|
const source = new _stream.default.Readable({
|
|
read() {}
|
|
});
|
|
const chunk1 = "1".repeat(10);
|
|
source.push(chunk1);
|
|
source.push(null);
|
|
data += chunk1;
|
|
let capacitor1 = new _.default();
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should start with 0 read streams"
|
|
);
|
|
source.pipe(capacitor1);
|
|
const capacitor1Stream1 = capacitor1.createReadStream("capacitor1Stream1");
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
1,
|
|
"should attach a new read stream before receiving data"
|
|
);
|
|
const result = await streamToString(capacitor1Stream1);
|
|
t.sameStrict(result, data, "should stream all data");
|
|
t.sameStrict(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should no longer have any attacheds read streams"
|
|
);
|
|
});
|
|
|
|
_tap.default.test(
|
|
"Data from an open stream, 1 chunk, no read streams.",
|
|
async t => {
|
|
let data = "";
|
|
const source = new _stream.default.Readable({
|
|
read() {}
|
|
});
|
|
let capacitor1 = new _.default();
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should start with 0 read streams"
|
|
);
|
|
source.pipe(capacitor1);
|
|
const chunk1 = "1".repeat(10);
|
|
source.push(chunk1);
|
|
source.push(null);
|
|
data += chunk1;
|
|
const capacitor1Stream1 = capacitor1.createReadStream("capacitor1Stream1");
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
1,
|
|
"should attach a new read stream before receiving data"
|
|
);
|
|
const result = await streamToString(capacitor1Stream1);
|
|
t.sameStrict(result, data, "should stream all data");
|
|
t.sameStrict(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should no longer have any attacheds read streams"
|
|
);
|
|
}
|
|
);
|
|
|
|
_tap.default.test(
|
|
"Data from an open stream, 1 chunk, 1 read stream.",
|
|
async t => {
|
|
let data = "";
|
|
const source = new _stream.default.Readable({
|
|
read() {}
|
|
});
|
|
let capacitor1 = new _.default();
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should start with 0 read streams"
|
|
);
|
|
source.pipe(capacitor1);
|
|
const capacitor1Stream1 = capacitor1.createReadStream("capacitor1Stream1");
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
1,
|
|
"should attach a new read stream before receiving data"
|
|
);
|
|
const chunk1 = "1".repeat(10);
|
|
source.push(chunk1);
|
|
source.push(null);
|
|
data += chunk1;
|
|
const result = await streamToString(capacitor1Stream1);
|
|
t.sameStrict(result, data, "should stream all data");
|
|
t.sameStrict(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should no longer have any attacheds read streams"
|
|
);
|
|
}
|
|
);
|
|
|
|
const withChunkSize = size =>
|
|
_tap.default.test(`--- with chunk size: ${size}`, async t => {
|
|
let data = "";
|
|
const source = new _stream.default.Readable({
|
|
read() {}
|
|
});
|
|
let capacitor1;
|
|
let capacitor1Stream1;
|
|
await t.test(
|
|
"can add a read stream before any data has been written",
|
|
async t => {
|
|
capacitor1 = new _.default();
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should start with 0 read streams"
|
|
);
|
|
capacitor1Stream1 = capacitor1.createReadStream("capacitor1Stream1");
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
1,
|
|
"should attach a new read stream before receiving data"
|
|
);
|
|
await t.test("creates a temporary file", async t => {
|
|
t.plan(3);
|
|
await new Promise(resolve => capacitor1.on("open", resolve));
|
|
t.type(
|
|
capacitor1.path,
|
|
"string",
|
|
"capacitor1.path should be a string"
|
|
);
|
|
t.type(capacitor1.fd, "number", "capacitor1.fd should be a number");
|
|
t.ok(_fs.default.existsSync(capacitor1.path), "creates a temp file");
|
|
});
|
|
}
|
|
);
|
|
source.pipe(capacitor1);
|
|
const chunk1 = "1".repeat(size);
|
|
source.push(chunk1);
|
|
data += chunk1;
|
|
await new Promise(resolve =>
|
|
waitForBytesWritten(capacitor1, size, resolve)
|
|
);
|
|
let capacitor1Stream2;
|
|
t.test("can add a read stream after data has been written", t => {
|
|
capacitor1Stream2 = capacitor1.createReadStream("capacitor1Stream2");
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
2,
|
|
"should attach a new read stream after first write"
|
|
);
|
|
t.end();
|
|
});
|
|
const writeEventBytesWritten = new Promise(resolve => {
|
|
capacitor1.once("write", () => {
|
|
resolve(capacitor1.bytesWritten);
|
|
});
|
|
});
|
|
const chunk2 = "2".repeat(size);
|
|
source.push(chunk2);
|
|
data += chunk2;
|
|
await new Promise(resolve =>
|
|
waitForBytesWritten(capacitor1, 2 * size, resolve)
|
|
);
|
|
await t.test("write event emitted after bytes are written", async t => {
|
|
t.strictSame(
|
|
await writeEventBytesWritten,
|
|
2 * size,
|
|
"bytesWritten should include new chunk"
|
|
);
|
|
});
|
|
const finished = new Promise(resolve => capacitor1.once("finish", resolve));
|
|
source.push(null);
|
|
await finished;
|
|
let capacitor1Stream3;
|
|
let capacitor1Stream4;
|
|
t.test("can create a read stream after the source has ended", t => {
|
|
capacitor1Stream3 = capacitor1.createReadStream("capacitor1Stream3");
|
|
capacitor1Stream4 = capacitor1.createReadStream("capacitor1Stream4");
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
4,
|
|
"should attach new read streams after end"
|
|
);
|
|
t.end();
|
|
});
|
|
await t.test("streams complete data to a read stream", async t => {
|
|
const result2 = await streamToString(capacitor1Stream2);
|
|
t.strictSame(
|
|
capacitor1Stream2.ended,
|
|
true,
|
|
"should mark read stream as ended"
|
|
);
|
|
t.strictSame(result2, data, "should stream complete data");
|
|
const result4 = await streamToString(capacitor1Stream4);
|
|
t.strictSame(
|
|
capacitor1Stream4.ended,
|
|
true,
|
|
"should mark read stream as ended"
|
|
);
|
|
t.strictSame(result4, data, "should stream complete data");
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
2,
|
|
"should detach an ended read stream"
|
|
);
|
|
});
|
|
await t.test("can destroy a read stream", async t => {
|
|
await new Promise(resolve => {
|
|
capacitor1Stream1.once("error", resolve);
|
|
capacitor1Stream1.destroy(new Error("test"));
|
|
});
|
|
t.strictSame(
|
|
capacitor1Stream1.destroyed,
|
|
true,
|
|
"should mark read stream as destroyed"
|
|
);
|
|
t.type(
|
|
capacitor1Stream1.error,
|
|
Error,
|
|
"should store an error on read stream"
|
|
);
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
1,
|
|
"should detach a destroyed read stream"
|
|
);
|
|
});
|
|
t.test("can delay destruction of a capacitor", t => {
|
|
capacitor1.destroy(null);
|
|
t.strictSame(
|
|
capacitor1.destroyed,
|
|
false,
|
|
"should not destroy while read streams exist"
|
|
);
|
|
t.strictSame(
|
|
capacitor1._destroyPending,
|
|
true,
|
|
"should mark for future destruction"
|
|
);
|
|
t.end();
|
|
});
|
|
await t.test("destroys capacitor once no read streams exist", async t => {
|
|
const readStreamDestroyed = new Promise(resolve =>
|
|
capacitor1Stream3.on("close", resolve)
|
|
);
|
|
const capacitorDestroyed = new Promise(resolve =>
|
|
capacitor1.on("close", resolve)
|
|
);
|
|
capacitor1Stream3.destroy(null);
|
|
await readStreamDestroyed;
|
|
t.strictSame(
|
|
capacitor1Stream3.destroyed,
|
|
true,
|
|
"should mark read stream as destroyed"
|
|
);
|
|
t.strictSame(
|
|
capacitor1Stream3.error,
|
|
null,
|
|
"should not store an error on read stream"
|
|
);
|
|
t.strictSame(
|
|
capacitor1._readStreams.size,
|
|
0,
|
|
"should detach a destroyed read stream"
|
|
);
|
|
await capacitorDestroyed;
|
|
t.strictSame(capacitor1.closed, true, "should mark capacitor as closed");
|
|
t.strictSame(capacitor1.fd, null, "should set fd to null");
|
|
t.strictSame(
|
|
capacitor1.destroyed,
|
|
true,
|
|
"should mark capacitor as destroyed"
|
|
);
|
|
t.notOk(_fs.default.existsSync(capacitor1.path), "removes its temp file");
|
|
});
|
|
t.test("cannot create a read stream after destruction", t => {
|
|
try {
|
|
capacitor1.createReadStream();
|
|
} catch (error) {
|
|
t.ok(
|
|
error instanceof _.ReadAfterDestroyedError,
|
|
"should not create a read stream once destroyed"
|
|
);
|
|
t.end();
|
|
}
|
|
});
|
|
const capacitor2 = new _.default();
|
|
const capacitor2Stream1 = capacitor2.createReadStream("capacitor2Stream1");
|
|
const capacitor2Stream2 = capacitor2.createReadStream("capacitor2Stream2");
|
|
const capacitor2ReadStream1Destroyed = new Promise(resolve =>
|
|
capacitor2Stream1.on("close", resolve)
|
|
);
|
|
const capacitor2Destroyed = new Promise(resolve =>
|
|
capacitor2.on("close", resolve)
|
|
);
|
|
capacitor2Stream1.destroy();
|
|
await capacitor2ReadStream1Destroyed;
|
|
await t.test("propagates errors to attached read streams", async t => {
|
|
capacitor2.destroy();
|
|
await new Promise(resolve => setImmediate(resolve));
|
|
t.strictSame(
|
|
capacitor2Stream2.destroyed,
|
|
false,
|
|
"should not immediately mark attached read streams as destroyed"
|
|
);
|
|
capacitor2.destroy(new Error("test"));
|
|
await capacitor2Destroyed;
|
|
t.type(capacitor2.error, Error, "should store an error on capacitor");
|
|
t.strictSame(
|
|
capacitor2.destroyed,
|
|
true,
|
|
"should mark capacitor as destroyed"
|
|
);
|
|
t.type(
|
|
capacitor2Stream2.error,
|
|
Error,
|
|
"should store an error on attached read streams"
|
|
);
|
|
t.strictSame(
|
|
capacitor2Stream2.destroyed,
|
|
true,
|
|
"should mark attached read streams as destroyed"
|
|
);
|
|
t.strictSame(
|
|
capacitor2Stream1.error,
|
|
null,
|
|
"should not store an error on detached read streams"
|
|
);
|
|
});
|
|
});
|
|
|
|
withChunkSize(10);
|
|
withChunkSize(100000);
|