<!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

var test1 = async_test('Piping from a ReadableStream from which lots of data are readable synchronously');
test1.step(function() {
    var events = 0;
    var rs = new ReadableStream({
        start: function(c) {
            for (var i = 0; i < 1000; ++i) {
                c.enqueue(i);
            }
            c.close();
        }
    });

    var ws = new WritableStream({}, new CountQueuingStrategy({ highWaterMark: 1000 }));

    assert_equals(ws.state, 'writable', 'writable stream state should start out writable');

    var pipeFinished = false;
    rs.pipeTo(ws).then(
        test1.step_func(function() {
            pipeFinished = true;
            rs.getReader().closed.then(test1.step_func(function() {
                assert_equals(++events, 2);
            }));
            assert_equals(ws.state, 'closed', 'writable stream state should be closed after pipe finishes');
            assert_equals(++events, 1);
        })
    ).catch(test1.step_func(function(e) { assert_unreached(e); }));

    setTimeout(test1.step_func(function() {
        assert_true(pipeFinished, 'pipe should have finished before a setTimeout(,0) since it should only be microtasks');
        assert_equals(++events, 3);
        test1.done();
    }), 0);
});

var test2 = async_test('Piping from a ReadableStream in readable state to a WritableStream in closing state');
test2.step(function() {
    var events = 0;
    var cancelReason;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('Hello');
        },
        cancel: function(reason) {
            assert_throws(new TypeError(), function() { throw reason; }, 'underlying source cancel should have been called with a TypeError');
            assert_equals(++events, 1);
            cancelReason = reason;
        }
    });

    var ws = new WritableStream({
        write: function() {
            assert_unreached('Unexpected write call');
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });

    ws.close();
    assert_equals(ws.state, 'closing', 'writable stream should be closing immediately after closing it');

    rs.pipeTo(ws).then(
        test2.step_func(function() { assert_unreached('promise returned by pipeTo should not fulfill'); }),
        test2.step_func(function(r) {
            assert_equals(r, cancelReason, 'the pipeTo promise should reject with the same error as the underlying source cancel was called with');
            assert_equals(++events, 2);
            rs.getReader().closed.then(test2.step_func(function() {
                assert_equals(++events, 3);
                test2.done();
            }));
        })
    ).catch(test2.step_func(function(e) { assert_unreached(e); }));
});

var test3 = async_test('Piping from a ReadableStream in readable state to a WritableStream in errored state');
test3.step(function() {
    var pullCount = 0;
    var cancelCalled = false;
    var passedError = new Error('horrible things');
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('Hello');
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function(reason) {
            assert_false(cancelCalled, 'cancel must not be called more than once');
            cancelCalled = true;

            assert_equals(reason, passedError);
        }
    });

    var writeCalled = false;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            assert_false(writeCalled, 'write must not be called more than once');
            writeCalled = true;

            assert_equals(chunk, 'Hello');

            return Promise.reject(passedError);
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });

    startPromise.then(test3.step_func(function() {
        ws.write('Hello');
        assert_true(writeCalled, 'write must be called');

        ws.ready.then(test3.step_func(function() {
            assert_equals(ws.state, 'errored', 'as a result of rejected promise, ws must be in errored state');

            rs.pipeTo(ws).catch(test3.step_func(function(e) {
                assert_equals(e, passedError, 'pipeTo promise should be rejected with the error');
                assert_true(cancelCalled, 'cancel should have been called');
                test3.done();
            }));
        }));
    }));
});

var test4 = async_test('Piping from a ReadableStream in the readable state which becomes closed after pipeTo call to a WritableStream in the writable state');
test4.step(function() {
    var closeReadableStream;
    var pullCount = 0;
    var events = 0;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('Hello');
            closeReadableStream = c.close.bind(c);
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var writeCalled = false;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            if (!writeCalled) {
                assert_equals(chunk, 'Hello', 'chunk written to writable stream should be the one enqueued into the readable stream');
                writeCalled = true;
                assert_equals(++events, 2);
            } else {
                assert_unreached('Unexpected extra write call');
            }
        },
        close: function() {
            assert_equals(pullCount, 1, 'underlying source pull should have been called once');
            assert_equals(++events, 3);
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });

    startPromise.then(test4.step_func(function() {
        rs.pipeTo(ws).then(test4.step_func(function() {
            assert_equals(ws.state, 'closed', 'writable stream should be closed after pipeTo completes');
            assert_equals(++events, 4);
            test4.done();
        }));

        assert_equals(ws.state, 'writable', 'writable stream should still be writable immediately after pipeTo');
        assert_equals(++events, 1);

        closeReadableStream();
    }));
});

