scriptc/tests/fixtures/npm/node_modules/node24-web-stream-fixture/readable-lifecycle.js

import { ReadableStream } from "node:stream/web";
import { out, deferred, result, caught, sameError, launch } from "./helpers.js";

export function reportReadableLifecycle() { launch(async () => {
  {
    const start = deferred(), entered = deferred(); let pulls = 0, controller;
    const source = {
      start(c) { controller = c; out("start-this", this === source); return start.promise; },
      pull(c) { pulls++; out("pull-this", this === source, c === controller); c.enqueue("value"); c.close(); entered.resolve(); },
    };
    const stream = new ReadableStream(source, { highWaterMark: 0 }), r = stream.getReader(), read = r.read();
    source.pull = () => { throw new Error("changed pull"); }; source.cancel = source.pull;
    await new Promise(resolve => setTimeout(resolve, 0)); out("start-gate", pulls);
    start.resolve(); await entered.promise; out("start-done", (await read).value, (await r.read()).done, pulls);
  }
  {
    let calls = 0;
    const stream = new ReadableStream({ pull() { calls++; } }, { highWaterMark: 0 });
    const r = stream.getReader(), pending = r.read();
    await new Promise(resolve => setTimeout(resolve, 0)); out("idle-pull", calls);
    await r.cancel(); out("idle-cancel", (await pending).done);
  }
  {
    const source = { cancel(e) { out("cancel-captured", this === source, e); } }, stream = new ReadableStream(source);
    source.cancel = () => { throw new Error("changed cancel"); }; await stream.cancel("reason");
  }
  for (const state of ["closing", "closed", "errored"]) {
    let c; const stream = new ReadableStream({ start(controller) { c = controller; } });
    if (state === "closing") { c.enqueue("queued"); c.close(); }
    if (state === "closed") c.close();
    if (state === "errored") c.error(new Error("state"));
    out("controller-state", state, c.desiredSize, caught(() => c.enqueue("late")), caught(() => c.close()));
    c.error(new Error("again")); out("error-noop", state, c.desiredSize);
    await result(stream.cancel());
  }
  for (const callback of ["start", "pull"]) {
    const reason = new Error(callback), source = { [callback]() { return Promise.reject(reason); } };
    const stream = new ReadableStream(source, { highWaterMark: 0 }), r = stream.getReader(), read = sameError(r.read(), reason);
    out("callback-failure", callback, await read, await sameError(r.closed, reason));
  }
  {
    const reason = new Error("start-sync"); out("sync-start", (() => { try { new ReadableStream({ start() { throw reason; } }); return false; } catch (e) { return e === reason; } })());
  }
  for (const value of [null, 1, "bad"]) out("invalid-source", caught(() => new ReadableStream(value)));
  for (const callback of ["pull", "cancel"]) out("invalid-callback", callback, caught(() => new ReadableStream({ [callback]: 1 })));
  out("invalid-type", caught(() => new ReadableStream({ type: "default" })));
  {
    let c, sizeCalls = 0;
    const stream = new ReadableStream({ start(controller) { c = controller; } }, { highWaterMark: 2, size(v) { sizeCalls++; return v; } });
    const r = stream.getReader(), pending = r.read(); c.enqueue(100);
    out("direct-read-size", (await pending).value, sizeCalls, c.desiredSize);
    c.enqueue(1); c.enqueue(2); out("queue-size", sizeCalls, c.desiredSize); c.close();
    out("drain-close", (await r.read()).value, c.desiredSize, (await r.read()).value, c.desiredSize, (await r.read()).done);
  }
  for (const size of [-1, NaN, Infinity]) {
    let c; const stream = new ReadableStream({ start(controller) { c = controller; } }, { size() { return size; } });
    const r = stream.getReader(); out("invalid-size", String(size), caught(() => c.enqueue("a")), await result(r.closed), await result(r.read()), c.desiredSize);
  }
}); }