Re: [streams] Operation stream (#287)

Thanks Domenic.

This PR includes two big proposals:

- merge of underlying sink/source, writable/readable stream into OperationStream
- precise flow control stuff

----

`WritableOperationStream` is based on a little modified version of the `WritableStream`. It has `.write()` that returns an object that represents a value that becomes available in the future. I used `OperationStatus` and `Operation` instead of a promise that will be fulfilled on arrival of the written value to the sink experimentally, but this can be simplified to return a promise.

```
const p = wos.write();
...
const op = ros.read();
op.complete();
...
at write side, we see p gets fulfilled.
```

And, `ReadableOperationStream` is based on `ReadableStream` but modified to have power to represent acknowledgement to the read data. With this change, ReadableOperationStream gained similar surface as the `underlying sink` concept for the `WritableStream`.

The motivation behind this was that:

- the async read() approach we wanted to add for supporting ReadFile/fread looked similar to `WritableStream`. I thought we may reuse it for implementing async read().
- I thought the buffer pool approach explored in the issue initiated by Ben could be realized by giving the reading API to have acknowledgement functionality. I.e. making ack mean return of the obtained ArrayBuffer. This lead to ReadableStream to ReadableOperationStream change of addition of `.complete()` and `.error()` to the return value (Operation object).
    - this resembles the `underlying sink` concept. underlying sink's write() accepts value from the queue of the WriteableStream, processes it and returns a promise to acknowledge it. It'll propagated to the original `write()` call on the WritableStream and fulfills/rejects the returned promise.
- in thinking about the dual for the async read() API (which is realized as WritableOperationStream in this PR), I found that ReadableOperationStream (see `pipeOperationStreams()`) can be easily connected to.

Based on the observations above, I tried to merge underlying source/sink, ReadableStream and WritableStream into two surfaces (`WritableOperationStream` and `ReadableOperationStream`) but backed by a single object named `OperationStream`. The stream creation methods on FakeFileBackedByteSource and BytesSetToOneExpectingByteSink explain how the API is used for construction of a stream object to give to the user. Either of WritableOperationStream or ReadableOperationStream is given to the object that processes actions to the I/O and the other is returned to the user. Unlike the current WritableStream/ReadableStream constructor that take UnderlyingSource/Sink object, the underlying source/sink sees the same API as the user sees and accesses it using promises. `_loop()` methods are optimized by having a single loop to forward output from the I/O to the stream synchronously. So, it may look more complicated (also due to the addition of the OperationStatus object instead of a promi
 se). But
  it can be simplified if the users are fine with a little less efficiency.

----

I also designed and implemented some precise flow control stuff using this opportunity. Sorry for confusing but they're independent parts.

- get space()
- waitForSpace()
- get window()
- set window()

----

> Based on the the tests, these look really complicated to create. How are they to use? How big is the public API for most authors? How does this map to the three classes of authors I outlined in #253 (comment) ? Do they interop with the existing API at all?

- (1): Obtains a ReadableOperationStream from a source, and calls pipeOperationStreams in the simplest case. Or use read(), state and ready on ReadableOperationStream manually. If the source is allocate-and-forget type, the read()'s complete() and error() don't need to be invoked. GC takes care of the buffers' lifetime.
- (2) and (3): Use automatic buffer allocation described by FakeFileBackedByteSource.createBufferProducingStreamWithPool(). Or use bring-your-own-buffer type reading described by FakeFileBackedByteSource.createBufferFillingStream().
    - FakeFileBackedByteSource.createBufferProducingStreamWithPool() returns a ReadableOperationStream that gives us allocated buffers filled with the contents read from the encapsulated source I/O together with acknowledgement functions `.complete()` and `.error()`. The acknowledgement is propagated to the code behind the FakeFileBackedByteSource instance and used to ok-to-return-to-the-pool signal.
    - FakeFileBackedByteSource.createBufferFillingStream() returns a WritableOperationStream that accepts buffers, fills them and returns (or just signals completion) via the returned OperationStatus. This is almost equivalent to your async read() API.


---
Reply to this email directly or view it on GitHub:
https://github.com/whatwg/streams/pull/287#issuecomment-75882613