scriptc/tests/fixtures/npm/node_modules/webstream/index.js

// Exercises the island's WHATWG-streams subset differentially: under Node
// these are the REAL streams, so every line pins the island implementation
// against the spec behavior the AI-SDK/eventsource-parser paths rely on —
// reader protocol, pull timing, async iteration, pipeThrough chains,
// TransformStream subclassing (the eventsource-parser shape), flush
// ordering, error propagation both directions, and cancel-on-break. The
// async work is left PENDING when run() returns: the island's job queue
// drains at loop quiescence, like Node's microtasks.
const out = (line) => process.stdout.write(line + "\n");

// This probe intentionally uses the private loader hook from inside an
// embedded npm module. Standalone islands do not install the hook because
// they have no module graph to load.
export function modulePresence() {
  const m = globalThis.__scr_require("node:stream/web");
  return [
    typeof m.ReadableStream,
    typeof m.TransformStream,
    typeof m.TextDecoderStream,
    typeof URL,
    typeof Buffer,
    typeof setImmediate,
    typeof clearImmediate,
    "WritableStream" in m,
    "TextEncoderStream" in m,
    "CountQueuingStrategy" in m,
    "ByteLengthQueuingStrategy" in m,
    "ReadableStreamDefaultReader" in m,
    "ReadableStreamDefaultController" in m,
    "WritableStreamDefaultWriter" in m,
  ].join(" ");
}

import {
  ReadableStream as ModuleReadableStream,
  TransformStream as ModuleTransformStream,
  TextDecoderStream as ModuleTextDecoderStream,
} from "node:stream/web";

