Revert back to internal APIs to implement tee(), pipeTo(), async iterator and TransformStream

[MattiasBuelens]

In #98, #99, #100 and #105, I rewrote large parts of the polyfill to use the public getReader() and getWriter() APIs for most methods and classes, rather than relying on internal APIs such as AcquireReadableStreamDefaultReader. The goal was to allow the polyfill to interoperate directly with native stream objects, such as a ReadableStream from Response.body.

Unfortunately, I now believe this is not going to work:

• When the polyfill calls getReader() through the public API, it still expects a reader that complies with the same version of the streams specification that the polyfill itself implements. However, if the specification makes changes to the reader's behavior (such as in Reject pending reads when releasing reader whatwg/streams#1168), then the polyfill has no way of knowing whether a given native reader implements this change or not (without resorting to browser sniffing).
• Making the polyfill pass all Web Platform Tests turns out to be pretty difficult if you cannot use internal APIs. I've had to deal with a bunch of timing issues, and in some cases I still ended up resorted to using an internal API anyway.
• Testing the actual interoperability would be difficult. We'd need modified versions of the Web Platform Tests that constructed a mix of polyfilled and native streams, to see if all combinations still pass all tests.

Instead, I think a better interoperability story would be to provide dedicated conversion methods, such as ReadableStream.from(nativeStream). Users would need to manually convert streams "on the boundary", but at least they can expect predictable results afterwards from those converted streams.

This PR reverts #98, #99, #100 and #105. We revert to (pretty much) the state of f43be79, see f43be79...8bdccbe.

[rattrayalex]

Hi @MattiasBuelens – in absence of your proposed ReadableStream.from(nativeStream), is there a good way to get an async iterable from the browser's Response.body (either using this library or otherwise)?

EDIT: should this work?

    const body = response.body;
    if (!body[Symbol.asyncIterator]) {
      body[Symbol.asyncIterator] = async function* () {
        const reader = (body as any as ReadableStream).getReader();
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          yield value;
        }
      };
    }

[jimmywarting]

Maybe this will do:

ReadableStream.prototype[Symbol.asyncIterator] ??= function () {
  const reader = this.getReader()
  return {
    next () {
      return reader.read()
    },
    return () {
      return reader.releaseLock()
    },
    throw (err) {
      this.return()
      throw err
    },
    [Symbol.asyncIterator] () {
      return this
    }
  }
}

[rattrayalex]

Thank you @jimmywarting ! That looks great.

Is it important to release the readers' locks if the stream is canceled as well? (asking because I saw that here: https://gist.github.com/dy/8ca154c96e2b2a823c6501d29972b8a8)

[MattiasBuelens]

No, that's not necessary. While the async iterator has the reader lock, you can't call stream.cancel() anyway.

Note that the Streams standard by default will cancel the stream when returning from the async iterator. If you want to match that, you can add an await reader.cancel() in the return() method of @jimmywarting 's implementation.

Also, you don't need the throw() method. A for await of loop will never call it, so Streams also doesn't define it. 😉

[rattrayalex]

Amazing, thank you both so much.

For posterity, here's what we ended up with:

function readableStreamAsyncIterable<T>(stream: any): AsyncIterableIterator<T> {
  if (stream[Symbol.asyncIterator]) {
    return stream;
  }

  const reader = stream.getReader();

  return {
    next() {
      return reader.read();
    },
    async return() {
      reader.cancel();
      reader.releaseLock();
      return { done: true, value: undefined };
    },
    [Symbol.asyncIterator]() {
      return this;
    },
  };
}

async function usage() {
    const response = await fetch(…);
    const iter = readableStreamAsyncIterable<Bytes>(response.body);
    for await (const chunk of iter) {
      // …
    }
}

(We decided not to polyfill/monkeypatch in our scenario, but I imagine ReadableStream.prototype[Symbol.asyncIterator] ??= would work for others if needed)

[rattrayalex]

I wonder if it might be worth exposing a polyfill that just does this? Would you like a PR?

(context in case you're curious: openai/openai-node#182 (comment))

[jedwards1211]

The spec/MDN docs make it seem like you'd need to releaseLock() before calling reader.cancel(), since it "behaves the same as stream.cancel()", which rejects if the stream is locked. But this is untrue and reader.releaseLock() actually causes the subsequent reader.cancel() to reject.

Reading the spec more carefully, reader.cancel() cancels the stream without checking if it's locked first (unlike stream.cancel()).

[MattiasBuelens]

The intended order is reader.cancel(); reader.releaseLock();. The other way around doesn't work.

And while reader.releaseLock(); stream.cancel(); would also work, it's less idiomatic in my opinion. Ideally, you should only release the lock when you're done using the stream.

If you want to be as close as possible to the specced behavior, you can use this:

function readableStreamAsyncIterable(stream, { preventCancel = false }) {
  if (stream[Symbol.asyncIterator]) {
    return stream[Symbol.asyncIterator]();
  }

  const reader = stream.getReader();

  return {
    async next() {
      try {
        const result = await reader.read();
        if (result.done) reader.releaseLock(); // release lock when stream becomes closed
        return result;
      } catch (e) {
        reader.releaseLock(); // release lock when stream becomes errored
        throw e;
      }
    },
    async return(arg) {
      if (!preventCancel) {
        const cancelPromise = reader.cancel(arg); // cancel first, but don't await yet
        reader.releaseLock(); // release lock first
        await cancelPromise; // now await it
      } else {
        reader.releaseLock();
      }
      return { done: true, value: undefined };
    },
    [Symbol.asyncIterator]() {
      return this;
    },
  };
}

[rattrayalex]

Thank you so much @MattiasBuelens ! After spending some time with the spec, I agree that looks closer to what it seems to intend (there are some nonsensical things it asks for instead, like returning the cancel promise from return(); I think what you have is better).

I wonder if this is worth making available as its own polyfill?

BTW, note that I had an error at the start, it should be:

  if (stream[Symbol.asyncIterator]) {
-   return stream[Symbol.asyncIterator];
+   return stream;
  }

[MattiasBuelens]

I have discontinued the next branch in favor of the v4 branch. This new branch does not contain the rewrites using the public API, and thus I no longer need to revert those rewrites either. 😁

See #1 (comment) for more information.