mirror of
https://github.com/vercel-labs/scriptc.git
synced 2026-10-23 06:10:02 +08:00
413 lines
12 KiB
JavaScript
413 lines
12 KiB
JavaScript
// The island's node:stream shim (with stream/promises and
|
|
// stream/consumers), differentially against Node: every line must match
|
|
// byte-for-byte. Each scenario records an ordered event log after
|
|
// quiescence — construction both styles (options and subclassing),
|
|
// flowing and paused modes, Node's observable tick ordering (data
|
|
// deferral at start, prefinish, finish/end order with and without a
|
|
// final/flush hook), pipe chains with drain-based backpressure,
|
|
// Transform/PassThrough, destroy with error propagation, setEncoding
|
|
// across chunk boundaries, objectMode, async iteration, pipeline error
|
|
// teardown, and finished()'s premature-close shape.
|
|
import {
|
|
Duplex, PassThrough, Readable, Transform, Writable, finished, pipeline,
|
|
} from "node:stream";
|
|
import { pipeline as pipelineP, finished as finishedP } from "node:stream/promises";
|
|
import { buffer as consumeBuffer, json as consumeJson, text as consumeText } from "node:stream/consumers";
|
|
import { EventEmitter } from "node:events";
|
|
|
|
const settle = () => new Promise((res) => setTimeout(res, 20));
|
|
const fmt = (c) => (typeof c === "string" ? "s:" + c : Buffer.isBuffer(c) ? "b:" + c.toString("utf8") : "o:" + JSON.stringify(c));
|
|
|
|
export async function report() {
|
|
const lines = [];
|
|
const scenario = async (label, fn) => {
|
|
let out;
|
|
try {
|
|
out = await fn();
|
|
} catch (e) {
|
|
out = "THREW:" + (e.code || e.constructor.name) + ":" + e.message;
|
|
}
|
|
await settle();
|
|
lines.push(label + " " + out);
|
|
};
|
|
|
|
await scenario("emitter:", async () => {
|
|
const log = [];
|
|
const em = new EventEmitter();
|
|
const a = () => log.push("a");
|
|
em.on("x", a);
|
|
em.prependListener("x", () => log.push("first"));
|
|
em.prependOnceListener("x", () => log.push("once-first"));
|
|
em.emit("x");
|
|
em.emit("x");
|
|
log.push("count:" + em.listenerCount("x"));
|
|
log.push("names:" + em.eventNames().join(","));
|
|
em.setMaxListeners(3);
|
|
log.push("max:" + em.getMaxListeners() + "," + new EventEmitter().getMaxListeners());
|
|
em.removeListener("x", a);
|
|
em.emit("x");
|
|
try {
|
|
em.emit("error", new Error("unhandled"));
|
|
} catch (e) {
|
|
log.push("threw:" + e.message);
|
|
}
|
|
const oncePromise = EventEmitter.once(em, "later");
|
|
em.emit("later", 1, "two");
|
|
const args = await oncePromise;
|
|
log.push("once:" + JSON.stringify(args));
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("flowing:", async () => {
|
|
const log = [];
|
|
const r = new Readable({ read() {} });
|
|
r.on("data", (c) => log.push("data:" + fmt(c)));
|
|
r.on("end", () => log.push("end"));
|
|
r.on("close", () => log.push("close"));
|
|
r.push("ab");
|
|
r.push("cd");
|
|
r.push(null);
|
|
await settle();
|
|
return log.join("|") + "|ended:" + r.readableEnded;
|
|
});
|
|
|
|
await scenario("paused-read:", async () => {
|
|
const r = new Readable({ read() {} });
|
|
r.push("hello ");
|
|
r.push("world");
|
|
r.push(null);
|
|
await settle();
|
|
return [fmt(r.read(3)), fmt(r.read()), String(r.read())].join("|");
|
|
});
|
|
|
|
await scenario("pull:", async () => {
|
|
const log = [];
|
|
let i = 0;
|
|
const r = new Readable({
|
|
read() {
|
|
i++;
|
|
if (i <= 3) this.push("c" + i);
|
|
else this.push(null);
|
|
},
|
|
});
|
|
r.on("data", (c) => log.push(fmt(c)));
|
|
await new Promise((res) => r.on("end", res));
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("pause-resume:", async () => {
|
|
const log = [];
|
|
const r = new Readable({ read() {} });
|
|
r.on("data", (c) => log.push(fmt(c)));
|
|
r.push("a");
|
|
await settle();
|
|
r.pause();
|
|
r.push("b");
|
|
await settle();
|
|
log.push("paused:" + r.isPaused());
|
|
r.resume();
|
|
await settle();
|
|
r.push(null);
|
|
await settle();
|
|
return log.join("|") + "|flow:" + r.readableFlowing;
|
|
});
|
|
|
|
await scenario("setEncoding:", async () => {
|
|
const log = [];
|
|
const r = new Readable({ read() {} });
|
|
r.setEncoding("utf8");
|
|
r.on("data", (c) => log.push(typeof c + ":" + c));
|
|
const snow = Buffer.from("a☃b");
|
|
r.push(snow.subarray(0, 2));
|
|
r.push(snow.subarray(2));
|
|
r.push(null);
|
|
await new Promise((res) => r.on("end", res));
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("objectMode:", async () => {
|
|
const log = [];
|
|
const r = new Readable({ objectMode: true, read() {} });
|
|
r.on("data", (c) => log.push(fmt(c)));
|
|
r.push({ n: 1 });
|
|
r.push("plain");
|
|
r.push(42);
|
|
r.push(null);
|
|
await new Promise((res) => r.on("end", res));
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("iterate:", async () => {
|
|
const got = [];
|
|
for await (const c of Readable.from(["x", "y", "z"])) got.push(fmt(c));
|
|
return got.join("|");
|
|
});
|
|
|
|
await scenario("destroy:", async () => {
|
|
const log = [];
|
|
const r = new Readable({ read() {} });
|
|
r.on("error", (e) => log.push("error:" + e.message));
|
|
r.on("close", () => log.push("close"));
|
|
r.destroy(new Error("boom"));
|
|
await settle();
|
|
return log.join("|") + "|destroyed:" + r.destroyed;
|
|
});
|
|
|
|
await scenario("writable:", async () => {
|
|
const log = [];
|
|
const w = new Writable({
|
|
write(chunk, enc, cb) {
|
|
log.push("w:" + fmt(chunk) + ":" + enc);
|
|
cb();
|
|
},
|
|
final(cb) {
|
|
log.push("final");
|
|
cb();
|
|
},
|
|
});
|
|
w.write("ab");
|
|
w.write(Buffer.from("cd"));
|
|
w.on("finish", () => log.push("finish"));
|
|
w.on("close", () => log.push("close"));
|
|
w.end("ef", () => log.push("endcb"));
|
|
await settle();
|
|
return log.join("|") + "|" + w.writableEnded + "," + w.writableFinished;
|
|
});
|
|
|
|
await scenario("write-after-end:", async () => {
|
|
const log = [];
|
|
const w = new Writable({ write(c, e, cb) { cb(); } });
|
|
w.on("error", (e) => log.push("error:" + e.code));
|
|
w.end();
|
|
w.write("late", (err) => log.push("cb:" + (err && err.code)));
|
|
await settle();
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("drain:", async () => {
|
|
const log = [];
|
|
const w = new Writable({
|
|
highWaterMark: 4,
|
|
write(c, e, cb) {
|
|
setTimeout(cb, 1);
|
|
},
|
|
});
|
|
const ok1 = w.write("aa");
|
|
const ok2 = w.write("bbbb");
|
|
w.on("drain", () => log.push("drain"));
|
|
await settle();
|
|
return [ok1, ok2, log.join("|")].join(",");
|
|
});
|
|
|
|
await scenario("write-error:", async () => {
|
|
const log = [];
|
|
const w = new Writable({
|
|
write(c, e, cb) {
|
|
cb(new Error("wfail"));
|
|
},
|
|
});
|
|
w.on("error", (e) => log.push("error:" + e.message));
|
|
w.on("close", () => log.push("close"));
|
|
w.write("x");
|
|
await settle();
|
|
return log.join("|") + "|" + w.destroyed;
|
|
});
|
|
|
|
await scenario("transform:", async () => {
|
|
const log = [];
|
|
const t = new Transform({
|
|
transform(chunk, enc, cb) {
|
|
cb(null, chunk.toString("utf8").toUpperCase());
|
|
},
|
|
flush(cb) {
|
|
cb(null, "!");
|
|
},
|
|
});
|
|
t.on("data", (c) => log.push("data:" + fmt(c)));
|
|
t.on("finish", () => log.push("finish"));
|
|
t.on("end", () => log.push("end"));
|
|
t.write("ab");
|
|
t.write("cd");
|
|
t.end();
|
|
await settle();
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("pipe-chain:", async () => {
|
|
const log = [];
|
|
const src = new Readable({ read() {} });
|
|
const p1 = new PassThrough();
|
|
const p2 = new PassThrough();
|
|
const sink = new Writable({
|
|
write(c, e, cb) {
|
|
log.push("sink:" + fmt(c));
|
|
cb();
|
|
},
|
|
});
|
|
sink.on("finish", () => log.push("finish"));
|
|
src.pipe(p1).pipe(p2).pipe(sink);
|
|
src.push("one");
|
|
src.push("two");
|
|
src.push(null);
|
|
await settle();
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("pipe-endfalse:", async () => {
|
|
const log = [];
|
|
const src = new Readable({ read() {} });
|
|
const sink = new Writable({ write(c, e, cb) { log.push(fmt(c)); cb(); } });
|
|
sink.on("finish", () => log.push("finish"));
|
|
src.pipe(sink, { end: false });
|
|
src.push("z");
|
|
src.push(null);
|
|
await settle();
|
|
return log.join("|") + "|" + sink.writableEnded;
|
|
});
|
|
|
|
await scenario("unpipe:", async () => {
|
|
const log = [];
|
|
const src = new Readable({ read() {} });
|
|
const sink = new Writable({ write(c, e, cb) { log.push(fmt(c)); cb(); } });
|
|
sink.on("unpipe", () => log.push("unpipe"));
|
|
src.pipe(sink);
|
|
src.push("a");
|
|
await settle();
|
|
src.unpipe(sink);
|
|
src.push("b");
|
|
src.resume();
|
|
await settle();
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("pipeline:", async () => {
|
|
const log = [];
|
|
const upper = new Transform({
|
|
objectMode: true,
|
|
transform(c, e, cb) { cb(null, String(c).toUpperCase()); },
|
|
});
|
|
const sink = new Writable({
|
|
objectMode: true,
|
|
write(c, e, cb) { log.push(fmt(c)); cb(); },
|
|
});
|
|
await new Promise((res) => pipeline(Readable.from(["p", "q"]), upper, sink, (err) => {
|
|
log.push("done:" + (err ? err.message : "null"));
|
|
res();
|
|
}));
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("pipeline-err:", async () => {
|
|
const log = [];
|
|
const src = new Readable({ read() {} });
|
|
const bad = new Transform({
|
|
transform(c, e, cb) { cb(new Error("mid")); },
|
|
});
|
|
const sink = new Writable({ write(c, e, cb) { cb(); } });
|
|
const p = new Promise((res) => pipeline(src, bad, sink, (err) => {
|
|
log.push("done:" + (err ? err.message : "null"));
|
|
res();
|
|
}));
|
|
src.push("x");
|
|
await p;
|
|
return log.join("|") + "|" + src.destroyed + "," + bad.destroyed + "," + sink.destroyed;
|
|
});
|
|
|
|
await scenario("promises:", async () => {
|
|
const sink = new Writable({ write(c, e, cb) { cb(); } });
|
|
await pipelineP(Readable.from(["1", "2"]), sink);
|
|
const r2 = Readable.from(["z"]);
|
|
r2.resume();
|
|
await finishedP(r2);
|
|
return "ok:" + sink.writableFinished + "," + r2.readableEnded;
|
|
});
|
|
|
|
await scenario("finished-premature:", async () => {
|
|
const r = new Readable({ read() {} });
|
|
const p = new Promise((res) => finished(r, (err) => res("fin:" + (err ? err.code : "null"))));
|
|
r.destroy();
|
|
return await p;
|
|
});
|
|
|
|
await scenario("consumers:", async () => {
|
|
const t = await consumeText(Readable.from([Buffer.from("hé"), Buffer.from("llo")]));
|
|
const j = await consumeJson(Readable.from(['{"a":', "1}"]));
|
|
const b = await consumeBuffer(Readable.from([Buffer.from([1]), Buffer.from([2])]));
|
|
return t + "|" + JSON.stringify(j) + "|" + b.toString("hex");
|
|
});
|
|
|
|
await scenario("duplex:", async () => {
|
|
const log = [];
|
|
const d = new Duplex({
|
|
read() {},
|
|
write(c, e, cb) {
|
|
log.push("w:" + fmt(c));
|
|
cb();
|
|
},
|
|
});
|
|
d.on("data", (c) => log.push("r:" + fmt(c)));
|
|
d.on("finish", () => log.push("finish"));
|
|
d.on("end", () => log.push("end"));
|
|
d.push("out");
|
|
d.write("in");
|
|
d.end();
|
|
d.push(null);
|
|
await settle();
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("subclass:", async () => {
|
|
const log = [];
|
|
class Counter extends Readable {
|
|
constructor() {
|
|
super({});
|
|
this.n = 0;
|
|
}
|
|
_read() {
|
|
this.n++;
|
|
if (this.n <= 2) this.push("n" + this.n);
|
|
else this.push(null);
|
|
}
|
|
}
|
|
class Sink extends Writable {
|
|
_write(c, e, cb) {
|
|
log.push("got:" + fmt(c));
|
|
cb();
|
|
}
|
|
_final(cb) {
|
|
log.push("final");
|
|
cb();
|
|
}
|
|
}
|
|
const r = await new Promise((res) => pipeline(new Counter(), new Sink(), (err) => res(err ? err.message : "done")));
|
|
return log.join("|") + "|" + r;
|
|
});
|
|
|
|
await scenario("readable-event:", async () => {
|
|
const log = [];
|
|
const r = new Readable({ read() {} });
|
|
r.on("readable", () => {
|
|
let c;
|
|
while ((c = r.read()) !== null) log.push(fmt(c));
|
|
});
|
|
r.on("end", () => log.push("end"));
|
|
r.push("m");
|
|
await settle();
|
|
r.push("n");
|
|
r.push(null);
|
|
await settle();
|
|
return log.join("|");
|
|
});
|
|
|
|
await scenario("unshift:", async () => {
|
|
const r = new Readable({ read() {} });
|
|
r.push("bc");
|
|
r.unshift("a");
|
|
r.push(null);
|
|
await settle();
|
|
return fmt(r.read());
|
|
});
|
|
|
|
return lines.join("\n");
|
|
}
|