export function run() {
  (async () => {
    // node:stream/web's implemented named exports are the same constructors
    // as the corresponding web globals in both runtimes.
    out("module-stream " + [
      typeof ModuleReadableStream,
      typeof ModuleTransformStream,
      typeof ModuleTextDecoderStream,
      ModuleReadableStream === ReadableStream,
      ModuleTransformStream === TransformStream,
      ModuleTextDecoderStream === TextDecoderStream,
    ].join(" "));

    // push source + async iteration
    {
      const rs = new ReadableStream({
        start(c) {
          c.enqueue("a");
          c.enqueue("b");
          c.close();
        },
      });
      const got = [];
      for await (const v of rs) got.push(v);
      out("push-iter " + got.join(","));
    }
    // pull source: pull count observes the spec's proactive HWM-1 pulling
    {
      let n = 0;
      const rs = new ReadableStream({
        pull(c) {
          n++;
          if (n <= 3) c.enqueue("p" + n);
          else c.close();
        },
      });
      const got = [];
      for await (const v of rs) got.push(v);
      out("pull-iter " + got.join(",") + " pulls=" + n);
    }
    // reader protocol + locking
    {
      const rs = new ReadableStream({
        start(c) {
          c.enqueue(1);
          c.close();
        },
      });
      const r = rs.getReader();
      const a = await r.read();
      const b = await r.read();
      out(`reader ${a.value} ${a.done} ${b.value} ${b.done} locked=${rs.locked}`);
      r.releaseLock();
      out("released " + rs.locked);
    }
    // enqueue AFTER a read is already waiting
    {
      let ctrl;
      const rs = new ReadableStream({ start(c) { ctrl = c; } });
      const r = rs.getReader();
      const p = r.read();
      ctrl.enqueue("late");
      const a = await p;
      ctrl.close();
      const b = await r.read();
      await r.closed;
      out(`deferred ${a.value} ${a.done} ${b.done} closed-resolved`);
    }
    // controller.error rejects waiting and future reads
    {
      let ctrl;
      const rs = new ReadableStream({ start(c) { ctrl = c; } });
      const r = rs.getReader();
      const p = r.read();
      ctrl.error(new Error("boom"));
      const m1 = await p.then(() => "no-throw", (e) => e.message);
      const m2 = await r.read().then(() => "no-throw", (e) => e.message);
      out(`errored ${m1} ${m2}`);
    }
    // pipeThrough + TransformStream SUBCLASS (the eventsource-parser shape)
    {
      class Upper extends TransformStream {
        constructor() {
          super({
            transform(chunk, c) {
              c.enqueue(String(chunk).toUpperCase());
            },
          });
        }
      }
      const src = new ReadableStream({
        start(c) {
          c.enqueue("x");
          c.enqueue("y");
          c.close();
        },
      });
      const got = [];
      for await (const v of src.pipeThrough(new Upper())) got.push(v);
      out("pipe-subclass " + got.join(","));
    }
    // start + transform + flush ordering
    {
      const t = new TransformStream({
        start(c) { c.enqueue("S"); },
        transform(chunk, c) { c.enqueue(chunk + "!"); },
        flush(c) { c.enqueue("F"); },
      });
      const src = new ReadableStream({
        start(c) {
          c.enqueue("m");
          c.close();
        },
      });
      const got = [];
      for await (const v of src.pipeThrough(t)) got.push(v);
      out("flush " + got.join(","));
    }
    // chained pipeThrough
    {
      const dup = () =>
        new TransformStream({
          transform(chunk, c) {
            c.enqueue(chunk);
            c.enqueue(chunk);
          },
        });
      const src = new ReadableStream({
        start(c) {
          c.enqueue("z");
          c.close();
        },
      });
      const got = [];
      for await (const v of src.pipeThrough(dup()).pipeThrough(dup())) got.push(v);
      out("chain " + got.join(","));
    }
    // a throwing transform errors the consumer
    {
      const t = new TransformStream({
        transform() {
          throw new Error("t-boom");
        },
      });
      const src = new ReadableStream({
        start(c) {
          c.enqueue("q");
          c.close();
        },
      });
      let msg = "no-throw";
      try {
        for await (const v of src.pipeThrough(t)) out("unreachable " + v);
      } catch (e) {
        msg = e.message;
      }
      out("transform-error " + msg);
    }
    // a source error propagates THROUGH a transform to the consumer
    {
      let ctrl;
      const t = new TransformStream({
        transform(chunk, c) {
          c.enqueue(chunk);
        },
      });
      const src = new ReadableStream({ start(c) { ctrl = c; } });
      const it = src.pipeThrough(t)[Symbol.asyncIterator]();
      ctrl.enqueue("one");
      const a = await it.next();
      ctrl.error(new Error("src-boom"));
      const m = await it.next().then(() => "no-throw", (e) => e.message);
      out(`source-error ${a.value} ${m}`);
    }
    // breaking out of for-await cancels the source and releases the lock
    {
      let cancelled = "no";
      const rs = new ReadableStream({
        start(c) {
          c.enqueue(1);
          c.enqueue(2);
          c.enqueue(3);
        },
        cancel() {
          cancelled = "yes";
        },
      });
      for await (const v of rs) {
        if (v === 1) break;
      }
      out(`break-cancel ${cancelled} locked=${rs.locked}`);
    }
    // writing a TransformStream's writable directly (writer protocol)
    {
      const t = new TransformStream();
      const w = t.writable.getWriter();
      const r = t.readable.getReader();
      const wp = w.write("idw");
      const a = await r.read();
      await wp;
      const cp = w.close();
      const b = await r.read();
      await cp;
      out(`identity-writer ${a.value} ${b.done}`);
    }
    // desiredSize under the default HWM-1 strategy
    {
      let ctrl;
      const rs = new ReadableStream({ start(c) { ctrl = c; } });
      const d0 = ctrl.desiredSize;
      ctrl.enqueue("a");
      const d1 = ctrl.desiredSize;
      const rd = rs.getReader();
      await rd.read();
      const d2 = ctrl.desiredSize;
      ctrl.close();
      out(`desired ${d0} ${d1} ${d2}`);
    }
    // stream.cancel(): source sees the reason, readers see done
    {
      let reason = "";
      const rs = new ReadableStream({
        start(c) {
          c.enqueue("x");
        },
        cancel(r) {
          reason = String(r);
        },
      });
      await rs.cancel("bye");
      const a = await rs.getReader().read();
      out(`cancel ${reason} ${a.done}`);
    }
    // misuse throws TypeErrors: enqueue-after-close, double getReader
    {
      let enq = "no-throw";
      const rs = new ReadableStream({
        start(c) {
          c.close();
          try {
            c.enqueue("x");
          } catch (e) {
            enq = String(e instanceof TypeError);
          }
        },
      });
      rs.getReader();
      let lock = "no-throw";
      try {
        rs.getReader();
      } catch (e) {
        lock = String(e instanceof TypeError);
      }
      out(`misuse enqueue-after-close=${enq} double-reader=${lock}`);
    }
    out("done");
  })();
}