mozilla-central/testing/web-platform/tests/streams/transform-streams/backpressure.any.js (file symbol)

Source code

Revision control

Copy as Markdown

Other Tools

Test Info:

// META: global=window,worker,shadowrealm
// META: script=../resources/recording-streams.js
// META: script=../resources/test-utils.js
'use strict';
const error1 = new Error('error1 message');
error1.name = 'error1';
promise_test(() => {
const ts = recordingTransformStream();
const writer = ts.writable.getWriter();
// This call never resolves.
writer.write('a');
return flushAsyncEvents().then(() => {
assert_array_equals(ts.events, [], 'transform should not be called');
});
}, 'backpressure allows no transforms with a default identity transform and no reader');
promise_test(() => {
const ts = recordingTransformStream({}, undefined, { highWaterMark: 1 });
const writer = ts.writable.getWriter();
// This call to write() resolves asynchronously.
writer.write('a');
// This call to write() waits for backpressure that is never relieved and never calls transform().
writer.write('b');
return flushAsyncEvents().then(() => {
assert_array_equals(ts.events, ['transform', 'a'], 'transform should be called once');
});
}, 'backpressure only allows one transform() with a identity transform with a readable HWM of 1 and no reader');
promise_test(() => {
// Without a transform() implementation, recordingTransformStream() never enqueues anything.
const ts = recordingTransformStream({
transform() {
// Discard all chunks. As a result, the readable side is never full enough to exert backpressure and transform()
// keeps being called.
}
}, undefined, { highWaterMark: 1 });
const writer = ts.writable.getWriter();
const writePromises = [];
for (let i = 0; i < 4; ++i) {
writePromises.push(writer.write(i));
}
return Promise.all(writePromises).then(() => {
assert_array_equals(ts.events, ['transform', 0, 'transform', 1, 'transform', 2, 'transform', 3],
'all 4 events should be transformed');
});
}, 'transform() should keep being called as long as there is no backpressure');
promise_test(() => {
const ts = new TransformStream({}, undefined, { highWaterMark: 1 });
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
const events = [];
const writerPromises = [
writer.write('a').then(() => events.push('a')),
writer.write('b').then(() => events.push('b')),
writer.close().then(() => events.push('closed'))];
return delay(0).then(() => {
assert_array_equals(events, ['a'], 'the first write should have resolved');
return reader.read();
}).then(({ value, done }) => {
assert_false(done, 'done should not be true');
assert_equals('a', value, 'value should be "a"');
return delay(0);
}).then(() => {
assert_array_equals(events, ['a', 'b', 'closed'], 'both writes and close() should have resolved');
return reader.read();
}).then(({ value, done }) => {
assert_false(done, 'done should still not be true');
assert_equals('b', value, 'value should be "b"');
return reader.read();
}).then(({ done }) => {
assert_true(done, 'done should be true');
return writerPromises;
});
}, 'writes should resolve as soon as transform completes');
promise_test(() => {
const ts = new TransformStream(undefined, undefined, { highWaterMark: 0 });
const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
const readPromise = reader.read();
writer.write('a');
return readPromise.then(({ value, done }) => {
assert_false(done, 'not done');
assert_equals(value, 'a', 'value should be "a"');
});
}, 'calling pull() before the first write() with backpressure should work');
promise_test(() => {
let reader;
const ts = recordingTransformStream({
transform(chunk, controller) {
controller.enqueue(chunk);
return reader.read();
}
}, undefined, { highWaterMark: 1 });
const writer = ts.writable.getWriter();
reader = ts.readable.getReader();
return writer.write('a');
}, 'transform() should be able to read the chunk it just enqueued');
promise_test(() => {
let resolveTransform;
const transformPromise = new Promise(resolve => {
resolveTransform = resolve;
});
const ts = recordingTransformStream({
transform() {
return transformPromise;
}
}, undefined, new CountQueuingStrategy({ highWaterMark: Infinity }));
const writer = ts.writable.getWriter();
assert_equals(writer.desiredSize, 1, 'desiredSize should be 1');
return delay(0).then(() => {
writer.write('a');
assert_array_equals(ts.events, ['transform', 'a']);
assert_equals(writer.desiredSize, 0, 'desiredSize should be 0');
return flushAsyncEvents();
}).then(() => {
assert_equals(writer.desiredSize, 0, 'desiredSize should still be 0');
resolveTransform();
return delay(0);
}).then(() => {
assert_equals(writer.desiredSize, 1, 'desiredSize should be 1');
});
}, 'blocking transform() should cause backpressure');
promise_test(t => {
const ts = new TransformStream();
ts.readable.cancel(error1);
return promise_rejects_exactly(t, error1, ts.writable.getWriter().closed, 'closed should reject');
}, 'writer.closed should resolve after readable is canceled during start');
promise_test(t => {
const ts = new TransformStream({}, undefined, { highWaterMark: 0 });
return delay(0).then(() => {
ts.readable.cancel(error1);
return promise_rejects_exactly(t, error1, ts.writable.getWriter().closed, 'closed should reject');
});
}, 'writer.closed should resolve after readable is canceled with backpressure');
promise_test(t => {
const ts = new TransformStream({}, undefined, { highWaterMark: 1 });
return delay(0).then(() => {
ts.readable.cancel(error1);
return promise_rejects_exactly(t, error1, ts.writable.getWriter().closed, 'closed should reject');
});
}, 'writer.closed should resolve after readable is canceled with no backpressure');
promise_test(() => {
const ts = new TransformStream({}, undefined, { highWaterMark: 1 });
const writer = ts.writable.getWriter();
return delay(0).then(() => {
const writePromise = writer.write('a');
ts.readable.cancel(error1);
return writePromise;
});
}, 'cancelling the readable should cause a pending write to resolve');
promise_test(t => {
const rs = new ReadableStream();
const ts = new TransformStream();
const pipePromise = rs.pipeTo(ts.writable);
ts.readable.cancel(error1);
return promise_rejects_exactly(t, error1, pipePromise, 'promise returned from pipeTo() should be rejected');
}, 'cancelling the readable side of a TransformStream should abort an empty pipe');
promise_test(t => {
const rs = new ReadableStream();
const ts = new TransformStream();
const pipePromise = rs.pipeTo(ts.writable);
return delay(0).then(() => {
ts.readable.cancel(error1);
return promise_rejects_exactly(t, error1, pipePromise, 'promise returned from pipeTo() should be rejected');
});
}, 'cancelling the readable side of a TransformStream should abort an empty pipe after startup');
promise_test(t => {
const rs = new ReadableStream({
start(controller) {
controller.enqueue('a');
controller.enqueue('b');
controller.enqueue('c');
}
});
const ts = new TransformStream();
const pipePromise = rs.pipeTo(ts.writable);
// Allow data to flow into the pipe.
return delay(0).then(() => {
ts.readable.cancel(error1);
return promise_rejects_exactly(t, error1, pipePromise, 'promise returned from pipeTo() should be rejected');
});
}, 'cancelling the readable side of a TransformStream should abort a full pipe');