Function monitorStream

packages/deno/src/utils/streaming.ts:79–107

* zero-copy monitoring of stream progress.

(
  stream: ReadableStream<Uint8Array<ArrayBufferLike>>,
  onDone: () => void,
)

 * zero-copy monitoring of stream progress.
 */
function monitorStream(
  stream: ReadableStream<Uint8Array<ArrayBufferLike>>,
  onDone: () => void,
): ReadableStream<Uint8Array<ArrayBufferLike>> {
  const reader = stream.getReader();
  reader.closed.then(
    () => onDone(),
    () => onDone(),
  );
  return new ReadableStream({
    async start(controller) {
      let result: ReadableStreamReadResult<Uint8Array<ArrayBufferLike>>;
      do {
        result = await reader.read();
        if (result.value) {
          try {
            controller.enqueue(result.value);
          } catch (er) {
            controller.error(er);
            reader.releaseLock();
            return;
          }
        }
      } while (!result.done);
      controller.close();
      reader.releaseLock();
    },
  });
}