Function disposeOnDone

packages/computer/src/shell.ts:509–550

(stream: ReadableStream<T>, onDone: () => void)

// Wrap `stream` so its capnweb envelope is released exactly once on clean
// completion, source failure, or consumer cancellation.
function disposeOnDone<T>(stream: ReadableStream<T>, onDone: () => void): ReadableStream<T> {
  const reader = stream.getReader();
  let finished = false;
  const finish = () => {
    if (finished) return;
    finished = true;
    try {
      reader.releaseLock();
    } catch {}
    try {
      onDone();
    } catch {
      // Disposer failures cannot be recovered at this boundary.
    }
  };
  return new ReadableStream<T>(
    {
      async pull(controller) {
        try {
          const { value, done } = await reader.read();
          if (done) {
            finish();
            controller.close();
          } else {
            controller.enqueue(value);
          }
        } catch (error) {
          finish();
          controller.error(error);
        }
      },
      async cancel(reason) {
        try {
          await reader.cancel(reason);
        } finally {
          finish();
        }
      },
    },
    { highWaterMark: 0 },
  );
}

// Best-effort dispose of a capnweb result envelope. Real envelopes
// expose [Symbol.dispose]; test fakes return plain objects, so the