scriptc/tests/fixtures/npm/node_modules/node24-web-stream-fixture/pipe-errors.js

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

export function reportPipeErrors() { launch(async () => {
  for (const preventAbort of [false, true]) for (const cleanupFails of [false, true]) {
    const reason = new Error("source"), cleanup = new Error("abort"), log = [];
    const source = new ReadableStream({ start(c) { c.error(reason); } });
    const sink = new WritableStream({ abort(e) { log.push(e === reason); if (cleanupFails) throw cleanup; } });
    const p = source.pipeTo(sink, { preventAbort });
    out("source-error", preventAbort, cleanupFails, await sameError(p, cleanupFails && !preventAbort ? cleanup : reason), log.join(","), source.locked, sink.locked);
    if (preventAbort) await sink.close();
  }
  for (const preventCancel of [false, true]) for (const cleanupFails of [false, true]) {
    const reason = new Error("sink"), cleanup = new Error("cancel"), log = [];
    const source = new ReadableStream({ start(c) { c.enqueue("chunk"); }, cancel(e) { log.push(e === reason); if (cleanupFails) throw cleanup; } });
    const sink = new WritableStream({ write() { throw reason; } });
    out("sink-error", preventCancel, cleanupFails, await sameError(source.pipeTo(sink, { preventCancel }), cleanupFails && !preventCancel ? cleanup : reason), log.join(","), source.locked, sink.locked);
    if (preventCancel) await result(source.cancel());
  }
  {
    const reason = new Error("already"), source = new ReadableStream({ cancel(e) { out("initial-cancel", e === reason); } });
    const sink = new WritableStream({ start(c) { c.error(reason); } });
    out("initial-error", await sameError(source.pipeTo(sink), reason), source.locked, sink.locked);
  }
  {
    const entered = deferred(), gate = deferred(), log = []; let controller;
    const reason = new Error("during-write");
    const source = new ReadableStream({ start(c) { controller = c; c.enqueue("a"); } });
    const sink = new WritableStream({ async write(v) { log.push(v); entered.resolve(); await gate.promise; log.push("written"); }, abort(e) { log.push("abort:" + (e === reason)); } });
    const p = sameError(source.pipeTo(sink), reason); await entered.promise; controller.error(reason);
    await Promise.resolve(); out("error-waits", log.join(","), source.locked, sink.locked);
    gate.resolve(); out("error-done", await p, log.join(","), source.locked, sink.locked);
  }
  {
    let controller; const entered = deferred(), reason = new Error("idle-destination");
    const source = new ReadableStream({ pull() { entered.resolve(); }, cancel(e) { out("idle-cancel", e === reason); } }, { highWaterMark: 0 });
    const sink = new WritableStream({ start(c) { controller = c; } });
    const p = source.pipeTo(sink); await entered.promise; controller.error(reason);
    out("idle-error", await sameError(p, reason), source.locked, sink.locked);
  }
  {
    const reason = new Error("close"), source = ReadableStream.from(["last"]);
    const sink = new WritableStream({ write(v) { out("close-last", v); }, close() { throw reason; } });
    out("close-error", await sameError(source.pipeTo(sink), reason), source.locked, sink.locked);
  }
}); }