var test5 = async_test('Piping from a ReadableStream in the readable state which becomes errored after pipeTo call to a WritableStream in the writable state');
test5.step(function() {
    var errorReadableStream;
    var pullCount = 0;
    var events = 0;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('Hello');
            errorReadableStream = c.error.bind(c);
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var passedError = new Error('horrible things');
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            assert_unreached('Unexpected extra write call');
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function(reason) {
            assert_equals(reason, passedError, 'underlying sink abort should receive the error from the readable stream');
            assert_equals(pullCount, 1, 'underlying source pull should have been called once');
            assert_equals(++events, 2);
        }
    });

    startPromise.then(test5.step_func(function() {
        rs.pipeTo(ws).catch(test5.step_func(function(e) {
            assert_equals(e, passedError, 'pipeTo should be rejected with the passed error');
            assert_equals(ws.state, 'errored', 'writable stream should be errored after pipeTo completes');
            assert_equals(++events, 3);
            test5.done();
        }));

        assert_equals(ws.state, 'writable', 'writable stream should still be writable immediately after pipeTo');
        assert_equals(++events, 1);

        errorReadableStream(passedError);
    }));
});

var test6 = async_test('Piping from an empty ReadableStream which becomes non-empty after pipeTo call to a WritableStream in the writable state');
test6.step(function() {
    var controller;
    var pullCount = 0;
    var rs = new ReadableStream({
        start: function(c) {
            controller = c;
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var ws = new WritableStream({
        write: function(chunk) {
            assert_equals(chunk, 'Hello', 'underlying sink write should be called with the single chunk');
            assert_equals(pullCount, 1, 'pull should have been called once');
            test6.done();
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function(reason) {
            assert_unreached('Unexpected abort call');
        }
    });

    rs.pipeTo(ws).then(test6.step_func(function() { assert_unreached('pipeTo promise should not fulfill'); }));
    assert_equals(ws.state, 'writable', 'writable stream should start in writable state');

    controller.enqueue('Hello');
});

var test7 = async_test('Piping from an empty ReadableStream which becomes errored after pipeTo call to a WritableStream in the writable state');
test7.step(function() {
    var errorReadableStream;
    var abortCalled = false;
    var rs = new ReadableStream({
        start: function(c) {
            errorReadableStream = c.error.bind(c);
        },
        pull: function() {
            assert_unreached('Unexpected pull call');
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var passedError = new Error('horrible things');
    var ws = new WritableStream({
        write: function() {
            assert_unreached('Unexpected write call');
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function(reason) {
            assert_equals(reason, passedError, 'underlying sink abort should receive the error from the readable stream');
            abortCalled = true;
        }
    });

    rs.pipeTo(ws).catch(test7.step_func(function(e) {
        assert_equals(e, passedError, 'pipeTo should reject with the passed error');
        assert_true(abortCalled);
        test7.done();
    }));
    assert_equals(ws.state, 'writable', 'writable stream should start out writable');
    errorReadableStream(passedError);
});

var test8 = async_test('Piping from an empty ReadableStream to a WritableStream in the writable state which becomes errored after a pipeTo call');
test8.step(function() {
    var cancelCalled = false;

    var theError = new Error('cancel with me!');

    var pullCount = 0;
    var rs = new ReadableStream({
        pull: function() {
            ++pullCount;
        },
        cancel: function(reason) {
            assert_equals(reason, theError, 'underlying source cancellation reason should be the writable stream error');
            assert_equals(pullCount, 2, 'pull should have been called twice by cancel-time');
            cancelCalled = true;
        }
    });

    var errorWritableStream;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function(error) {
            errorWritableStream = error;
            return startPromise;
        },
        write: function(chunk) {
            assert_unreached('Unexpected write call');
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });

    startPromise.then(test8.step_func(function() {
        assert_equals(ws.state, 'writable', 'ws should start writable');

        rs.pipeTo(ws).catch(test8.step_func(function(e) {
            assert_equals(e, theError, 'pipeTo should reject with the passed error');
            assert_true(cancelCalled);
            test8.done();
        }));
        assert_equals(ws.state, 'writable', 'ws should be writable after pipe');

        errorWritableStream(theError);
        assert_equals(ws.state, 'errored', 'ws should be errored after erroring it');
    }));
});

var test9 = async_test('Piping from a non-empty ReadableStream to a WritableStream in the waiting state which becomes writable after a pipeTo call');
test9.step(function() {
    var pullCount = 0;
    var readyFulfilled = false;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('World');
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var resolveWritePromise;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            if (!resolveWritePromise) {
                assert_equals(chunk, 'Hello', 'the first chunk to arrive in write should be the first chunk written');
                return new Promise(test9.step_func(function(resolve) { resolveWritePromise = resolve; }));
            } else {
                assert_equals(chunk, 'World', 'the second chunk to arrive in write should be from the readable stream');
                assert_equals(pullCount, 1, 'the readable stream\'s pull should have been called once');
                assert_true(readyFulfilled);
                test9.done();
            }
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });
    ws.write('Hello');

    startPromise.then(test9.step_func(function() {
        assert_equals(ws.state, 'waiting', 'writable stream state should start waiting');

        rs.pipeTo(ws);
        assert_equals(ws.state, 'waiting', 'writable stream state should still be waiting immediately after piping');

        resolveWritePromise();
        ws.ready.then(test9.step_func(function() {
            assert_equals(ws.state, 'writable', 'writable stream should eventually become writable (when ready fulfills)');
            readyFulfilled = true;
        })).catch(test9.step_func(function(e) { assert_unreached(e); }));
    }));
});

var test10 = async_test('Piping from a non-empty ReadableStream to a WritableStream in waiting state which becomes errored after a pipeTo call');
test10.step(function() {
    var writeCalled = false;

    var pullCount = 0;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('World');
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_true(writeCalled);
            assert_equals(pullCount, 1);
            test10.done();
        }
    });

    var errorWritableStream;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function(error) {
            errorWritableStream = error;
            return startPromise;
        },
        write: function(chunk) {
            assert_false(writeCalled);
            assert_equals(chunk, 'Hello');
            writeCalled = true;
            return new Promise(test10.step_func(function() { }));
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });
    ws.write('Hello');

    startPromise.then(test10.step_func(function() {
        assert_equals(ws.state, 'waiting');

        rs.pipeTo(ws);
        assert_equals(ws.state, 'waiting');

        errorWritableStream();
        assert_equals(ws.state, 'errored');
    }));
});

var test11 = async_test('Piping from a non-empty ReadableStream which becomes errored after pipeTo call to a WritableStream in the waiting state');
test11.step(function() {
    var errorReadableStream;
    var pullCount = 0;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('World');
            errorReadableStream = c.error.bind(c);
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var writeCalled = false;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            assert_false(writeCalled);
            writeCalled = true;

            assert_equals(chunk, 'Hello');

            return new Promise(test11.step_func(function() { }));
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function() {
            test11.done('underlying source abort was called');
        }
    });
    ws.write('Hello');

    startPromise.then(test11.step_func(function() {
        assert_equals(ws.state, 'waiting');
        assert_equals(pullCount, 0);

        rs.pipeTo(ws);
        assert_equals(ws.state, 'waiting');

        errorReadableStream();
    }));
});

