Diffusion Ant (type-hints-typescript) Code

ant/examples/spec/streams-pipe.js
899dad6b8a9etype-hints-typescript

streams-pipe.js
View Options

			import { test, testDeep, testThrows, summary } from './helpers.js';

			console.log('ReadableStream pipe operations (tee, pipeTo, pipeThrough) Tests\n');

			async function testTeeBasic() {
			const rs = new ReadableStream({
			start(c) {
			c.enqueue('a');
			c.enqueue('b');
			c.close();
			}
			});
			const [b1, b2] = rs.tee();
			test('tee returns array', Array.isArray([b1, b2]), true);
			test('tee branch1 is ReadableStream', b1 instanceof ReadableStream, true);
			test('tee branch2 is ReadableStream', b2 instanceof ReadableStream, true);
			test('tee locks original', rs.locked, true);

			const r1 = b1.getReader();
			const r2 = b2.getReader();

			const v1a = await r1.read();
			test('tee b1 read1 value', v1a.value, 'a');
			test('tee b1 read1 done', v1a.done, false);
			const v1b = await r1.read();
			test('tee b1 read2 value', v1b.value, 'b');
			const v1c = await r1.read();
			test('tee b1 done', v1c.done, true);

			const v2a = await r2.read();
			test('tee b2 read1 value', v2a.value, 'a');
			const v2b = await r2.read();
			test('tee b2 read2 value', v2b.value, 'b');
			const v2c = await r2.read();
			test('tee b2 done', v2c.done, true);
			}

			async function testTeePullBased() {
			let pullCount = 0;
			const rs = new ReadableStream({
			pull(c) {
			pullCount++;
			if (pullCount <= 3) c.enqueue(pullCount);
			else c.close();
			}
			});
			const [b1, b2] = rs.tee();
			const r1 = b1.getReader();
			const r2 = b2.getReader();

			const chunks1 = [];
			const chunks2 = [];
			while (true) {
			const { value, done } = await r1.read();
			if (done) break;
			chunks1.push(value);
			}
			while (true) {
			const { value, done } = await r2.read();
			if (done) break;
			chunks2.push(value);
			}
			testDeep('tee pull b1 chunks', chunks1, [1, 2, 3]);
			testDeep('tee pull b2 chunks', chunks2, [1, 2, 3]);
			}

			async function testTeeCancelOne() {
			let cancelReason = null;
			const rs = new ReadableStream({
			pull(c) {
			c.enqueue('x');
			},
			cancel(r) {
			cancelReason = r;
			}
			});
			const [b1, b2] = rs.tee();
			const cancelPromise = b1.cancel('reason1');
			test('tee cancel one does not cancel original', cancelReason, null);

			const r2 = b2.getReader();
			const v = await r2.read();
			test('tee other branch still works', v.value, 'x');

			// The first branch's cancel does not settle until the other branch finishes.
			await r2.cancel('reason2');
			await cancelPromise;
			}

			async function testTeeCancelBoth() {
			let cancelReason = null;
			const rs = new ReadableStream({
			pull(c) {
			c.enqueue('x');
			},
			cancel(r) {
			cancelReason = r;
			}
			});
			const [b1, b2] = rs.tee();
			const p1 = b1.cancel('r1');
			const p2 = b2.cancel('r2');
			await Promise.all([p1, p2]);
			test('tee cancel both cancels original', Array.isArray(cancelReason), true);
			testDeep('tee composite reason', cancelReason, ['r1', 'r2']);
			}

			testThrows('tee locked stream throws', () => {
			const rs = new ReadableStream();
			rs.getReader();
			rs.tee();
			});

			async function testPipeToBasic() {
			const chunks = [];
			const rs = new ReadableStream({
			start(c) {
			c.enqueue(1);
			c.enqueue(2);
			c.enqueue(3);
			c.close();
			}
			});
			const ws = new WritableStream({
			write(chunk) {
			chunks.push(chunk);
			}
			});
			await rs.pipeTo(ws);
			testDeep('pipeTo basic chunks', chunks, [1, 2, 3]);
			}

			async function testPipeToClosesDestination() {
			let closed = false;
			const rs = new ReadableStream({
			start(c) {
			c.enqueue('a');
			c.close();
			}
			});
			const ws = new WritableStream({
			write() {},
			close() {
			closed = true;
			}
			});
			await rs.pipeTo(ws);
			test('pipeTo closes dest', closed, true);
			}

			async function testPipeToPreventClose() {
			let closed = false;
			const rs = new ReadableStream({
			start(c) {
			c.close();
			}
			});
			const ws = new WritableStream({
			close() {
			closed = true;
			}
			});
			await rs.pipeTo(ws, { preventClose: true });
			test('pipeTo preventClose', closed, false);
			}

			async function testPipeToErrorPropagation() {
			const err = new Error('source error');
			const rs = new ReadableStream({
			start(c) {
			c.error(err);
			}
			});
			let abortReason = null;
			const ws = new WritableStream({
			write() {},
			abort(r) {
			abortReason = r;
			}
			});
			try {
			await rs.pipeTo(ws);
			test('pipeTo should reject on source error', false, true);
			} catch (e) {
			test('pipeTo rejects with source error', e, err);
			}
			}

			async function testPipeToPreventAbort() {
			const err = new Error('source error');
			const rs = new ReadableStream({
			start(c) {
			c.error(err);
			}
			});
			let aborted = false;
			const ws = new WritableStream({
			write() {},
			abort() {
			aborted = true;
			}
			});
			try {
			await rs.pipeTo(ws, { preventAbort: true });
			} catch (e) {}
			test('pipeTo preventAbort', aborted, false);
			}

			async function testPipeToLocksStreams() {
			const rs = new ReadableStream({
			start(c) {
			c.close();
			}
			});
			const ws = new WritableStream();
			const p = rs.pipeTo(ws);
			test('pipeTo locks source', rs.locked, true);
			test('pipeTo locks dest', ws.locked, true);
			await p;
			test('pipeTo unlocks source', rs.locked, false);
			test('pipeTo unlocks dest', ws.locked, false);
			}

			async function testPipeToWithSignal() {
			const ac = new AbortController();
			const rs = new ReadableStream({
			start(c) {
			c.enqueue('a');
			c.enqueue('b');
			c.close();
			}
			});
			const chunks = [];
			const ws = new WritableStream({
			write(chunk) {
			chunks.push(chunk);
			ac.abort();
			}
			});

			try {
			await rs.pipeTo(ws, { signal: ac.signal });
			test('pipeTo with signal should reject', false, true);
			} catch (e) {
			test('pipeTo aborted by signal', true, true);
			test('pipeTo signal rejects with abort semantics',
			!(e instanceof TypeError) && !(e && /AbortSignal/.test(String(e && e.message))),
			true);
			}
			testDeep('pipeTo signal stops further writes', chunks, ['a']);
			}

			async function testPipeToAlreadyAbortedSignal() {
			const ac = new AbortController();
			ac.abort('pre-aborted');
			const rs = new ReadableStream({
			start(c) {
			c.enqueue('x');
			c.close();
			}
			});
			const ws = new WritableStream();
			try {
			await rs.pipeTo(ws, { signal: ac.signal });
			test('pipeTo pre-aborted should reject', false, true);
			} catch (e) {
			test('pipeTo pre-aborted signal rejects', true, true);
			}
			}

			async function testPipeToRejectsLocked() {
			const rs = new ReadableStream();
			rs.getReader();
			const ws = new WritableStream();
			try {
			await rs.pipeTo(ws);
			test('pipeTo locked source should reject', false, true);
			} catch (e) {
			test('pipeTo locked source rejects', true, true);
			}
			}

			async function testPipeThroughBasic() {
			const rs = new ReadableStream({
			start(c) {
			c.enqueue(1);
			c.enqueue(2);
			c.enqueue(3);
			c.close();
			}
			});

			let transformController;
			const transform = {
			writable: new WritableStream({
			write(chunk) {
			transformController.enqueue(chunk * 10);
			},
			close() {
			transformController.close();
			}
			}),
			readable: new ReadableStream({
			start(c) {
			transformController = c;
			}
			})
			};

			const result = rs.pipeThrough(transform);
			test('pipeThrough returns readable', result instanceof ReadableStream, true);

			const reader = result.getReader();
			const chunks = [];
			while (true) {
			const { value, done } = await reader.read();
			if (done) break;
			chunks.push(value);
			}
			testDeep('pipeThrough transforms data', chunks, [10, 20, 30]);
			}

			testThrows('pipeThrough locked source throws', () => {
			const rs = new ReadableStream();
			rs.getReader();
			rs.pipeThrough({ writable: new WritableStream(), readable: new ReadableStream() });
			});

			testThrows('pipeThrough missing transform throws', () => {
			new ReadableStream().pipeThrough();
			});

			testThrows('pipeThrough locked writable throws', () => {
			const ws = new WritableStream();
			ws.getWriter();
			new ReadableStream().pipeThrough({ writable: ws, readable: new ReadableStream() });
			});

			await testTeeBasic();
			await testTeePullBased();
			await testTeeCancelOne();
			await testTeeCancelBoth();

			await testPipeToBasic();
			await testPipeToClosesDestination();
			await testPipeToPreventClose();
			await testPipeToErrorPropagation();
			await testPipeToPreventAbort();
			await testPipeToLocksStreams();
			await testPipeToWithSignal();
			await testPipeToAlreadyAbortedSignal();
			await testPipeToRejectsLocked();

			await testPipeThroughBasic();

			summary();

Owner Packages

No Owners