// META: global=window,worker,shadowrealm // META: script=../resources/recording-streams.js // META: script=../resources/rs-utils.js // META: script=../resources/test-utils.js 'use strict'; // The size() function of readableStrategy can re-entrantly call back into the TransformStream implementation. This // makes it risky to cache state across the call to ReadableStreamDefaultControllerEnqueue. These tests attempt to catch // such errors. They are separated from the other strategy tests because no real user code should ever do anything like // this. // // There is no such issue with writableStrategy size() because it is never called from within TransformStream // algorithms. const error1 = new Error('error1'); error1.name = 'error1'; promise_test(() => { let controller; let calls = 0; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { ++calls; if (calls < 2) { controller.enqueue('b'); } return 1; }, highWaterMark: Infinity }); const writer = ts.writable.getWriter(); return Promise.all([writer.write('a'), writer.close()]) .then(() => readableStreamToArray(ts.readable)) .then(array => assert_array_equals(array, ['b', 'a'], 'array should contain two chunks')); }, 'enqueue() inside size() should work'); promise_test(() => { let controller; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { // The readable queue is empty. controller.terminate(); // The readable state has gone from "readable" to "closed". return 1; // This chunk will be enqueued, but will be impossible to read because the state is already "closed". }, highWaterMark: Infinity }); const writer = ts.writable.getWriter(); return writer.write('a') .then(() => readableStreamToArray(ts.readable)) .then(array => assert_array_equals(array, [], 'array should contain no chunks')); // The chunk 'a' is still in readable's queue. readable is closed so 'a' cannot be read. writable's queue is empty and // it is still writable. }, 'terminate() inside size() should work'); promise_test(t => { let controller; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { controller.error(error1); return 1; }, highWaterMark: Infinity }); const writer = ts.writable.getWriter(); return writer.write('a') .then(() => promise_rejects_exactly(t, error1, ts.readable.getReader().read(), 'read() should reject')); }, 'error() inside size() should work'); promise_test(() => { let controller; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { assert_equals(controller.desiredSize, 1, 'desiredSize should be 1'); return 1; }, highWaterMark: 1 }); const writer = ts.writable.getWriter(); return Promise.all([writer.write('a'), writer.close()]) .then(() => readableStreamToArray(ts.readable)) .then(array => assert_array_equals(array, ['a'], 'array should contain one chunk')); }, 'desiredSize inside size() should work'); promise_test(t => { let cancelPromise; const ts = new TransformStream({}, undefined, { size() { cancelPromise = ts.readable.cancel(error1); return 1; }, highWaterMark: Infinity }); const writer = ts.writable.getWriter(); return writer.write('a') .then(() => { promise_rejects_exactly(t, error1, writer.closed, 'writer.closed should reject'); return cancelPromise; }); }, 'readable cancel() inside size() should work'); promise_test(() => { let controller; let pipeToPromise; const ws = recordingWritableStream(); const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { if (!pipeToPromise) { pipeToPromise = ts.readable.pipeTo(ws); } return 1; }, highWaterMark: 1 }); // Allow promise returned by start() to resolve so that enqueue() will happen synchronously. return delay(0).then(() => { controller.enqueue('a'); assert_not_equals(pipeToPromise, undefined); // Some pipeTo() implementations need an additional chunk enqueued in order for the first one to be processed. See // https://github.com/whatwg/streams/issues/794 for background. controller.enqueue('a'); // Give pipeTo() a chance to process the queued chunks. return delay(0); }).then(() => { assert_array_equals(ws.events, ['write', 'a', 'write', 'a'], 'ws should contain two chunks'); controller.terminate(); return pipeToPromise; }).then(() => { assert_array_equals(ws.events, ['write', 'a', 'write', 'a', 'close'], 'target should have been closed'); }); }, 'pipeTo() inside size() should work'); promise_test(() => { let controller; let readPromise; let calls = 0; let reader; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { // This is triggered by controller.enqueue(). The queue is empty and there are no pending reads. pull() is called // synchronously, allowing transform() to proceed asynchronously. This results in a second call to enqueue(), // which resolves this pending read() without calling size() again. readPromise = reader.read(); ++calls; return 1; }, highWaterMark: 0 }); reader = ts.readable.getReader(); const writer = ts.writable.getWriter(); let writeResolved = false; const writePromise = writer.write('b').then(() => { writeResolved = true; }); return flushAsyncEvents().then(() => { assert_false(writeResolved); controller.enqueue('a'); assert_equals(calls, 1, 'size() should have been called once'); return delay(0); }).then(() => { assert_true(writeResolved); assert_equals(calls, 1, 'size() should only be called once'); return readPromise; }).then(({ value, done }) => { assert_false(done, 'done should be false'); // See https://github.com/whatwg/streams/issues/794 for why this chunk is not 'a'. assert_equals(value, 'b', 'chunk should have been read'); assert_equals(calls, 1, 'calls should still be 1'); return writePromise; }); }, 'read() inside of size() should work'); promise_test(() => { let writer; let writePromise1; let calls = 0; const ts = new TransformStream({}, undefined, { size() { ++calls; if (calls < 2) { writePromise1 = writer.write('a'); } return 1; }, highWaterMark: Infinity }); writer = ts.writable.getWriter(); // Give pull() a chance to be called. return delay(0).then(() => { // This write results in a synchronous call to transform(), enqueue(), and size(). const writePromise2 = writer.write('b'); assert_equals(calls, 1, 'size() should have been called once'); return Promise.all([writePromise1, writePromise2, writer.close()]); }).then(() => { assert_equals(calls, 2, 'size() should have been called twice'); return readableStreamToArray(ts.readable); }).then(array => { assert_array_equals(array, ['b', 'a'], 'both chunks should have been enqueued'); assert_equals(calls, 2, 'calls should still be 2'); }); }, 'writer.write() inside size() should work'); promise_test(() => { let controller; let writer; let writePromise; let calls = 0; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { ++calls; if (calls < 2) { writePromise = writer.write('a'); } return 1; }, highWaterMark: Infinity }); writer = ts.writable.getWriter(); // Give pull() a chance to be called. return delay(0).then(() => { // This enqueue results in synchronous calls to size(), write(), transform() and enqueue(). controller.enqueue('b'); assert_equals(calls, 2, 'size() should have been called twice'); return Promise.all([writePromise, writer.close()]); }).then(() => { return readableStreamToArray(ts.readable); }).then(array => { // Because one call to enqueue() is nested inside the other, they finish in the opposite order that they were // called, so the chunks end up reverse order. assert_array_equals(array, ['a', 'b'], 'both chunks should have been enqueued'); assert_equals(calls, 2, 'calls should still be 2'); }); }, 'synchronous writer.write() inside size() should work'); promise_test(() => { let writer; let closePromise; let controller; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { closePromise = writer.close(); return 1; }, highWaterMark: 1 }); writer = ts.writable.getWriter(); const reader = ts.readable.getReader(); // Wait for the promise returned by start() to be resolved so that the call to close() will result in a synchronous // call to TransformStreamDefaultSink. return delay(0).then(() => { controller.enqueue('a'); return reader.read(); }).then(({ value, done }) => { assert_false(done, 'done should be false'); assert_equals(value, 'a', 'value should be correct'); return reader.read(); }).then(({ done }) => { assert_true(done, 'done should be true'); return closePromise; }); }, 'writer.close() inside size() should work'); promise_test(t => { let abortPromise; let controller; const ts = new TransformStream({ start(c) { controller = c; } }, undefined, { size() { abortPromise = ts.writable.abort(error1); return 1; }, highWaterMark: 1 }); const reader = ts.readable.getReader(); // Wait for the promise returned by start() to be resolved so that the call to abort() will result in a synchronous // call to TransformStreamDefaultSink. return delay(0).then(() => { controller.enqueue('a'); return reader.read(); }).then(({ value, done }) => { assert_false(done, 'done should be false'); assert_equals(value, 'a', 'value should be correct'); return Promise.all([promise_rejects_exactly(t, error1, reader.read(), 'read() should reject'), abortPromise]); }); }, 'writer.abort() inside size() should work');