LayoutTests/streams/reference-implementation/transform-stream.html

<!DOCTYPE html>
<script src='../../resources/testharness.js'></script>
<script src='../../resources/testharnessreport.js'></script>
<script src='resources/streams-utils.js'></script>
<script>
// This is updated till https://github.com/whatwg/streams/commit/4ba861e6f60c248060811830e11271c84b439cc3

test(function() {
    new TransformStream({ transform: function() { } }); // TransformStream constructed with no errors.
}, 'TransformStream can be constructed with a transform function');

test(function() {
    assert_throws_js(TypeError, function() { new TransformStream(); }, 'TransformStream cannot be constructed with no arguments');
    assert_throws_js(TypeError, function() { new TransformStream({}); }, 'TransformStream cannot be constructed with an empty object');
}, 'TransformStream cannot be constructed with no transform function');

test(function() {
    var ts = new TransformStream({ transform: function() { } });

    assert_true(Object.prototype.hasOwnProperty.call(ts, 'writable'), 'it has a writable property');
    assert_true(ts.writable instanceof WritableStream, 'writable is an instance of WritableStream');

    assert_true(Object.prototype.hasOwnProperty.call(ts, 'readable'), 'it has a readable property');
    assert_true(ts.readable instanceof ReadableStream, 'readable is an instance of ReadableStream');
}, 'TransformStream instances must have writable and readable properties of the correct types');

test(function() {
    var ts = new TransformStream({ transform: function() { } });

    assert_equals(ts.writable.state, 'writable', 'writable starts writable');
}, 'TransformStream writable starts in the writable state');

var test1 = async_test('Pass-through sync TransformStream: can read from readable what is put into writable');
test1.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            enqueue(chunk);
            done();
        }
    });

    ts.writable.write('a');

    assert_equals(ts.writable.state, 'waiting', 'writable is waiting after one write');
    ts.readable.getReader().read().then(test1.step_func(function(result) {
        assert_object_equals(result, { value: 'a', done: false }, 'result from reading the readable is the same as was written to writable');

        return ts.writable.ready.then(test1.step_func(function() {
            assert_equals(ts.writable.state, 'writable', 'writable becomes writable again');
            test1.done();
        }));
    })).catch(test1.step_func(function(e) { assert_unreached(e); }));
});

var test2 = async_test('Uppercaser sync TransformStream: can read from readable transformed version of what is put into writable');
test2.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            enqueue(chunk.toUpperCase());
            done();
        }
    });

    ts.writable.write('a');

    assert_equals(ts.writable.state, 'waiting', 'writable is waiting after one write');

    ts.readable.getReader().read().then(test2.step_func(function(result) {
        assert_object_equals(result, { value: 'A', done: false },
                             'result from reading the readable is the transformation of what was written to writable');

        return ts.writable.ready.then(test2.step_func(function() {
            assert_equals(ts.writable.state, 'writable', 'writable becomes writable again');
            test2.done();
        }));
    })).catch(test2.step_func(function(e) { assert_unreached(e); }));
});

var test3 = async_test('Uppercaser-doubler sync TransformStream: can read both chunks put into the readable');
test3.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            enqueue(chunk.toUpperCase());
            enqueue(chunk.toUpperCase());
            done();
        }
    });

    ts.writable.write('a');

    assert_equals(ts.writable.state, 'waiting', 'writable is waiting after one write');

    var reader = ts.readable.getReader();

    reader.read().then(test3.step_func(function(result1) {
        assert_object_equals(result1, { value: 'A', done: false },
                             'the first chunk read is the transformation of the single chunk written');

        return reader.read().then(test3.step_func(function(result2) {
            assert_object_equals(result2, { value: 'A', done: false },
                                 'the second chunk read is also the transformation of the single chunk written');

            return ts.writable.ready.then(test3.step_func(function() {
                assert_equals(ts.writable.state, 'writable', 'writable becomes writable again');
                test3.done();
            }));
        }));
    })).catch(test3.step_func(function(e) { assert_unreached(e); }));
});

