Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157// process.stdin: a Readable over fd 0, fed by the kernel's `stdin` messages. Reading// follows Node's stdin handle: it starts when the stream first asks for data, stops when// the stream is paused (Node's onpause, a tick after 'pause') or full, and only an// active, ref'd read holds the loop, so a stdin nobody reads never keeps the process// alive. Input is taken in the poll phase, everything that has arrived in one callback,// as libuv reads a ready pipe until it is drained: data and EOF that arrive together// are both pushed before a 'data' listener's pause() takes effect.import { Buffer } from "./buffer";import type { Loop } from "./loop";import { Readable } from "./stream";import { nextTick } from "./ticks";
const READ_SIZE = 64 * 1024;
export class StdinStream extends Readable { readonly fd = 0; #loop: Loop; /** Bytes the kernel delivered that the stream has not taken; null is EOF. */ #inbox: Array<Uint8Array | null> = []; #drainQueued = false; /** The stream has a _read outstanding. */ #wanted = false; /** The handle is reading: what libuv's readStart/readStop toggle. */ #reading = false; #ref = true; #holding = false;
constructor(loop: Loop, highWaterMark = 64 * 1024) { super({ highWaterMark }); this.#loop = loop; this.on("pause", () => nextTick(() => { if (this.readableFlowing !== true) this.#setReading(false); })); this.on("resume", () => { if (this.#wanted) this.#setReading(true); }); }
/** The kernel delivered input (null = EOF), taken in libuv's 64 KiB reads. */ receive(bytes: Uint8Array | null): void { if (bytes === null) this.#inbox.push(null); else for (let at = 0; at < bytes.length; at += READ_SIZE) this.#inbox.push(bytes.subarray(at, at + READ_SIZE)); this.#serveRaw(); this.#queueDrain(); }
/** fs.read() on fd 0: bytes straight off the pipe, as read(2) gives them, past the * stream (Go's wasm runtime reads its stdin this way). `done` runs in the poll phase * with at most `max` bytes, or null at EOF; a pending read holds the loop. */ readRaw(max: number, done: (bytes: Uint8Array | null) => void): void { this.#rawReads.push({ max, done }); this.#serveRaw(); }
/** node:wasi's fd_read on fd 0 (preview1 has no async read): takes whatever is * already in the inbox without the poll-phase hop readRaw uses, since a wasm guest * needs a number back now, not next tick. undefined means nothing has arrived yet -- * genuinely blocking until more does would need a kernel-level sync stdin syscall * (stdin is delivered by postMessage, not the Atomics mailbox fs reads use), so a * caller past this point sees an empty read rather than deadlocking the worker. */ readSyncRaw(max: number): Uint8Array | null | undefined { if (this.#inbox.length === 0) return undefined; const head = this.#inbox[0]!; if (head === null) { this.#inbox.shift(); return null; } const bytes = head.length <= max ? head : head.subarray(0, max); if (head.length <= max) this.#inbox.shift(); else this.#inbox[0] = head.subarray(max); return bytes; }
#rawReads: Array<{ max: number; done: (bytes: Uint8Array | null) => void }> = []; #rawHolding = false;
#serveRaw(): void { while (this.#rawReads.length && this.#inbox.length) { const { max, done } = this.#rawReads.shift()!; const head = this.#inbox[0]!; let bytes: Uint8Array | null = null; if (head !== null) { bytes = head.length <= max ? head : head.subarray(0, max); if (head.length <= max) this.#inbox.shift(); else this.#inbox[0] = head.subarray(max); } this.#loop.enqueueCompletion(() => done(bytes)); } const hold = this.#rawReads.length > 0; if (hold === this.#rawHolding) return; this.#rawHolding = hold; if (hold) this.#loop.hold(); else this.#loop.release(); }
override _read(): void { this.#wanted = true; this.#setReading(true); }
ref(): this { this.#ref = true; this.#updateHold(); return this; }
unref(): this { this.#ref = false; this.#updateHold(); return this; }
#setReading(reading: boolean): void { this.#reading = reading; this.#queueDrain(); this.#updateHold(); }
#queueDrain(): void { if (this.#drainQueued || !this.#reading || !this.#wanted || this.#inbox.length === 0) return; this.#drainQueued = true; this.#loop.enqueueCompletion(() => { this.#drainQueued = false; this.#drain(); }); }
#drain(): void { while (this.#reading && this.#wanted && this.#inbox.length) { const chunk = this.#inbox.shift()!; if (chunk === null) { this.#wanted = false; this.#reading = false; this.push(null); break; } // push() returning false means the buffer is full: stop until the next _read. if (!this.push(Buffer.from(chunk))) { this.#wanted = false; this.#reading = false; } } this.#updateHold(); }
#updateHold(): void { const hold = this.#reading && this.#ref && !this.destroyed; if (hold === this.#holding) return; this.#holding = hold; if (hold) this.#loop.hold(); else this.#loop.release(); }
override _destroy(error: Error | null, callback: (error?: Error | null) => void): void { this.#reading = false; this.#updateHold(); callback(error); }}