Description
Principle: cloning a stream should be a no-op (except that queue size increases by one chunk).
Cloning formulas
"Cloning" is an operation which locks the original stream and constructs a new stream that behaves just like it. It is a useful building block for stream manipulations and it is also used in the construction of transferable streams.
Cloning a readable
const rs2 = rs1.pipeThrough(new TransformStream());
Cloning a writable
const internalTS = new TransformStream();
internalTS.readable.pipeTo(ws1);
ws2 = internalTS.writable;
Problem
The cloned readable stream rs2
preserves the properties of the original stream rs1
.
However, the cloned writable stream ws2
loses information.
If you do readable.pipeTo(ws1)
then the promise returned by pipeTo()
will not resolve until underlyingSink.close()
has resolved. However, if you do readable.pipeTo(ws2)
, pipeTo()
will resolve as soon as readable
is closed.
This problem was identified while working on the tests for transferable streams.
Cause
An underlying source is not notified when downstream has finished consuming the data. Once the ReadableStream is closed, it gets no further notification about the state of the pipe. pipeTo()
waits for all writes to complete by design, but this is independent of the readable stream's main state machine.
Proposed fix
ReadableStream's underlyingSource will gain an additional method, finally()
, which is called once the stream is completely quiescent, ie. after it is CLOSED
or ERRORED
and no operations are pending. See #636 for background and other use cases for the finally()
method. There will be a new abstract operation, ReadableStreamMakeFinallyWaitFor(promise)
which will delay the call to finally()
until the passed-in promise is settled. This new operation could in principle be exposed on the ReadableStreamDefaultReader
object, but isn't because of its limited utility. Instead, we keep it as an internal abstract operation for now, only used by the pipeTo implementation.
The finally()
method will be passed a TBD argument indicating whether the ReadableStream closed without error. If the promise passed to ReadableStreamMakeFinallyWaitFor
is rejected, then this argument will indicate an error occurred.
CreateReadableStream()
will gain an extra operation argument corresponding to the finally()
method. This will be used by the TransformStream implementation to control the timing of changing the state of the writable side to CLOSED
or ERRORED
, which in turn will change the timing of when a pipe to the writable side completes.
The new effect is that a pipe to ws2
will close no sooner than close()
or abort()
on ws1
's underylingSink complete.