var test12 = async_test('Piping from a non-empty ReadableStream to a WritableStream in the waiting state where both become ready after a pipeTo');
test12.step(function() {
    var controller;
    var pullCount = 0;
    var rs = new ReadableStream({
        start: function(c) {
            controller = c;
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var writeCount = 0;
    var resolveWritePromise;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            ++writeCount;

            if (writeCount === 1) {
                assert_equals(chunk, 'Hello', 'first chunk written should equal the one passed to ws.write');
                return new Promise(test12.step_func(function(resolve) { resolveWritePromise = resolve; }));
            }
            if (writeCount === 2) {
                assert_equals(chunk, 'Goodbye', 'second chunk written should be from the source readable stream');
                test12.done();
            }
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function(reason) {
            assert_unreached('Unexpected abort call');
        }
    });
    ws.write('Hello');

    startPromise.then(test12.step_func(function() {
        assert_equals(writeCount, 1, 'exactly one write should have happened');
        assert_equals(ws.state, 'waiting', 'writable stream should be waiting');

        assert_equals(pullCount, 1, 'pull should have been called only once');
        rs.pipeTo(ws);

        controller.enqueue('Goodbye');

        // Check that nothing happens before calling resolveWritePromise(), and then call resolveWritePromise()
        // to check that pipeTo is woken up.
        assert_equals(pullCount, 1, 'after the pipeTo and enqueue, pull still should have been called only once');
        resolveWritePromise();
    }));
});

var test13 = async_test('Piping from an empty ReadableStream to a WritableStream in the waiting state which becomes writable after a pipeTo call');
test13.step(function() {
    var pullCount = 0;
    var rs = new ReadableStream({
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var resolveWritePromise;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            assert_true(!resolveWritePromise);
            assert_equals(chunk, 'Hello');
            return new Promise(test13.step_func(function(resolve) { resolveWritePromise = resolve; }));
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });
    ws.write('Hello');

    startPromise.then(test13.step_func(function() {
        assert_equals(ws.state, 'waiting');

        rs.pipeTo(ws);
        assert_equals(ws.state, 'waiting');
        assert_equals(pullCount, 1);

        resolveWritePromise();
        setTimeout(test13.step_func(function() {
            assert_equals(pullCount, 2);

            test13.done();
        }), 500);
    }));
});

var test14 = async_test('Piping from an empty ReadableStream which becomes closed after a pipeTo call to a WritableStream in the waiting state whose writes never complete');
test14.step(function() {
    var closeReadableStream;
    var pullCount = 0;
    var rs = new ReadableStream({
        start: function(c) {
            closeReadableStream = c.close.bind(c);
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var writeCalled = false;
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            if (!writeCalled) {
                assert_equals(chunk, 'Hello', 'the chunk should be written to the writable stream');
                writeCalled = true;
                closeReadableStream();
            } else {
                assert_unreached('Unexpected extra write call');
            }
            return new Promise(test14.step_func(function() { }));
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function() {
            assert_unreached('Unexpected abort call');
        }
    });
    ws.write('Hello');

    startPromise.then(test14.step_func(function() {
        assert_equals(ws.state, 'waiting', 'the writable stream should be in the waiting state after starting');

        rs.pipeTo(ws);

        setTimeout(test14.step_func(function() {
            assert_equals(ws.state, 'waiting', 'the writable stream should still be waiting since the write never completed');
            assert_equals(pullCount, 1, 'pull should have been called only once');
            assert_true(writeCalled);
            test14.done();
        }), 500);
    }));
});

var test15 = async_test('Piping from an empty ReadableStream which becomes errored after a pipeTo call to a WritableStream in the waiting state');
test15.step(function() {
    var errorReadableStream;
    var pullCount = 0;
    var rs = new ReadableStream({
        start: function(c) {
            errorReadableStream = c.error.bind(c);
        },
        pull: function() {
            ++pullCount;
        },
        cancel: function() {
            assert_unreached('Unexpected cancel call');
        }
    });

    var writeCalled = false;
    var passedError = new Error('horrible things');
    var startPromise = Promise.resolve();
    var ws = new WritableStream({
        start: function() {
            return startPromise;
        },
        write: function(chunk) {
            if (!writeCalled) {
                assert_equals(chunk, 'Hello');
                writeCalled = true;
            } else {
                assert_unreached('Unexpected extra write call');
            }
            return new Promise(test15.step_func(function() { }));
        },
        close: function() {
            assert_unreached('Unexpected close call');
        },
        abort: function(reason) {
            assert_equals(reason, passedError);
            assert_true(writeCalled);
            assert_equals(pullCount, 1);
            test15.done();
        }
    });
    ws.write('Hello');

    startPromise.then(test15.step_func(function() {
        assert_equals(ws.state, 'waiting');

        rs.pipeTo(ws);

        errorReadableStream(passedError);
    }));
});

var test16 = async_test('Piping to a duck-typed asynchronous "writable stream" works');
test16.step(function() {
    var rs = sequentialReadableStream(5, { async: true });

    var chunksWritten = [];
    var dest = {
        state: 'writable',
        write: function(chunk) {
            chunksWritten.push(chunk);
            return Promise.resolve();
        },
        get ready() {
            return Promise.resolve();
        },
        close: function() {
            assert_array_equals(chunksWritten, [1, 2, 3, 4, 5]);
            test16.done();
            return Promise.resolve();
        },
        abort: function() {
            assert_unreached('Should not call abort');
        },
        closed: new Promise(test16.step_func(function() { }))
    };

    rs.pipeTo(dest);
});

var test17 = async_test('Piping to a stream that has been aborted passes through the error as the cancellation reason');
test17.step(function() {
    var recordedReason;
    var rs = new ReadableStream({
        cancel: function(reason) {
            recordedReason = reason;
        }
    });

    var ws = new WritableStream();
    var passedReason = new Error('I don\'t like you.');
    ws.abort(passedReason);

    rs.pipeTo(ws).catch(test17.step_func(function(e) {
        assert_equals(e, passedReason, 'pipeTo rejection reason should be the cancellation reason');
        assert_equals(recordedReason, passedReason, 'the recorded cancellation reason must be the passed abort reason');
        test17.done();
    }));
});

var test18 = async_test('Piping to a stream and then aborting it passes through the error as the cancellation reason');
test18.step(function() {
    var recordedReason;
    var rs = new ReadableStream({
        cancel: function(reason) {
            recordedReason = reason;
        }
    });

    var ws = new WritableStream();
    var passedReason = new Error('I don\'t like you.');

    var pipeToPromise = rs.pipeTo(ws);
    ws.abort(passedReason);

    pipeToPromise.catch(test18.step_func(function(e) {
        assert_equals(e, passedReason, 'pipeTo rejection reason should be the abortion reason');
        assert_equals(recordedReason, passedReason, 'the recorded cancellation reason must be the passed abort reason');
        test18.done();
    }));
});

var test19 = async_test('Piping to a stream that has been closed propagates a TypeError cancellation reason backward');
test19.step(function() {
    var recordedReason;
    var rs = new ReadableStream({
        cancel: function(reason) {
            recordedReason = reason;
        }
    });

    var ws = new WritableStream();
    ws.close();

    rs.pipeTo(ws).catch(test19.step_func(function(e) {
        assert_throws(new TypeError(), function() { throw e; }, 'the rejection reason for the pipeTo promise should be a TypeError');
        assert_throws(new TypeError(), function() { throw recordedReason; }, 'the recorded cancellation reason should be a TypeError');
        test19.done();
    }));
});

var test20 = async_test('Piping to a stream and then closing it propagates a TypeError cancellation reason backward');
test20.step(function() {
    var recordedReason;
    var rs = new ReadableStream({
        cancel: function(reason) {
            recordedReason = reason;
        }
    });

    var ws = new WritableStream();

    var pipeToPromise = rs.pipeTo(ws);
    ws.close();

    pipeToPromise.catch(test20.step_func(function(e) {
        assert_throws(new TypeError(), function() { throw e; }, 'the rejection reason for the pipeTo promise should be a TypeError');
        assert_throws(new TypeError(), function() { throw recordedReason; }, 'the recorded cancellation reason should be a TypeError');
        test20.done();
    }));
});

var test21 = async_test('Piping to a stream that errors on write should pass through the error as the cancellation reason');
test21.step(function() {
    var recordedReason;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('a');
            c.enqueue('b');
            c.enqueue('c');
        },
        cancel: function(reason) {
            assert_equals(reason, passedError, 'the recorded cancellation reason must be the passed error');
            test21.done();
        }
    });

    var written = 0;
    var passedError = new Error('I don\'t like you.');
    var ws = new WritableStream({
        write: function(chunk) {
            return new Promise(test21.step_func(function(resolve, reject) {
                if (++written > 1) {
                    reject(passedError);
                } else {
                    resolve();
                }
            }));
        }
    });

    rs.pipeTo(ws);
});

