import { TransformStream, TransformStreamDefaultController, ByteLengthQueuingStrategy } from "node:stream/web"; import { out, deferred, result, caught, sameError, launch } from "./helpers.js"; export function reportTransforms() { launch(async () => { out("controller-export", TransformStreamDefaultController === globalThis.TransformStreamDefaultController, caught(() => new TransformStreamDefaultController())); out("controller-brand", ...["enqueue", "error", "terminate"].map(name => caught(() => TransformStreamDefaultController.prototype[name].call({}))), caught(() => Object.getOwnPropertyDescriptor(TransformStreamDefaultController.prototype, "desiredSize").get.call({}))); out("transform-brand", ...["readable", "writable"].map(name => caught(() => Object.getOwnPropertyDescriptor(TransformStream.prototype, name).get.call({})))); { let controller, transformed = 0; const t = new TransformStream({ start(c) { controller = c; }, transform(v, c) { transformed++; c.enqueue(v); } }); const w = t.writable.getWriter(), r = t.readable.getReader(); const first = w.write("a"), second = w.write("b"); await new Promise(resolve => setTimeout(resolve, 0)); out("no-demand", transformed, controller.desiredSize, w.desiredSize, controller instanceof TransformStreamDefaultController, Object.prototype.toString.call(controller)); out("first-demand", (await r.read()).value); await first; out("one-demand", transformed, controller.desiredSize); out("second-demand", (await r.read()).value); await second; await w.close(); out("demand-closed", (await r.read()).done, controller.desiredSize, await result(w.closed)); out("enqueue-closed", caught(() => controller.enqueue("late"))); } { const entered = deferred(); let controller, transformed = 0; const t = new TransformStream({ start(c) { controller = c; }, transform(v, c) { transformed++; c.enqueue(v); if (transformed === 2) entered.resolve(); } }, undefined, new ByteLengthQueuingStrategy({ highWaterMark: 4 })); const w = t.writable.getWriter(), r = t.readable.getReader(); const writes = [w.write(new Uint8Array(3)), w.write(new Uint8Array(2)), w.write(new Uint8Array(1))]; await entered.promise; await writes[1]; out("transform-capacity", transformed, controller.desiredSize); out("sized-read", (await r.read()).value.byteLength); await writes[2]; out("transform-resumed", transformed, controller.desiredSize); const close = w.close(), got = []; for (;;) { const v = await r.read(); if (v.done) break; got.push(v.value.byteLength); } await close; out("sized-rest", got.join(",")); } { const start = deferred(), flush = deferred(), entered = deferred(), log = []; const transformer = { start(c) { log.push("start:" + (this === transformer)); c.enqueue("start"); return start.promise; }, transform(v, c) { log.push("transform:" + (this === transformer)); c.enqueue(v); }, async flush(c) { log.push("flush:" + (this === transformer)); entered.resolve(); await flush.promise; c.enqueue("flush"); } }; const t = new TransformStream(transformer), w = t.writable.getWriter(), r = t.readable.getReader(); transformer.transform = () => { throw new Error("changed"); }; transformer.flush = transformer.transform; const write = w.write("value"); out("start-buffer", (await r.read()).value, log.join(",")); const read = r.read(); start.resolve(); out("start-release", (await read).value); await write; const close = w.close(); await entered.promise; out("flush-pending", log.join(",")); const final = r.read(); flush.resolve(); out("flush-value", (await final).value); await close; out("flush-done", (await r.read()).done); } for (const kind of ["error", "terminate", "cancel"]) { let c; const reason = new Error("controller-error"); const t = new TransformStream({ start(controller) { c = controller; } }); const w = t.writable.getWriter(), r = t.readable.getReader(); const write = result(w.write("blocked")); await new Promise(resolve => setTimeout(resolve, 0)); if (kind === "error") c.error(reason); if (kind === "terminate") c.terminate(); if (kind === "cancel") await r.cancel(reason); out("unblock", kind, await write, await result(w.closed), await result(r.read()), c.desiredSize); } for (const kind of ["transform", "flush", "start"]) { const reason = new Error(kind), t = new TransformStream({ [kind]() { return Promise.reject(reason); } }); const w = t.writable.getWriter(), r = t.readable.getReader(), read = sameError(r.read(), reason); if (kind === "transform") out("transform-failure", await sameError(w.write("x"), reason)); if (kind === "flush") out("flush-failure", await sameError(w.close(), reason)); out("both-error", kind, await read, await sameError(w.closed, reason), await sameError(r.closed, reason)); } for (const direction of ["readable", "writable"]) for (const fails of [false, true]) { const entered = deferred(), gate = deferred(), reason = new Error("cancel"), failure = new Error("cancel-failure"), log = []; const transformer = { cancel(e) { log.push(this === transformer && e === reason); entered.resolve(); return gate.promise; } }; const t = new TransformStream(transformer), w = t.writable.getWriter(), r = t.readable.getReader(); transformer.cancel = () => { throw new Error("changed"); }; const end = direction === "readable" ? r.cancel(reason) : w.abort(reason); await entered.promise; out("cancel-wait", direction, fails, log.join(",")); if (fails) gate.reject(failure); else gate.resolve(); out("cancel-end", direction, fails, await result(end), await sameError(direction === "readable" ? w.closed : r.closed, fails ? failure : reason)); } { const entered = deferred(), gate = deferred(), reason = new Error("race"), log = []; const t = new TransformStream({ async flush() { log.push("flush"); entered.resolve(); await gate.promise; }, cancel() { log.push("cancel"); } }); const w = t.writable.getWriter(), r = t.readable.getReader(), close = result(w.close()); await entered.promise; const cancel = result(r.cancel(reason)); gate.resolve(); out("flush-cancel-race", await close, await cancel, log.join(","), await result(w.closed), (await r.read()).done); } }); }