"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);