var test4 = async_test('Uppercaser async TransformStream: can read from readable transformed version of what is put into writable');
test4.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            setTimeout(test4.step_func(function() { enqueue(chunk.toUpperCase()); }), 100);
            setTimeout(test4.step_func(done), 500);
        }
    });

    ts.writable.write('a');

    assert_equals(ts.writable.state, 'waiting', 'writable is waiting after one write');

    ts.readable.getReader().read().then(test4.step_func(function(result) {
        assert_object_equals(result, { value: 'A', done: false },
                             'result from reading the readable is the transformation of what was written to writable');

        return ts.writable.ready.then(test4.step_func(function() {
            assert_equals(ts.writable.state, 'writable', 'writable becomes writable again');
            test4.done();
        }));
    })).catch(test4.step_func(function(e) { assert_unreached(e); }));
});

var test5 = async_test('Uppercaser-doubler async TransformStream: can read both chunks put into the readable');
test5.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            setTimeout(test5.step_func(function() { enqueue(chunk.toUpperCase()); }), 100);
            setTimeout(test5.step_func(function() { enqueue(chunk.toUpperCase()); }), 500);
            setTimeout(test5.step_func(done), 900);
        }
    });

    var reader = ts.readable.getReader();

    ts.writable.write('a');

    assert_equals(ts.writable.state, 'waiting', 'writable is waiting after one write');
    reader.read().then(test5.step_func(function(result1) {
        assert_object_equals(result1, { value: 'A', done: false },
                             'the first chunk read is the transformation of the single chunk written');

        return reader.read().then(test5.step_func(function(result2) {
            assert_object_equals(result2, { value: 'A', done: false },
                                 'the second chunk read is also the transformation of the single chunk written');

            return ts.writable.ready.then(test5.step_func(function() {
                assert_equals(ts.writable.state, 'writable', 'writable becomes writable again');
                test5.done();
            }));
        }));
    })).catch(test5.step_func(function(e) { assert_unreached(e); }));
});

var test6 = async_test('TransformStream: by default, closing the writable closes the readable (when there are no queued writes)');
test6.step(function()
{
    var ts = new TransformStream({ transform: function() { } });

    ts.writable.close();
    assert_equals(ts.writable.state, 'closing', 'writable is closing');

    Promise.all([ts.writable.closed, ts.readable.getReader().closed]).then(test6.step_func(function() {
        assert_equals(ts.writable.state, 'closed', 'writable state becomes closed eventually');
        test6.done('both writable and readable closed promises fulfill');
    })).catch(test6.step_func(function(e) { assert_unreached(e); }));
});

var test7 = async_test('TransformStream: by default, closing the writable waits for transforms to finish before closing both');
test7.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            setTimeout(test7.step_func(done), 500);
        }
    });

    ts.writable.write('a');
    ts.writable.close();
    assert_equals(ts.writable.state, 'closing', 'writable is closing');

    var rsClosed = false;
    ts.readable.getReader().closed.then(test7.step_func(function() {
        rsClosed = true;
    }));

    setTimeout(test7.step_func(function() {
        assert_equals(rsClosed, false, 'readable is not closed after a tick');

        ts.writable.closed.then(test7.step_func(function() {
            assert_equals(ts.writable.state, 'closed', 'writable becomes closed eventually');
            assert_true(rsClosed, 'readable is closed at that point');
            test7.done();
        })).catch(test7.step_func(function(e) { assert_unreached(e); }));
    }), 0);
});

var test8 = async_test('TransformStream: by default, closing the writable closes the readable after sync enqueues and async done');
test8.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            enqueue('x');
            enqueue('y');
            setTimeout(test8.step_func(done), 500);
        }
    });

    ts.writable.write('a');
    ts.writable.close();
    assert_equals(ts.writable.state, 'closing', 'writable is closing');

    ts.writable.closed.then(test8.step_func(function() {
        assert_equals(ts.writable.state, 'closed', 'writable becomes closed eventually');

        return readableStreamToArray(ts.readable).then(test8.step_func(function(chunks) {
            assert_array_equals(chunks, ['x', 'y'], 'both enqueued chunks can be read from the readable');
            test8.done();
        }));
    })).catch(test8.step_func(function(e) { assert_unreached(e); }));
});

var test9 = async_test('TransformStream: by default, closing the writable closes the readable after async enqueues and async done');
test9.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            setTimeout(test9.step_func(function() { enqueue('x'); }), 100);
            setTimeout(test9.step_func(function() { enqueue('y'); }), 500);
            setTimeout(test9.step_func(done), 900);
        }
    });

    ts.writable.write('a');
    ts.writable.close();
    assert_equals(ts.writable.state, 'closing', 'writable is closing');

    ts.writable.closed.then(test9.step_func(function() {
        assert_equals(ts.writable.state, 'closed', 'writable becomes closed eventually');

        return readableStreamToArray(ts.readable).then(test9.step_func(function(chunks) {
            assert_array_equals(chunks, ['x', 'y'], 'both enqueued chunks can be read from the readable');
            test9.done();
        }));
    })).catch(test9.step_func(function(e) { assert_unreached(e); }));
});

