| // META: global=window,worker,jsshell |
| // 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 Promise.all([promise_rejects_exactly(t, error1, reader.read(), 'read() should reject'), abortPromise]); |
| }); |
| }, 'writer.abort() inside size() should work'); |