var test22 = async_test('Piping to a stream that errors on write should not pass through the error if the stream is already closed');
test22.step(function() {
    var cancelCalled = false;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('a');
            c.enqueue('b');
            c.enqueue('c');
            c.close();
        },
        cancel: function() {
            cancelCalled = true;
        }
    });

    var written = 0;
    var passedError = new Error('I don\'t like you.');
    var ws = new WritableStream({
        write: function(chunk) {
            return new Promise(test22.step_func(function(resolve, reject) {
                if (++written > 1) {
                    reject(passedError);
                } else {
                    resolve();
                }
            }));
        }
    });

    rs.pipeTo(ws).then(
        test22.step_func(function() { assert_unreached('pipeTo should not fulfill'); }),
        test22.step_func(function(r) {
            assert_equals(r, passedError, 'pipeTo should reject with the same error as the write');
            assert_equals(cancelCalled, false, 'cancel should not have been called');
            test22.done();
        })
    );
});

var test23 = async_test('Piping to a stream that errors soon after writing should pass through the error as the cancellation reason');
test23.step(function() {
    var recordedReason;
    var rs = new ReadableStream({
        start: function(c) {
            c.enqueue('a');
            c.enqueue('b');
            c.enqueue('c');
        },
        cancel: function(reason) {
            assert_equals(reason, passedError, 'the recorded cancellation reason must be the passed error');
            test23.done();
        }
    });

    var written = 0;
    var passedError = new Error('I don\'t like you.');
    var ws = new WritableStream({
        write: function(chunk) {
            return new Promise(test23.step_func(function(resolve, reject) {
                if (++written > 1) {
                    setTimeout(test23.step_func(function() { reject(passedError); }), 200);
                } else {
                    resolve();
                }
            }));
        }
    });

    rs.pipeTo(ws);
});

// This test is alone under streams/pipe-to.html for timing reasons.
// var test24 = async_test('Piping to a writable stream that does not consume the writes fast enough exerts backpressure on the source');
</script>