var test10 = async_test('TransformStream flush is called immediately when the writable is closed, if no writes are queued');
test10.step(function()
{
    var flushCalled = false;
    var ts = new TransformStream({
        transform: function() { },
        flush: function(enqueue) {
            flushCalled = true;
        }
    });

    ts.writable.close().then(test10.step_func(function() {
        assert_true(flushCalled, 'closing the writable triggers the transform flush immediately');
        test10.done();
    }));
});

var test11 = async_test('TransformStream flush is called after all queued writes finish, once the writable is closed');
test11.step(function()
{
    var flushCalled = false;
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            setTimeout(test11.step_func(done), 100);
        },
        flush: function(enqueue) {
            flushCalled = true;
        }
    });

    ts.writable.write('a');
    ts.writable.close();
    assert_false(flushCalled, 'closing the writable does not immediately call flush if writes are not finished');

    var rsClosed = false;
    ts.readable.getReader().closed.then(test11.step_func(function() {
        rsClosed = true;
    }));

    setTimeout(test11.step_func(function() {
        assert_true(flushCalled, 'flush is eventually called');
        assert_false(rsClosed, 'if flush does not call close, the readable does not become closed');
        test11.done();
    }), 500);
});

var test12 = async_test('TransformStream flush gets a chance to enqueue more into the readable');
test12.step(function()
{
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            done();
        },
        flush: function(enqueue) {
            enqueue('x');
            enqueue('y');
        }
    });

    var reader = ts.readable.getReader();

    ts.writable.write('a');
    ts.writable.close();
    reader.read().then(test12.step_func(function(result1) {
        assert_object_equals(result1, { value: 'x', done: false }, 'the first chunk read is the first one enqueued in flush');

        return reader.read().then(test12.step_func(function(result2) {
            assert_object_equals(result2, { value: 'y', done: false }, 'the second chunk read is the second one enqueued in flush');
            test12.done();
        }));
    })).catch(test12.step_func(function(e) { assert_unreached(e); }));
});

var test13 = async_test('TransformStream flush gets a chance to enqueue more into the readable, and can then async close');
test13.step(function()
{
    var promiseCalls = 0;
    var ts = new TransformStream({
        transform: function(chunk, enqueue, done) {
            done();
        },
        flush: function(enqueue, close) {
            enqueue('x');
            enqueue('y');
            setTimeout(test13.step_func(close), 100);
        }
    });

    var reader = ts.readable.getReader();

    ts.writable.write('a');
    ts.writable.close();
    reader.read().then(test13.step_func(function(result1) {
        assert_equals(++promiseCalls, 1);
        assert_object_equals(result1, { value: 'x', done: false }, 'the first chunk read is the first one enqueued in flush');

        return reader.read().then(test13.step_func(function(result2) {
            assert_equals(++promiseCalls, 2);
            assert_object_equals(result2, { value: 'y', done: false }, 'the second chunk read is the second one enqueued in flush');
        }));
    })).catch(test13.step_func(function(e) { assert_unreached(e); }));

    reader.closed.then(test13.step_func(function() {
        assert_equals(++promiseCalls, 3);
        test13.done('readable reader becomes closed');
    })).catch(test13.step_func(function(e) { assert_unreached(e); }));
});

var test14 = async_test('Transform stream should call transformer methods as methods');
test14.step(function()
{
    var ts = new TransformStream({
        suffix: '-suffix',

        transform: function(chunk, enqueue, done) {
            enqueue(chunk + this.suffix);
            done();
        },

        flush: function(enqueue, close) {
            enqueue('flushed' + this.suffix);
            close();
        }
    });

    ts.writable.write('a');
    ts.writable.close();
    ts.writable.closed.then(
        test14.step_func(function() {
            return readableStreamToArray(ts.readable).then(test14.step_func(function(chunks) {
                assert_array_equals(chunks, ['a-suffix', 'flushed-suffix'], 'both enqueued chunks have suffixes');
                test14.done();
            }));
        }),
        test14.step_func(function(e) { assert_unreached(e); })
    );
});
</script>