We are in the process of transitioning this specification from a GitHub README into something a bit more palatable. The official-lookin' version is developed in the official-lookin
branch's index.html
file, which you can see on GitHub, or in its rendered glory at this long URL.
Right now, we've transferred over most of the concepts and text, but none of the algorithms or APIs. We'll be iterating on the APIs a bit more here, in Markdown format, until we feel confident in them. In the meantime, please check out the rendered spec for all of the interesting stage-setting text.
By the way, this transition is being tracked as #62.
class BaseReadableStream {
constructor({
function start = () => {},
function pull = () => {},
function cancel = () => {}
})
// Reading data from the underlying source
any read()
Promise<undefined> wait()
get ReadableStreamState state
// Composing with other streams
WritableStream pipeTo(WritableStream dest, { ToBoolean close = true } = {})
ReadableStream pipeThrough({ WritableStream in, ReadableStream out }, options)
// Stop accumulating data
void cancel(any reason)
// Useful helper
get Promise<undefined> closed
// Internal properties
Array [[buffer]] = []
boolean [[started]] = false
boolean [[draining]] = false
boolean [[pulling]] = false
string [[state]] = "waiting"
any [[storedError]]
Promise<undefined> [[waitPromise]]
Promise<undefined> [[closedPromise]]
Promise [[startedPromise]]
function [[onCancel]]
function [[onPull]]
// Internal methods for use by the underlying source
[[push]](any data)
[[close]]()
[[error]](any e)
// Other internal helper methods
[[callPull]]()
}
enum ReadableStreamState {
"readable" // the buffer has something in it; read at will
"waiting" // the source is not ready or the buffer is empty; you should call wait
"closed" // all data has been read from both the source and the buffer
"errored" // the source errored so the stream is now dead
}
The constructor is passed several functions, all optional:
start(push, close, error)
is typically used to adapt a push-based data source, as it is called immediately so it can set up any relevant event listeners, or to acquire access to a pull-based data source.pull(push, close, error)
is typically used to adapt a pull-based data source, as it is called in reaction toread
calls, or to start the flow of data in push-based data sources. Once it is called, it will not be called again until its passedpush
function is called.cancel()
is called when the readable stream is canceled, and should perform whatever source-specific steps are necessary to clean up and stop reading.
Both start
and pull
are given the ability to manipulate the stream's internal buffer and state by being passed the this.[[push]]
, this.[[close]]
, and this.[[error]]
functions.
- If IsCallable(start) is false, throw a TypeError exception.
- If IsCallable(pull) is false, throw a TypeError exception.
- If IsCallable(cancel) is false, throw a TypeError exception.
- Set
this.[[onCancel]]
tocancel
. - Set
this.[[onPull]]
topull
. - Let
this.[[waitPromise]]
be a newly-created pending promise. - Let
this.[[closedPromise]]
be a newly-created pending promise. - Let startResult be the result of
start(this.[[push]], this.[[close]], this.[[error]])
. - ReturnIfAbrupt(startResult).
- Let
this.[[startedPromise]]
be the result of casting startResult to a promise. - Upon fulfillment of
this.[[startedPromise]]
, setthis.[[started]]
totrue
. - Upon rejection of
this.[[startedPromise]]
with reasonr
, callthis.[[error]](r)
.
- Return
this.[[state]]
.
- If
this.[[state]]
is"waiting"
or"closed"
, throw aTypeError
exception. - If
this.[[state]]
is"errored"
, throwthis.[[storedError]]
. - Assert:
this.[[state]]
is"readable"
. - Assert:
this.[[buffer]]
is not empty. - Let
data
be the result of shifting an element off of the front ofthis.[[buffer]]
. - If
this.[[buffer]]
is now empty,- If
this.[[draining]]
istrue
,- Set
this.[[state]]
to"closed"
. - Let
this.[[waitPromise]]
be a newly-created promise rejected with aTypeError
exception. - Resolve
this.[[closedPromise]]
withundefined
.
- Set
- If
this.[[draining]]
isfalse
,- Set
this.[[state]]
to"waiting"
. - Let
this.[[waitPromise]]
be a newly-created pending promise. - Call
this.[[callPull]]()
.
- Set
- If
- Return
data
.
- If
this.[[state]]
is"waiting"
,- Call
this.[[callPull]]()
.
- Call
- Return
this.[[waitPromise]]
.
- If
this.[[state]]
is"closed"
, return a new promise resolved with undefined. - If
this.[[state]]
is"errored"
, return a new promise rejected withthis.[[storedError]]
. - If
this.[[state]]
is"waiting"
, resolvethis.[[waitPromise]]
with undefined. - If
this.[[state]]
is"readable"
, letthis.[[waitPromise]]
be a new promise resolved with undefined. - Clear
this.[[buffer]]
. - Set
this.[[state]]
to"closed"
. - Resolve
this.[[closedPromise]]
with undefined. - Return the result of promise-calling
this.[[onCancel]]()
.
- Return
this.[[closedPromise]]
.
BaseReadableStream.prototype.pipeTo = (dest, { close = true } = {}) => {
const source = this;
close = Boolean(close);
fillDest();
return dest;
function fillDest() {
if (dest.state === "writable") {
pumpSource();
} else if (dest.state === "waiting") {
dest.wait().then(fillDest, cancelSource);
} else {
// Source has either been closed by someone else, or has errored in the course of
// someone else writing. Either way, we're not going to be able to do anything
// else useful.
cancelSource();
}
}
function pumpSource() {
if (source.state === "readable") {
dest.write(source.read()).catch(cancelSource);
fillDest();
} else if (source.state === "waiting") {
source.wait().then(fillDest, abortDest);
} else if (source.state === "closed") {
closeDest();
} else {
abortDest();
}
}
function cancelSource(reason) {
source.cancel(reason);
}
function closeDest() {
if (close) {
dest.close();
}
}
function abortDest(reason) {
// ISSUE: should this be preventable via an option or via `options.close`?
dest.abort(reason);
}
};
- If Type(input) is not Object, then throw a TypeError exception.
- If Type(output) is not Object, then throw a TypeError exception.
- Let stream be the this value.
- Let result be the result of calling Invoke(stream,
"pipeTo"
, (options)). - ReturnIfAbrupt(result).
- Return output.
- If
this.[[state]]
is"waiting"
,- Push
data
ontothis.[[buffer]]
. - Set
this.[[pulling]]
tofalse
. - Set
this.[[state]]
to"readable"
. - Resolve
this.[[waitPromise]]
withundefined
.
- Push
- If
this.[[state]]
is"readable"
,- Push
data
ontothis.[[buffer]]
. - Set
this.[[pulling]]
tofalse
.
- Push
- Return
false
.
- If
this.[[state]]
is"waiting"
,- Resolve
this.[[waitPromise]]
withundefined
. - Resolve
this.[[closedPromise]]
withundefined
. - Set
this.[[state]]
to"closed"
.
- Resolve
- If
this.[[state]]
is"readable"
,- Set
this.[[draining]]
totrue
.
- Set
- If
this.[[state]]
is"waiting"
,- Set
this.[[state]]
to"errored"
. - Set
this.[[storedError]]
toe
. - Reject
this.[[waitPromise]]
withe
. - Reject
this.[[closedPromise]]
withe
.
- Set
- If
this.[[state]]
is"readable"
,- Clear
this.[[buffer]]
. - Set
this.[[state]]
to"errored"
. - Set
this.[[storedError]]
toe
. - Let
this.[[waitPromise]]
be a newly-created promise object rejected withe
. - Reject
this.[[closedPromise]]
withe
.
- Clear
- If
this.[[pulling]]
istrue
, return. - Set
this.[[pulling]]
totrue
. - If
this.[[started]]
isfalse
,- Upon fulfillment of
this.[[startedPromise]]
,- Let
pullResult
be the result ofthis.[[onPull]](this.[[push]], this.[[close]], this.[[error]])
. - If
pullResult
is an abrupt completion, callthis.[[error]](pullResult.[[value]])
.
- Let
- Upon fulfillment of
- If
this.[[started]]
istrue
,- Let
pullResult
be the result ofthis.[[onPull]](this.[[push]], this.[[close]], this.[[error]])
. - If
pullResult
is an abrupt completion, callthis.[[error]](pullResult.[[value]])
.
- Let
class ReadableStream extends BaseReadableStream {
// Adds a backpressure strategy argument.
constructor({
function start = () => {},
function pull = () => {},
function cancel = () => {},
strategy: { function count, function needsMoreData }
})
// Overriden to do bookkeeping for the backpressure strategy
any read()
// Supports multi-pipe by default, overriding base behavior.
WritableStream pipeTo(WritableStream dest, { ToBoolean close = true } = {})
// Overriden to take into account the backpressure strategy.
// You can also think of this as part of the constructor override, i.e. it passes
// in a different function to `start` and `pull`.
[[push]](data)
// Internal properties
[[tee]]
[[strategy]]
[[bufferSize]] = 0
}
- Set
this.[[strategy]]
tostrategy
. - Call
super({ start, pull, cancel })
.
- Let
data
besuper()
. - Subtract
this.[[strategy]].count(data)
fromthis.[[bufferSize]]
. - Return
data
.
- Let
alreadyPiping
betrue
. - If
this.[[tee]]
isundefined
, letthis.[[tee]]
be a newTeeStream
and setalreadyPiping
tofalse
. - Call
this.[[tee]].addOut(dest, { close })
. - If
alreadyPiping
isfalse
, callsuper(this.[[tee]], { close: true })
. - Return
dest
.
- Call
BaseReadableStream
's version ofthis.[[push]](data)
. - If
this.[[state]]
is now"readable"
,- Add
this.[[strategy]].count(data)
tothis.[[bufferSize]]
. - Return
this.[[strategy]].needsMoreData(this.[[bufferSize]])
.
- Add
- Return
false
.
class BaseWritableStream {
constructor({
function start = () => {},
function write = () => {},
function close = () => {},
function abort = close
})
// Writing data to the underlying sink
Promise<undefined> write(any data)
Promise<undefined> wait()
get WritableStreamState state
// Close off the underlying sink gracefully; we are done.
Promise<undefined> close()
// Close off the underlying sink forcefully; everything written so far is suspect.
Promise<undefined> abort(any reason)
// Useful helpers
get Promise<undefined> closed
// Internal methods
[[error]](any e)
[[advanceBuffer]]()
[[doClose]]()
[[doNextWrite]]({ type, promise, data })
// Internal properties
Array [[buffer]] = []
string [[state]] = "waiting"
any [[storedError]]
Promise<undefined> [[currentWritePromise]]
Promise<undefined> [[writablePromise]]
Promise<undefined> [[closedPromise]]
function [[onWrite]]
function [[onClose]]
function [[onAbort]]
}
enum WritableStreamState {
"writable" // the sink is ready and the buffer is not yet full; write at will
"waiting" // the sink is not ready or the buffer is full; you should call wait
"closing" // the sink is being closed; no more writing
"closed" // the sink has been closed
"errored" // the sink errored so the stream is now dead
}
The constructor is passed several functions, all optional:
start()
is called when the writable stream is created, and should open the underlying writable sink. If this process is asynchronous, it can return a promise to signal success or failure.write(data, done, error)
should writedata
to the underlying sink. It can call itsdone
orerror
parameters, either synchronously or asynchronously, to respectively signal that the underlying resource is ready for more data or that an error occurred writing. The stream implementation guarantees that this function will be called only after previous writes have succeeded (i.e. called theirdone
parameter), and never afterclose
orabort
is called.close()
should close the underlying sink. If this process is asynchronous, it can return a promise to signal success or failure. The stream implementation guarantees that this function will be called only after all queued-up writes have succeeded.abort()
is an abrupt close, signaling that all data written so far is suspect. It should clean up underlying resources, much likeclose
, but perhaps with some custom handling. Unlikeclose
,abort
will be called even if writes are queued up, throwing away that data.
In reaction to calls to the stream's .write()
method, the write
constructor option is given data from the internal buffer, along with the means to signal that the data has been successfully or unsuccessfully written.
- If IsCallable(start) is false, throw a TypeError exception.
- If IsCallable(write) is false, throw a TypeError exception.
- If IsCallable(close) is false, throw a TypeError exception.
- If IsCallable(abort) is false, throw a TypeError exception.
- Set
this.[[onWrite]]
towrite
. - Set
this.[[onClose]]
toclose
. - Set
this.[[onAbort]]
toabort
. - Let
this.[[writablePromise]]
be a newly-created pending promise. - Let
this.[[closedPromise]]
be a newly-created pending promise. - Call
start()
and letstartedPromise
be the result of casting the return value to a promise. - When/if
startedPromise
is fulfilled, callthis.[[advanceBuffer]]()
. - When/if
startedPromise
is rejected with reasonr
, callthis.[[error]](r)
.
- Return
this.[[closedPromise]]
.
- Return
this.[[state]]
.
- If
this.[[state]]
is"writable"
,- Set
this.[[state]]
to"waiting"
. - Set
this.[[writablePromise]]
to be a newly-created pending promise. - Let
promise
be a newly-created pending promise. - Call
this.[[doNextWrite]]({ type: "data", promise, data })
. - Return
promise
.
- Set
- If
this.[[state]]
is"waiting"
,- Let
promise
be a newly-created pending promise. - Push
{ type: "data", promise, data }
ontothis.[[buffer]]
. - Return
promise
.
- Let
- If
this.[[state]]
is"closing"
or"closed"
,- Return a promise rejected with a
TypeError
exception.
- Return a promise rejected with a
- If
this.[[state]]
is"errored"
,- Return a promise rejected with
this.[[storedError]]
.
- Return a promise rejected with
- If
this.[[state]]
is"writable"
,- Set
this.[[state]]
to"closing"
. - Call
this.[[doClose]]()
. - Return
this.[[closedPromise]]
.
- Set
- If
this.[[state]]
is"waiting"
,- Set
this.[[state]]
to"closing"
. - Push
{ type: "close", promise: undefined, data: undefined }
ontothis.[[buffer]]
. - Return
this.[[closedPromise]]
.
- Set
- If
this.[[state]]
is"closing"
or"closed"
,- Return a promise rejected with a
TypeError
exception.
- Return a promise rejected with a
- If
this.[[state]]
is"errored"
,- Return a promise rejected with
this.[[storedError]]
.
- Return a promise rejected with
- If
this.[[state]]
is"closed"
, return a new promise resolved with undefined. - If
this.[[state]]
is"errored"
, return a new promise rejected withthis.[[storedError]]
. - Call
this.[[error]](reason)
. - Return the result of promise-calling
this.[[onAbort]]()
.
- Return
this.[[writablePromise]]
.
- If
this.[[state]]
is"closed"
or"errored"
, return. - For each entry
{ type, promise, data }
inthis.[[buffer]]
, rejectpromise
withr
. - Clear
this.[[buffer]]
. - Set
this.[[state]]
to"errored"
. - Set
this.[[storedError]]
toe
. - Reject
this.[[writablePromise]]
withe
. - Reject
this.[[closedPromise]]
withe
.
- If
this.[[buffer]]
is not empty,- Shift
entry
off ofthis.[[buffer]]
. - Call
this.[[doNextWrite]](entry)
.
- Shift
- If
this.[[buffer]]
is empty,- Set
this.[[state]]
to"writable"
. - Resolve
this.[[writablePromise]]
withundefined
.
- Set
- Reject
this.[[writablePromise]]
with a TypeError exception. - Let closePromise be the result of promise-calling
this.[[onClose]]()
. - Upon fulfillment of closePromise,
- Set
this.[[state]]
to"closed"
. - Resolve
this.[[closedPromise]]
with undefined.
- Set
- Upon rejection of closePromise with reason r,
- Call
this.[[error]](r)
.
- Call
- If
type
is"close"
,- Assert:
this.[[state]]
is"closing"
. - Call
this.[[doClose]]()
. - Return.
- Assert:
- Assert:
type
must be"data"
. - Set
this.[[currentWritePromise]]
topromise
. - Let
signalDone
be a new function of zero arguments, closing overthis
andpromise
, that performs the following steps:- If
this.[[currentWritePromise]]
is notpromise
, return. - Set
this.[[currentWritePromise]]
toundefined
. - If
this.[[state]]
is"waiting"
,- Resolve
promise
withundefined
. - Call
this.[[advanceBuffer]]()
.
- Resolve
- If
this.[[state]]
is"closing"
,- Resolve
promise
withundefined
. - If
this.[[buffer]]
is not empty,- Shift
entry
off ofthis.[[buffer]]
. - Call
this.[[doNextWrite]](entry)
.
- Shift
- Resolve
- If
- Call
this.[[onWrite]](data, signalDone, [[error]])
. - If the call throws an exception
e
, callthis.[[error]](e)
.
Note: if the constructor's write
option calls done
more than once, or after calling error
, or after the stream has been aborted, then signalDone
ends up doing nothing.
class WritableStream extends BaseWritableStream {
// Adds a backpressure strategy argument.
constructor({
function start = () => {},
function write = () => {},
function close = () => {},
function abort = close,
strategy: { function count, function needsMoreData }
})
// Overriden to take into account backpressure strategy
Promise<undefined> write(data)
// Overriden to take into account backpressure strategy.
// You can also think of this as part of the the constructor and write override.
[[doNextWrite]]({ type, promise, data })
// Internal properties
[[strategy]]
[[bufferSize]] = 0
}
- Set
this.[[strategy]]
tostrategy
. - Call
super({ start, write, close, abort })
.
- If
this.[[state]]
is"writable"
or"waiting"
,- Add
this.[[strategy]].count(data)
tothis.[[bufferSize]]
.
- Add
- If
this.[[state]]
is"writable"
,- Let
promise
be a newly-created pending promise. - If
ToBoolean(this.[[strategy]].needsMoreData(this.[[bufferSize]]))
isfalse
,- Set
this.[[state]]
to"waiting"
. - Set
this.[[writablePromise]]
to be a newly-created pending promise.
- Set
- If
this.[[buffer]]
is empty,- Call
this.[[doNextWrite]]({ type: "data", promise, data })
.
- Call
- Otherwise,
- Push
{ type: "data", promise, data }
ontothis.[[buffer]]
.
- Push
- Return
promise
.
- Let
- Return
super(data)
.
- Subtract
this.[[strategy]].count(data)
fromthis.[[bufferSize]]
. - Return the result of calling
BaseWritableStream
's version ofthis.[[doNextWrite]]({ type, promise, data })
.
A "tee stream" is a writable stream which, when written to, itself writes to multiple destinations. It aggregates backpressure and abort signals from those destinations, propagating the appropriate aggregate signals backward.
class TeeStream extends BaseWritableStream {
constructor() {
this.[[outputs]] = [];
super({
write(data) {
return Promise.all(this.[[outputs]].map(o => o.dest.write(data)));
},
close() {
const outputsToClose = this.[[outputs]].filter(o => o.close);
return Promise.all(outputsToClose.map(o => o.dest.write(data)));
},
abort(reason) {
return Promise.all(this.[[outputs]].map(o => o.dest.abort(reason)));
}
});
}
addOut(dest, { close = true } = {}) {
this.[[outputs]].push({ dest, close });
}
}
A common buffering strategy when dealing with binary or string data is to wait until the accumulated length
properties of the incoming data reaches a specified highWaterMark
. As such, this is provided as a built-in helper along with the stream APIs.
class LengthBufferingStrategy {
constructor({ highWaterMark }) {
this.highWaterMark = Number(highWaterMark);
if (Number.isNaN(this.highWaterMark) || this.highWaterMark < 0) {
throw new RangeError("highWaterMark must be a nonnegative number.");
}
}
count(chunk) {
return chunk.length;
}
needsMoreData(bufferSize) {
return bufferSize < this.highWaterMark;
}
}
A common buffering strategy when dealing with object streams is to simply count the number of objects that have been accumulated so far, waiting until this number reaches a specified highWaterMark
. As such, this strategy is also provided as a built-in helper.
class CountBufferingStrategy {
constructor({ highWaterMark }) {
this.highWaterMark = Number(highWaterMark);
if (Number.isNaN(this.highWaterMark) || this.highWaterMark < 0) {
throw new RangeError("highWaterMark must be a nonnegative number.");
}
}
count(chunk) {
return 1;
}
needsMoreData(bufferSize) {
return bufferSize < this.highWaterMark;
}
}