mozilla-central/testing/web-platform/tests/streams/readable-streams/async-iterator.any.js (file symbol)

Source code

Revision control

Copy as Markdown

Other Tools

Test Info:

// META: global=window,worker,shadowrealm
// META: script=../resources/rs-utils.js
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
'use strict';
const error1 = new Error('error1');
function assert_iter_result(iterResult, value, done, message) {
const prefix = message === undefined ? '' : `${message} `;
assert_equals(typeof iterResult, 'object', `${prefix}type is object`);
assert_equals(Object.getPrototypeOf(iterResult), Object.prototype, `${prefix}[[Prototype]]`);
assert_array_equals(Object.getOwnPropertyNames(iterResult).sort(), ['done', 'value'], `${prefix}property names`);
assert_equals(iterResult.value, value, `${prefix}value`);
assert_equals(iterResult.done, done, `${prefix}done`);
}
test(() => {
const s = new ReadableStream();
const it = s.values();
const proto = Object.getPrototypeOf(it);
const AsyncIteratorPrototype = Object.getPrototypeOf(Object.getPrototypeOf(async function* () {}).prototype);
assert_equals(Object.getPrototypeOf(proto), AsyncIteratorPrototype, 'prototype should extend AsyncIteratorPrototype');
const methods = ['next', 'return'].sort();
assert_array_equals(Object.getOwnPropertyNames(proto).sort(), methods, 'should have all the correct methods');
for (const m of methods) {
const propDesc = Object.getOwnPropertyDescriptor(proto, m);
assert_true(propDesc.enumerable, 'method should be enumerable');
assert_true(propDesc.configurable, 'method should be configurable');
assert_true(propDesc.writable, 'method should be writable');
assert_equals(typeof it[m], 'function', 'method should be a function');
assert_equals(it[m].name, m, 'method should have the correct name');
}
assert_equals(it.next.length, 0, 'next should have no parameters');
assert_equals(it.return.length, 1, 'return should have 1 parameter');
assert_equals(typeof it.throw, 'undefined', 'throw should not exist');
}, 'Async iterator instances should have the correct list of properties');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.enqueue(1);
c.enqueue(2);
c.enqueue(3);
c.close();
}
});
const chunks = [];
for await (const chunk of s) {
chunks.push(chunk);
}
assert_array_equals(chunks, [1, 2, 3]);
}, 'Async-iterating a push source');
promise_test(async () => {
let i = 1;
const s = new ReadableStream({
pull(c) {
c.enqueue(i);
if (i >= 3) {
c.close();
}
i += 1;
}
});
const chunks = [];
for await (const chunk of s) {
chunks.push(chunk);
}
assert_array_equals(chunks, [1, 2, 3]);
}, 'Async-iterating a pull source');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.enqueue(undefined);
c.enqueue(undefined);
c.enqueue(undefined);
c.close();
}
});
const chunks = [];
for await (const chunk of s) {
chunks.push(chunk);
}
assert_array_equals(chunks, [undefined, undefined, undefined]);
}, 'Async-iterating a push source with undefined values');
promise_test(async () => {
let i = 1;
const s = new ReadableStream({
pull(c) {
c.enqueue(undefined);
if (i >= 3) {
c.close();
}
i += 1;
}
});
const chunks = [];
for await (const chunk of s) {
chunks.push(chunk);
}
assert_array_equals(chunks, [undefined, undefined, undefined]);
}, 'Async-iterating a pull source with undefined values');
promise_test(async () => {
let i = 1;
const s = recordingReadableStream({
pull(c) {
c.enqueue(i);
if (i >= 3) {
c.close();
}
i += 1;
},
}, new CountQueuingStrategy({ highWaterMark: 0 }));
const it = s.values();
assert_array_equals(s.events, []);
const read1 = await it.next();
assert_iter_result(read1, 1, false);
assert_array_equals(s.events, ['pull']);
const read2 = await it.next();
assert_iter_result(read2, 2, false);
assert_array_equals(s.events, ['pull', 'pull']);
const read3 = await it.next();
assert_iter_result(read3, 3, false);
assert_array_equals(s.events, ['pull', 'pull', 'pull']);
const read4 = await it.next();
assert_iter_result(read4, undefined, true);
assert_array_equals(s.events, ['pull', 'pull', 'pull']);
}, 'Async-iterating a pull source manually');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.error('e');
},
});
try {
for await (const chunk of s) {}
assert_unreached();
} catch (e) {
assert_equals(e, 'e');
}
}, 'Async-iterating an errored stream throws');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.close();
}
});
for await (const chunk of s) {
assert_unreached();
}
}, 'Async-iterating a closed stream never executes the loop body, but works fine');
promise_test(async () => {
const s = new ReadableStream();
const loop = async () => {
for await (const chunk of s) {
assert_unreached();
}
assert_unreached();
};
await Promise.race([
loop(),
flushAsyncEvents()
]);
}, 'Async-iterating an empty but not closed/errored stream never executes the loop body and stalls the async function');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.enqueue(1);
c.enqueue(2);
c.enqueue(3);
c.close();
},
});
const reader = s.getReader();
const readResult = await reader.read();
assert_iter_result(readResult, 1, false);
reader.releaseLock();
const chunks = [];
for await (const chunk of s) {
chunks.push(chunk);
}
assert_array_equals(chunks, [2, 3]);
}, 'Async-iterating a partially consumed stream');
for (const type of ['throw', 'break', 'return']) {
for (const preventCancel of [false, true]) {
promise_test(async () => {
const s = recordingReadableStream({
start(c) {
c.enqueue(0);
}
});
// use a separate function for the loop body so return does not stop the test
const loop = async () => {
for await (const c of s.values({ preventCancel })) {
if (type === 'throw') {
throw new Error();
} else if (type === 'break') {
break;
} else if (type === 'return') {
return;
}
}
};
try {
await loop();
} catch (e) {}
if (preventCancel) {
assert_array_equals(s.events, ['pull'], `cancel() should not be called`);
} else {
assert_array_equals(s.events, ['pull', 'cancel', undefined], `cancel() should be called`);
}
}, `Cancellation behavior when ${type}ing inside loop body; preventCancel = ${preventCancel}`);
}
}
for (const preventCancel of [false, true]) {
promise_test(async () => {
const s = recordingReadableStream({
start(c) {
c.enqueue(0);
}
});
const it = s.values({ preventCancel });
await it.return();
if (preventCancel) {
assert_array_equals(s.events, [], `cancel() should not be called`);
} else {
assert_array_equals(s.events, ['cancel', undefined], `cancel() should be called`);
}
}, `Cancellation behavior when manually calling return(); preventCancel = ${preventCancel}`);
}
promise_test(async t => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
if (timesPulled === 0) {
c.enqueue(0);
++timesPulled;
} else {
c.error(error1);
}
}
});
const it = s[Symbol.asyncIterator]();
const iterResult1 = await it.next();
assert_iter_result(iterResult1, 0, false, '1st next()');
await promise_rejects_exactly(t, error1, it.next(), '2nd next()');
}, 'next() rejects if the stream errors');
promise_test(async () => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
if (timesPulled === 0) {
c.enqueue(0);
++timesPulled;
} else {
c.error(error1);
}
}
});
const it = s[Symbol.asyncIterator]();
const iterResult = await it.return('return value');
assert_iter_result(iterResult, 'return value', true);
}, 'return() does not rejects if the stream has not errored yet');
promise_test(async t => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
// Do not error in start() because doing so would prevent acquiring a reader/async iterator.
c.error(error1);
}
});
const it = s[Symbol.asyncIterator]();
await flushAsyncEvents();
await promise_rejects_exactly(t, error1, it.return('return value'));
}, 'return() rejects if the stream has errored');
promise_test(async t => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
if (timesPulled === 0) {
c.enqueue(0);
++timesPulled;
} else {
c.error(error1);
}
}
});
const it = s[Symbol.asyncIterator]();
const iterResult1 = await it.next();
assert_iter_result(iterResult1, 0, false, '1st next()');
await promise_rejects_exactly(t, error1, it.next(), '2nd next()');
const iterResult3 = await it.next();
assert_iter_result(iterResult3, undefined, true, '3rd next()');
}, 'next() that succeeds; next() that reports an error; next()');
promise_test(async () => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
if (timesPulled === 0) {
c.enqueue(0);
++timesPulled;
} else {
c.error(error1);
}
}
});
const it = s[Symbol.asyncIterator]();
const iterResults = await Promise.allSettled([it.next(), it.next(), it.next()]);
assert_equals(iterResults[0].status, 'fulfilled', '1st next() promise status');
assert_iter_result(iterResults[0].value, 0, false, '1st next()');
assert_equals(iterResults[1].status, 'rejected', '2nd next() promise status');
assert_equals(iterResults[1].reason, error1, '2nd next() rejection reason');
assert_equals(iterResults[2].status, 'fulfilled', '3rd next() promise status');
assert_iter_result(iterResults[2].value, undefined, true, '3rd next()');
}, 'next() that succeeds; next() that reports an error(); next() [no awaiting]');
promise_test(async t => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
if (timesPulled === 0) {
c.enqueue(0);
++timesPulled;
} else {
c.error(error1);
}
}
});
const it = s[Symbol.asyncIterator]();
const iterResult1 = await it.next();
assert_iter_result(iterResult1, 0, false, '1st next()');
await promise_rejects_exactly(t, error1, it.next(), '2nd next()');
const iterResult3 = await it.return('return value');
assert_iter_result(iterResult3, 'return value', true, 'return()');
}, 'next() that succeeds; next() that reports an error(); return()');
promise_test(async () => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
if (timesPulled === 0) {
c.enqueue(0);
++timesPulled;
} else {
c.error(error1);
}
}
});
const it = s[Symbol.asyncIterator]();
const iterResults = await Promise.allSettled([it.next(), it.next(), it.return('return value')]);
assert_equals(iterResults[0].status, 'fulfilled', '1st next() promise status');
assert_iter_result(iterResults[0].value, 0, false, '1st next()');
assert_equals(iterResults[1].status, 'rejected', '2nd next() promise status');
assert_equals(iterResults[1].reason, error1, '2nd next() rejection reason');
assert_equals(iterResults[2].status, 'fulfilled', 'return() promise status');
assert_iter_result(iterResults[2].value, 'return value', true, 'return()');
}, 'next() that succeeds; next() that reports an error(); return() [no awaiting]');
promise_test(async () => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
c.enqueue(timesPulled);
++timesPulled;
}
});
const it = s[Symbol.asyncIterator]();
const iterResult1 = await it.next();
assert_iter_result(iterResult1, 0, false, 'next()');
const iterResult2 = await it.return('return value');
assert_iter_result(iterResult2, 'return value', true, 'return()');
assert_equals(timesPulled, 2);
}, 'next() that succeeds; return()');
promise_test(async () => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
c.enqueue(timesPulled);
++timesPulled;
}
});
const it = s[Symbol.asyncIterator]();
const iterResults = await Promise.allSettled([it.next(), it.return('return value')]);
assert_equals(iterResults[0].status, 'fulfilled', 'next() promise status');
assert_iter_result(iterResults[0].value, 0, false, 'next()');
assert_equals(iterResults[1].status, 'fulfilled', 'return() promise status');
assert_iter_result(iterResults[1].value, 'return value', true, 'return()');
assert_equals(timesPulled, 2);
}, 'next() that succeeds; return() [no awaiting]');
promise_test(async () => {
const rs = new ReadableStream();
const it = rs.values();
const iterResult1 = await it.return('return value');
assert_iter_result(iterResult1, 'return value', true, 'return()');
const iterResult2 = await it.next();
assert_iter_result(iterResult2, undefined, true, 'next()');
}, 'return(); next()');
promise_test(async () => {
const rs = new ReadableStream();
const it = rs.values();
const resolveOrder = [];
const iterResults = await Promise.allSettled([
it.return('return value').then(result => {
resolveOrder.push('return');
return result;
}),
it.next().then(result => {
resolveOrder.push('next');
return result;
})
]);
assert_equals(iterResults[0].status, 'fulfilled', 'return() promise status');
assert_iter_result(iterResults[0].value, 'return value', true, 'return()');
assert_equals(iterResults[1].status, 'fulfilled', 'next() promise status');
assert_iter_result(iterResults[1].value, undefined, true, 'next()');
assert_array_equals(resolveOrder, ['return', 'next'], 'next() resolves after return()');
}, 'return(); next() [no awaiting]');
promise_test(async () => {
let resolveCancelPromise;
const rs = recordingReadableStream({
cancel(reason) {
return new Promise(r => resolveCancelPromise = r);
}
});
const it = rs.values();
let returnResolved = false;
const returnPromise = it.return('return value').then(result => {
returnResolved = true;
return result;
});
await flushAsyncEvents();
assert_false(returnResolved, 'return() should not resolve while cancel() promise is pending');
resolveCancelPromise();
const iterResult1 = await returnPromise;
assert_iter_result(iterResult1, 'return value', true, 'return()');
const iterResult2 = await it.next();
assert_iter_result(iterResult2, undefined, true, 'next()');
}, 'return(); next() with delayed cancel()');
promise_test(async () => {
let resolveCancelPromise;
const rs = recordingReadableStream({
cancel(reason) {
return new Promise(r => resolveCancelPromise = r);
}
});
const it = rs.values();
const resolveOrder = [];
const returnPromise = it.return('return value').then(result => {
resolveOrder.push('return');
return result;
});
const nextPromise = it.next().then(result => {
resolveOrder.push('next');
return result;
});
assert_array_equals(rs.events, ['cancel', 'return value'], 'return() should call cancel()');
assert_array_equals(resolveOrder, [], 'return() should not resolve before cancel() resolves');
resolveCancelPromise();
const iterResult1 = await returnPromise;
assert_iter_result(iterResult1, 'return value', true, 'return() should resolve with original reason');
const iterResult2 = await nextPromise;
assert_iter_result(iterResult2, undefined, true, 'next() should resolve with done result');
assert_array_equals(rs.events, ['cancel', 'return value'], 'no pull() after cancel()');
assert_array_equals(resolveOrder, ['return', 'next'], 'next() should resolve after return() resolves');
}, 'return(); next() with delayed cancel() [no awaiting]');
promise_test(async () => {
const rs = new ReadableStream();
const it = rs.values();
const iterResult1 = await it.return('return value 1');
assert_iter_result(iterResult1, 'return value 1', true, '1st return()');
const iterResult2 = await it.return('return value 2');
assert_iter_result(iterResult2, 'return value 2', true, '1st return()');
}, 'return(); return()');
promise_test(async () => {
const rs = new ReadableStream();
const it = rs.values();
const resolveOrder = [];
const iterResults = await Promise.allSettled([
it.return('return value 1').then(result => {
resolveOrder.push('return 1');
return result;
}),
it.return('return value 2').then(result => {
resolveOrder.push('return 2');
return result;
})
]);
assert_equals(iterResults[0].status, 'fulfilled', '1st return() promise status');
assert_iter_result(iterResults[0].value, 'return value 1', true, '1st return()');
assert_equals(iterResults[1].status, 'fulfilled', '2nd return() promise status');
assert_iter_result(iterResults[1].value, 'return value 2', true, '1st return()');
assert_array_equals(resolveOrder, ['return 1', 'return 2'], '2nd return() resolves after 1st return()');
}, 'return(); return() [no awaiting]');
test(() => {
const s = new ReadableStream({
start(c) {
c.enqueue(0);
c.close();
},
});
s.values();
assert_throws_js(TypeError, () => s.values(), 'values() should throw');
}, 'values() throws if there\'s already a lock');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.enqueue(1);
c.enqueue(2);
c.enqueue(3);
c.close();
}
});
const chunks = [];
for await (const chunk of s) {
chunks.push(chunk);
}
assert_array_equals(chunks, [1, 2, 3]);
const reader = s.getReader();
await reader.closed;
}, 'Acquiring a reader after exhaustively async-iterating a stream');
promise_test(async t => {
let timesPulled = 0;
const s = new ReadableStream({
pull(c) {
if (timesPulled === 0) {
c.enqueue(0);
++timesPulled;
} else {
c.error(error1);
}
}
});
const it = s[Symbol.asyncIterator]({ preventCancel: true });
const iterResult1 = await it.next();
assert_iter_result(iterResult1, 0, false, '1st next()');
await promise_rejects_exactly(t, error1, it.next(), '2nd next()');
const iterResult2 = await it.return('return value');
assert_iter_result(iterResult2, 'return value', true, 'return()');
// i.e. it should not reject with a generic "this stream is locked" TypeError.
const reader = s.getReader();
await promise_rejects_exactly(t, error1, reader.closed, 'closed on the new reader should reject with the error');
}, 'Acquiring a reader after return()ing from a stream that errors');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.enqueue(1);
c.enqueue(2);
c.enqueue(3);
c.close();
},
});
// read the first two chunks, then cancel
const chunks = [];
for await (const chunk of s) {
chunks.push(chunk);
if (chunk >= 2) {
break;
}
}
assert_array_equals(chunks, [1, 2]);
const reader = s.getReader();
await reader.closed;
}, 'Acquiring a reader after partially async-iterating a stream');
promise_test(async () => {
const s = new ReadableStream({
start(c) {
c.enqueue(1);
c.enqueue(2);
c.enqueue(3);
c.close();
},
});
// read the first two chunks, then release lock
const chunks = [];
for await (const chunk of s.values({preventCancel: true})) {
chunks.push(chunk);
if (chunk >= 2) {
break;
}
}
assert_array_equals(chunks, [1, 2]);
const reader = s.getReader();
const readResult = await reader.read();
assert_iter_result(readResult, 3, false);
await reader.closed;
}, 'Acquiring a reader and reading the remaining chunks after partially async-iterating a stream with preventCancel = true');
for (const preventCancel of [false, true]) {
test(() => {
const rs = new ReadableStream();
rs.values({ preventCancel }).return();
// The test passes if this line doesn't throw.
rs.getReader();
}, `return() should unlock the stream synchronously when preventCancel = ${preventCancel}`);
}
promise_test(async () => {
const rs = new ReadableStream({
async start(c) {
c.enqueue('a');
c.enqueue('b');
c.enqueue('c');
await flushAsyncEvents();
// At this point, the async iterator has a read request in the stream's queue for its pending next() promise.
// Closing the stream now causes two things to happen *synchronously*:
// 1. ReadableStreamClose resolves reader.[[closedPromise]] with undefined.
// 2. ReadableStreamClose calls the read request's close steps, which calls ReadableStreamReaderGenericRelease,
// which replaces reader.[[closedPromise]] with a rejected promise.
c.close();
}
});
const chunks = [];
for await (const chunk of rs) {
chunks.push(chunk);
}
assert_array_equals(chunks, ['a', 'b', 'c']);
}, 'close() while next() is pending');