Function ReadableStreamPipeTo

examples/io/streams/modules/streams.js:771–979

(source, dest, preventClose, preventAbort, preventCancel, signal)

	return stream.reader !== undefined;
}
function ReadableStreamPipeTo(source, dest, preventClose, preventAbort, preventCancel, signal) {
	assert(ReadableStreamBrand.has(source.self));
	assert(WritableStreamBrand.has(dest.self));
	assert(typeof preventClose === 'boolean');
	assert(typeof preventAbort === 'boolean');
	assert(typeof preventCancel === 'boolean');
	if (signal !== undefined)
		signal = AbortSignalBrand.get(signal);
	assert(IsReadableStreamLocked(source) === false);
	assert(IsWritableStreamLocked(dest) === false);
	const reader = AcquireReadableStreamDefaultReader(source);
	const writer = AcquireWritableStreamDefaultWriter(dest);
	source.disturbed = true;
	let shuttingDown = false;
	// This is used to keep track of the spec's requirement that we wait for ongoing writes during shutdown.
	let currentWrite = createResolvedPromise(undefined);
	return new Promise((resolve, reject) => {
		let abortAlgorithm;
		if (signal !== undefined) {
			abortAlgorithm = () => {
				const error = signal.reason;
				const actions = [];
				if (preventAbort === false) {
					actions.push(() => {
						if (dest.state === 'writable') {
							return WritableStreamAbort(dest, error);
						}
						return Promise.resolve(undefined);
					});
				}
				if (preventCancel === false) {
					actions.push(() => {
						if (source.state === 'readable') {
							return ReadableStreamCancel(source, error);
						}
						return Promise.resolve(undefined);
					});
				}
				shutdownWithAction(() => Promise.all(actions.map(action => action())), true, error);
			};
			if (signal.aborted === true) {
				abortAlgorithm();
				return;
			}
			signal.self.addEventListener('abort', abortAlgorithm);
		}

		// Using reader and writer, read all chunks from this and write them to dest
		// - Backpressure must be enforced
		// - Shutdown must stop all activity
		function pipeLoop() {
			return new Promise((resolveLoop, rejectLoop) => {
				function next(done) {
					if (done) {
						resolveLoop();
					} else {
						pipeStep().then(next, rejectLoop);
					}