trending/scriptc
1
0
Fork 0
mirror of https://github.com/vercel-labs/scriptc.git synced 2026-11-01 07:33:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Keep typed records unboxed in object-mode streams

Browse Source
This commit is contained in:
Chris Tate
2026-10-29 23:46:36 -05:00
parent 8d92f6a886
commit 2353a04540
34 changed files with 1010 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/compiler/src/backend/llvm/expr-stream-bridges.ts
+48
View File
@@ -40,6 +40,7 @@ import {
llFieldType,
loadStructField,
releaseSym,
traceAdapter,
traceArg,
vAdapters,
type ShapeHost,
@@ -754,6 +755,50 @@ function symbolKeyLines(
};
}
/** The ScrStreamTypedDesc constant (scr_stream.h) of a record type an
* object-mode stream holds as the record itself, or null for a type whose
* chunks stay boxed: records without declared fields, with an index
* signature or accessor slots, tuples, the text codec records (they box as
* handles), and records the collector traces (a stream does not trace the
* records it holds). The constant's `box` converts the record as a
* live-reference conversion to dyn does (streamTypedRefBoxValue). */
export function streamTypedDesc(host: LlvmEmitterContext, t: IrType): string | null {
if (t.kind !== "record") return null;
const shape = host.recordsById.get(t.shapeId);
if (
!shape ||
shape.fields.length === 0 ||
shape.indexValue !== undefined ||
shape.tuple ||
shapeHasAccessorSlots(shape) ||
shape.fields.some((field) => field.name === "handleEvent" && field.type.kind === "func") ||
textCodecRecordClass(t, (id) => host.recordsById.get(id)) !== null ||
traceAdapter(host.shapeHost, t) !== null
)
return null;
const key = typeKey(t);
const cacheKey = `std:${key}`;
const existing = host.resolveThunks.get(cacheKey);
if (existing) return `@${existing}`;
const sym = host.contentSymbol("sc_std_", key);
host.resolveThunks.set(cacheKey, sym);
host.journal?.allocated(sym, "streamTypedDesc", [t]);
const B = new BlockBuilder();
const boxed = streamTypedRefBoxValue(host, B, t, "%p");
B.terminate(`ret ptr ${boxed}`);
const rc = vAdapters(host.shapeHost, t);
host.resolveThunkDefs.push(
`define internal ptr @${sym}_box(ptr %p) ${FN_ATTRS} { ; box a typed stream chunk ${key}`,
B.render(),
`}`,
``,
`@${sym} = internal constant { ptr, ${host.sizeType}, ptr, ptr, ptr } { ptr ${host.cstr(key)}, ${host.sizeType} ${Buffer.byteLength(key, "utf8")}, ptr ${rc.retain}, ptr ${rc.release}, ptr @${sym}_box }`,
``,
);
host.journal?.built();
return `@${sym}`;
}
export function streamTypedRefBoxValue(
host: LlvmEmitterContext,
B: BlockBuilder,
@@ -2018,6 +2063,9 @@ export function streamFromArrayAdapter(
}
registerReplay("dynPromiseAdapter", (host, args) => dynPromiseAdapter(host, args[0] as IrType));
registerReplay("streamTypedDesc", (host, args) =>
(streamTypedDesc(host, args[0] as IrType) ?? "").slice(1),
);
registerReplay("liveDynUnionRefAdapter", (host, args) =>
liveDynUnionRefAdapter(host, args[0] as IrType & { kind: "union" }),
);
packages/compiler/src/backend/llvm/expr-stream-callbacks.ts
+87 -15
View File
@@ -1,10 +1,12 @@
/* Focused LLVM expression emission extracted from emitter.ts. */
import { InternalCompilerError } from "../../errors.js";
import { type IrType, isRefCounted, STRING, typeKey } from "../../ir/ir.js";
import { streamTypedRefEligible } from "../../ir/analysis.js";
import { FN_ATTRS, releaseSym, retainSym, traceArg, vAdapters } from "./shapes.js";
import type { LlvmEmitterContext } from "./expr-context.js";
import { unionPeekLines, unionTagLines } from "./union-repr.js";
import { registerReplay } from "./emit-journal.js";
import { streamTypedDesc } from "./expr-stream-bridges.js";
export function streamDataAdapter(
host: LlvmEmitterContext,
@@ -259,6 +261,31 @@ export function streamDoneFnFor(
strTag = def!.arms.findIndex((a) => a.kind === "string");
}
}
// data of one record type or nothing (a typed completion,
// pinTypedDataCompletion): a record with a typed-chunk descriptor
// completes as a typed chunk, and another record boxes by reference
// (the conversion a dyn `data` parameter receives).
let typedTag = -1;
let typedDesc: string | null = null;
let liveData = false;
if (valueData && dataT?.kind === "union") {
const arms = host.unionsById.get(dataT.unionId)?.arms ?? [];
const record = arms.findIndex((a) => a.kind === "record");
if (arms.length === 2 && record >= 0 && arms.some((a) => a.kind === "undefinedT")) {
typedDesc = streamTypedDesc(host, arms[record]!);
if (typedDesc !== null) typedTag = record;
else {
// the records a conversion to dyn keeps by reference (Lowerer's
// liveReference)
const armT = arms[record]!;
const shape = armT.kind === "record" ? host.recordsById.get(armT.shapeId) : undefined;
liveData =
streamTypedRefEligible(armT) &&
shape !== undefined &&
(shape.fields.length > 0 || shape.indexValue !== undefined);
}
}
}
const entry =
kind === "c"
? "scr_stream_construct_done"
@@ -269,15 +296,19 @@ export function streamDoneFnFor(
: kind === "d"
? "scr_stream_destroy_done"
: kind === "t"
? valueData
? "scr_stream_transform_done_value"
: "scr_stream_transform_done"
: valueData
? "scr_stream_flush_done_value"
: "scr_stream_flush_done";
? typedDesc !== null
? "scr_stream_transform_done_typed"
: valueData
? "scr_stream_transform_done_value"
: "scr_stream_transform_done"
: typedDesc !== null
? "scr_stream_flush_done_typed"
: valueData
? "scr_stream_flush_done_value"
: "scr_stream_flush_done";
const twoSlot = (kind === "t" || kind === "l") && !valueData;
host.declare(
`declare void @${entry}(ptr, ptr${twoSlot ? ", ptr, ptr" : valueData ? ", ptr" : ""})`,
`declare void @${entry}(ptr, ptr${typedDesc !== null ? ", ptr, ptr" : twoSlot ? ", ptr, ptr" : valueData ? ", ptr" : ""})`,
);
host.declare(`declare ptr @scr_error_retain_v(ptr)`);
host.declare(`declare void @scr_union_release(ptr)`);
@@ -291,6 +322,7 @@ export function streamDoneFnFor(
`entry:`,
` %eslot = alloca ptr`,
...(twoSlot ? [` %bslot = alloca ptr`, ` %sslot = alloca ptr`] : []),
...(typedDesc !== null ? [` %tslot = alloca ptr`] : []),
` %capsp = getelementptr inbounds %ScrClosure, ptr %self, i64 1`,
` %bx = load ptr, ptr %capsp`,
// the stream, borrowed: the box holds it, and the closure (alive for
@@ -299,10 +331,12 @@ export function streamDoneFnFor(
` %s = load ptr, ptr %sp`,
` store ptr null, ptr %eslot`,
...(twoSlot ? [` store ptr null, ptr %bslot`, ` store ptr null, ptr %sslot`] : []),
...(typedDesc !== null ? [` store ptr null, ptr %tslot`] : []),
];
// A tag-guarded retained peek out of a union argument into a slot.
// A tag-guarded peek out of a union argument into a slot, retained
// unless retain is null.
let arm = 0;
const unwrap = (u: string, tag: number, retain: string, slot: string): void => {
const unwrap = (u: string, tag: number, retain: string | null, slot: string): void => {
const a = arm++;
const union = u === "%e" ? errT : dataT;
const nullable = union?.kind === "union" ? host.nullableUnions.get(union.unionId) : null;
@@ -315,8 +349,9 @@ export function streamDoneFnFor(
` br i1 %hit${a}, label %yes${a}, label %done${a}`,
`yes${a}:`,
...unionPeekLines(nullable, u, `%pv${a}`),
` %rt${a} = call ptr ${retain}(ptr %pv${a})`,
` store ptr %rt${a}, ptr ${slot}`,
...(retain !== null
? [` %rt${a} = call ptr ${retain}(ptr %pv${a})`, ` store ptr %rt${a}, ptr ${slot}`]
: [` store ptr %pv${a}, ptr ${slot}`]),
` br label %done${a}`,
`done${a}:`,
);
@@ -330,8 +365,21 @@ export function streamDoneFnFor(
host.declare(`declare ptr @scr_str_retain_v(ptr)`);
unwrap("%d", strTag, "@scr_str_retain_v", "%sslot");
}
if (typedDesc !== null) unwrap("%d", typedTag, null, "%tslot");
d.push(` %ev = load ptr, ptr %eslot`);
if (valueData) {
if (typedDesc !== null) {
d.push(
` %tv = load ptr, ptr %tslot`,
` call void @${entry}(ptr %s, ptr %ev, ptr %tv, ptr ${typedDesc}) ; moves err; borrows s and the record`,
);
} else if (liveData) {
host.declare(`declare void @scr_dyn_release(ptr)`);
d.push(
` %dv = call ptr @${host.liveDynUnionRefAdapter(dataT as IrType & { kind: "union" })}(ptr %d)`,
` call void @${entry}(ptr %s, ptr %ev, ptr %dv) ; moves err; borrows s and the value`,
` call void @scr_dyn_release(ptr %dv)`,
);
} else if (valueData) {
const conv = host.dyn.toDynHelper(dataT!);
host.declare(`declare void @scr_dyn_release(ptr)`);
d.push(
@@ -840,18 +888,42 @@ export function streamCbThunkFor(
host.declare(`declare void @scr_dyn_release(ptr)`);
host.declare(`declare zeroext i1 @scr_exc_pending()`);
if (hasThis) host.declare(`declare void @scr_stream_release_v(ptr)`);
// A typed chunk of the declared record type (scr_stream_push_typed)
// is the record itself; any other chunk converts.
const desc = streamTypedDesc(host, p);
if (desc !== null) {
host.declare(`declare ptr @scr_stream_chunk_typed(ptr, ptr, ptr)`);
d.push(
` br label %ctchk${i}`,
`ctchk${i}:`,
` %ctt${i} = call ptr @scr_stream_chunk_typed(ptr %s, ptr %chunk, ptr ${desc})`,
` %cthit${i} = icmp ne ptr %ctt${i}, null`,
` br i1 %cthit${i}, label %ctok${i}, label %ctconv${i}`,
`ctconv${i}:`,
);
}
d.push(
` %cv${i} = call ptr @scr_stream_chunk_value(ptr %s, ptr %chunk)`,
` %ct${i} = call ${ty === "i1" ? "zeroext i1" : ty} @${helper}(ptr %cv${i}, ptr null)`,
` call void @scr_dyn_release(ptr %cv${i})`,
` %ctbad${i} = call zeroext i1 @scr_exc_pending()`,
` br i1 %ctbad${i}, label %ctfail${i}, label %ctok${i}`,
` br i1 %ctbad${i}, label %ctfail${i}, label %${desc !== null ? "ctconvok" : "ctok"}${i}`,
`ctfail${i}:`,
...(hasThis ? [` call void @scr_stream_release_v(ptr %sr)`] : []),
` ret void`,
`ctok${i}:`,
);
passed.push(`${ty} %ct${i}`);
if (desc !== null) {
d.push(
`ctconvok${i}:`,
` br label %ctok${i}`,
`ctok${i}:`,
` %ctv${i} = phi ptr [ %ctt${i}, %ctchk${i} ], [ %ct${i}, %ctconvok${i} ]`,
);
passed.push(`ptr %ctv${i}`);
} else {
d.push(`ctok${i}:`);
passed.push(`${ty} %ct${i}`);
}
} else if (isEncPos && p.kind === "string") {
// Node's encoding for decoded (Buffer) chunks is 'buffer'; an
// object-mode side passes the write's encoding argument ("" for
packages/compiler/src/backend/llvm/lib-stream.ts
+34
View File
@@ -3,6 +3,8 @@ import { InternalCompilerError } from "../../errors.js";
import type { LlvmEmitterContext, LibCallExpr, LlValue } from "./expr-context.js";
import { f64Lit } from "./common.js";
import { pipelinePumpedRuns } from "../../ir/builtin-signatures.js";
import { emitBorrowedInput } from "./borrowed-inputs.js";
import { streamTypedDesc } from "./expr-stream-bridges.js";
const STREAM_CANONICAL_CBS: Record<string, ("r" | "w" | "f" | "d" | "t" | "l")[]> = {
readable: ["r", "d"],
@@ -14,6 +16,10 @@ const STREAM_CANONICAL_CBS: Record<string, ("r" | "w" | "f" | "d" | "t" | "l")[]
export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue {
const B = host.B;
if (e.fn === "readable.pushDyn" || e.fn === "writable.writeDyn") {
const typed = emitTypedChunkCall(host, e);
if (typed !== null) return typed;
}
if (e.fn === "stream.addAbortSignal" || e.fn === "stream.addAbortSignalFn") {
// stream.addAbortSignal: [stream, signal] — the constructors' signal
// option chained over the fresh stream (a falsy signal is ignored);
@@ -872,6 +878,34 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
return host.emitGenericLibCall(e);
}
/** push(value) or write(value) of a record that converts to dyn by
* reference: the stream receives the record as a typed chunk
* (scr_stream_push_typed / scr_stream_write_typed), boxed only for a
* consumer that reads it as a dyn value. Null for another chunk. */
function emitTypedChunkCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue | null {
const chunk = e.args[1];
if (chunk === undefined || chunk.kind !== "dynFrom" || !chunk.liveRef) return null;
const desc = streamTypedDesc(host, chunk.value.type);
if (desc === null) return null;
const B = host.B;
const recv = host.emitExpr(e.args[0]!);
const rec = emitBorrowedInput(host, chunk.value);
const t = B.tmp();
if (e.fn === "readable.pushDyn") {
host.declare(`declare zeroext i1 @scr_stream_push_typed(ptr, ptr, ptr)`);
B.line(
`${t} = call zeroext i1 @scr_stream_push_typed(ptr ${recv.name}, ptr ${rec.name}, ptr ${desc})`,
);
} else {
host.declare(`declare zeroext i1 @scr_stream_write_typed(ptr, ptr, ptr, ptr)`);
B.line(
`${t} = call zeroext i1 @scr_stream_write_typed(ptr ${recv.name}, ptr ${rec.name}, ptr ${desc}, ptr null)`,
);
}
host.emitPendingCheck();
return e.type.kind === "void" ? { name: "", type: e.type } : { name: t, type: e.type };
}
/** The encoding argument a write passes when the call has none: undefined
* (a null pointer), or the null value when a callback took its place. */
function writeEncNull(host: LlvmEmitterContext, which: "null" | "undef"): string {
packages/compiler/src/backend/llvm/record-views.ts
+11
View File
@@ -202,6 +202,17 @@ export function planRecordViews(mod: IrModule, options: ViewOptions): RecordView
source: e.value.value.type.shapeId,
});
} else if (e.kind === "dynFrom" && e.liveRef === true) boxed.push(e.value.type);
else if (e.kind === "libCall" && /^(transform|passthrough)\.(new|init)$/.test(e.fn)) {
// A transform or flush callback's typed completion data
// (pinTypedDataCompletion) reaches untyped consumers as a live
// capsule: scr_stream_push_typed boxes it for them.
for (const arg of e.args) {
if (arg.type.kind !== "func") continue;
for (const param of arg.type.params)
if (param.kind === "func" && param.params[1] !== undefined)
boxed.push(param.params[1]);
}
}
return true;
},
});
packages/compiler/src/frontend/lowering/lower-stream.ts
+87 -1
View File
@@ -808,6 +808,90 @@ function pinAnyDataCompletion(
return symbol;
}
/** An object-mode Transform's completion callback whose every use calls
* it with one record type as `data`, or with no data or undefined: the
* callback parameter is typed with `data` as that record or undefined
* while the option callback lowers, so the record reaches the stream as a
* typed chunk (the emitted completion thunk) instead of converting to a
* checked-dynamic value at the call. Returns the pinned parameter symbol
* (the caller unpins it), or null when another value, a callback that
* escapes, or no record is passed. */
function pinTypedDataCompletion(
lowerer: Lowerer,
which: string,
fnNode: ts.MethodDeclaration | ts.FunctionExpression | ts.ArrowFunction,
): ts.Symbol | null {
const index = which === "transform" ? 2 : which === "flush" ? 0 : -1;
const param = index < 0 ? undefined : fnNode.parameters.nodes[index];
if (param === undefined || !ts.isIdentifier(param.name) || param.dotDotDotToken) return null;
const symbol = lowerer.checker.getSymbolAtLocation(param.name);
if (!symbol || lowerer.stageSourceTypes.has(symbol)) return null;
const declared = lowerer.paramShape(param).type;
if (declared.kind !== "func" || declared.rest || declared.params[1]?.kind !== "dyn") return null;
if (fnNode.body === undefined) return null;
const record = completionDataRecord(lowerer, fnNode.body, symbol, param.name);
if (record === null) return null;
const data = unionOf(lowerer, [record, { kind: "undefinedT" }]);
lowerer.stageSourceTypes.set(
symbol,
funcOf([declared.params[0]!, data, ...declared.params.slice(2)], declared.ret),
);
return symbol;
}
/** The one record type every call of the completion callback passes as
* `data` (records a conversion to dyn keeps by reference: declared fields,
* no index signature), or null when a call passes another value, the
* callback is used other than by a call, or no call passes a record. */
function completionDataRecord(
lowerer: Lowerer,
body: ts.Node,
symbol: ts.Symbol,
declaration: ts.Node,
): IrType | null {
let ok = true;
const found: { record: (IrType & { kind: "record" }) | null } = { record: null };
const visit = (node: ts.Node): void => {
if (!ok) return;
if (
ts.isIdentifier(node) &&
node !== declaration &&
lowerer.checker.getSymbolAtLocation(node) === symbol
) {
const call = node.parent;
if (!call || !ts.isCallExpression(call) || call.expression !== node) {
ok = false;
return;
}
const data = call.arguments.nodes[1];
if (data !== undefined) {
let t: IrType | null;
try {
t = lowerer.mapTypeOf(lowerer.typeOf(data));
} catch {
t = null;
}
if (t?.kind === "undefinedT") return;
const shape = t?.kind === "record" ? lowerer.shapes.get(t.shapeId) : undefined;
if (
t?.kind !== "record" ||
shape === undefined ||
shape.fields.length === 0 ||
shape.indexValue !== undefined ||
(found.record !== null && found.record.shapeId !== t.shapeId)
) {
ok = false;
return;
}
found.record = t;
}
}
ts.forEachChild(node, visit);
};
visit(body);
return ok ? found.record : null;
}
/** Whether every reference to the completion callback `symbol` in `body`
* calls it with a `data` argument that is a Buffer, a string, null or
* undefined (or with no data). */
@@ -989,7 +1073,9 @@ function parseStreamOptions(
continue;
}
const full = tuples[name]!.map((t) => (t === "fn" ? null : t));
const pinned = out.readableObject ? null : pinAnyDataCompletion(lowerer, name, fnNode);
const pinned = out.readableObject
? pinTypedDataCompletion(lowerer, name, fnNode)
: pinAnyDataCompletion(lowerer, name, fnNode);
let anyChunk: ts.Symbol | undefined;
try {
// The completion-callback tail: typed by the checker contextually;
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-boxed-kinds.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-boxed-kinds.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-data-and-pipe.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-data-and-pipe.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-for-await.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-for-await.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-identity.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-identity.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mismatch.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-mismatch.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mutation.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-mutation.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-read-paused.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-read-paused.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-weakmap.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-weakmap.ts"
],
"diags": []
}
packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-writev.ts.json
+6
View File
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-typed-chunk-writev.ts"
],
"diags": []
}
packages/runtime/src/scr_stream.c
+210 -63
View File
@@ -80,11 +80,13 @@ static ScrBytes *scr_stream_buffer(ScrBytes *b) {
/* ── state ────────────────────────────────────────────────────────────── */
typedef struct ScrWEntry {
ScrBytes *chunk; /* owned; a ScrDyn* value on an object-mode writable side */
ScrBytes *chunk; /* owned; a ScrDyn* value on an object-mode writable side,
* or a typed record when desc is set */
ScrClosure *cb; /* owned or NULL (the user's write(chunk, cb) callback) */
ScrDyn *enc; /* owned or NULL = undefined: the encoding argument an
* object-mode write passed (Node hands it to _write
* unchanged; null when the callback took its place) */
const ScrStreamTypedDesc *desc; /* the typed chunk's type, or NULL */
} ScrWEntry;
struct ScrStreamState {
@@ -150,6 +152,11 @@ struct ScrStreamState {
* Buffer, or both kinds: every entry is an owned ScrDyn*
* (scr_stream_add_value). Takes precedence over `encoded`. */
bool dyn_entries;
/* The type of the typed records an object-mode buffer holds
* (scr_stream_push_typed), or NULL. Set by the first entry and kept:
* an entry taken out of the buffer is released through it even after
* the buffer turned to dyn entries, which take precedence. */
const ScrStreamTypedDesc *typed;
/* The objectMode or readableObjectMode option made the buffer hold
* object entries (a stream without _read is not a seeded
* Readable.from source). */
@@ -220,6 +227,9 @@ struct ScrStreamState {
bool final_pending; /* _final called, its callback not yet run */
ScrClosure *inflight_cb; /* owned or NULL */
ScrDyn *cur_enc; /* borrowed: the running write's encoding argument */
/* the running write's typed chunk type, or NULL (a dyn value or a
* Buffer) */
const ScrStreamTypedDesc *cur_desc;
ScrClosure *write_cb;
ScrStreamChunkInv write_inv;
ScrClosure *final_cb;
@@ -341,12 +351,21 @@ static bool scr_stream_is(const ScrEmitter *em) {
/* Readable-buffer entry accessors: bytes chunks, strings once an
* encoding is set (lengths then count JS string units, Node's encoded
* accounting), or checked-dynamic values in an object-mode buffer that
* holds other values. */
typedef enum { SCR_ENTRY_BYTES, SCR_ENTRY_STR, SCR_ENTRY_DYN } ScrEntryKind;
* accounting), checked-dynamic values in an object-mode buffer that holds
* other values, or typed records of the buffer's one type (r.typed). */
typedef enum { SCR_ENTRY_BYTES, SCR_ENTRY_STR, SCR_ENTRY_DYN, SCR_ENTRY_TYPED } ScrEntryKind;
static ScrEntryKind scr_stream_rkind(const ScrStreamState *st) {
return st->r.dyn_entries ? SCR_ENTRY_DYN : st->r.encoded ? SCR_ENTRY_STR : SCR_ENTRY_BYTES;
return st->r.dyn_entries ? SCR_ENTRY_DYN
: st->r.typed ? SCR_ENTRY_TYPED
: st->r.encoded ? SCR_ENTRY_STR
: SCR_ENTRY_BYTES;
}
/* Two descriptors of one type: the same constant, or the same type key
* (separately emitted constants). */
static bool scr_stream_desc_same(const ScrStreamTypedDesc *a, const ScrStreamTypedDesc *b) {
return a == b || (a->key_len == b->key_len && memcmp(a->key, b->key, a->key_len) == 0);
}
static size_t scr_stream_entry_len(const ScrStreamState *st, void *e) {
@@ -357,21 +376,25 @@ static size_t scr_stream_entry_len(const ScrStreamState *st, void *e) {
/* Releases an entry by the kind it had when it left the buffer: a 'data'
* listener may call setEncoding() or push another kind of value while the
* entry is out. */
static void scr_stream_entry_release_kind(void *e, ScrEntryKind kind) {
static void scr_stream_entry_release_kind(const ScrStreamState *st, void *e, ScrEntryKind kind) {
if (kind == SCR_ENTRY_DYN) scr_dyn_release((ScrDyn *)e);
else if (kind == SCR_ENTRY_TYPED) st->r.typed->release(e);
else if (kind == SCR_ENTRY_STR) scr_str_release((ScrStr *)e);
else scr_bytes_release((ScrBytes *)e);
}
static void scr_stream_entry_release(const ScrStreamState *st, void *e) {
scr_stream_entry_release_kind(e, scr_stream_rkind(st));
scr_stream_entry_release_kind(st, e, scr_stream_rkind(st));
}
/* An entry of the given kind as a dyn value (+1; moves entry). */
static ScrDyn *scr_stream_entry_box(void *entry, ScrEntryKind kind) {
static ScrDyn *scr_stream_entry_box(const ScrStreamState *st, void *entry, ScrEntryKind kind) {
if (kind == SCR_ENTRY_DYN) return entry;
ScrDyn *d;
if (kind == SCR_ENTRY_STR) {
if (kind == SCR_ENTRY_TYPED) {
d = st->r.typed->box(entry);
st->r.typed->release(entry);
} else if (kind == SCR_ENTRY_STR) {
d = scr_dyn_new_str((ScrStr *)entry);
scr_str_release((ScrStr *)entry);
} else {
@@ -394,13 +417,19 @@ static size_t scr_stream_wchunk_len(const ScrStreamState *st, ScrBytes *chunk) {
return 1;
}
static ScrBytes *scr_stream_wchunk_retain(const ScrStreamState *st, ScrBytes *chunk) {
/* desc: the chunk's type when it is a typed record (an object-mode side's
* typed write), else NULL. */
static ScrBytes *scr_stream_wchunk_retain(const ScrStreamState *st, ScrBytes *chunk,
const ScrStreamTypedDesc *desc) {
if (desc) return desc->retain(chunk);
if (st->w.values) return (ScrBytes *)scr_dyn_retain((ScrDyn *)chunk);
return scr_bytes_retain(chunk);
}
static void scr_stream_wchunk_release(const ScrStreamState *st, ScrBytes *chunk) {
if (st->w.values) scr_dyn_release((ScrDyn *)chunk);
static void scr_stream_wchunk_release(const ScrStreamState *st, ScrBytes *chunk,
const ScrStreamTypedDesc *desc) {
if (desc) desc->release(chunk);
else if (st->w.values) scr_dyn_release((ScrDyn *)chunk);
else scr_bytes_release(chunk);
}
@@ -442,8 +471,10 @@ static void scr_stream_state_drop(ScrStreamState *st, bool gc) {
for (size_t i = 0; i < st->drain_srcs.n; i++) scr_stream_release(st->drain_srcs.src[i]);
}
for (size_t i = st->w.head; i < st->w.n; i++) {
bool traced = st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk);
if (!gc || !traced) scr_stream_wchunk_release(st, st->w.q[i].chunk);
/* typed chunks are of types the collector does not trace */
const ScrStreamTypedDesc *desc = st->w.q[i].desc;
bool traced = !desc && st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk);
if (!gc || !traced) scr_stream_wchunk_release(st, st->w.q[i].chunk, desc);
if (!gc && st->w.q[i].cb) scr_closure_release(st->w.q[i].cb);
ScrDyn *enc = st->w.q[i].enc;
if (enc && (!gc || !scr_stream_dyn_entry_traced(enc))) scr_dyn_release(enc);
@@ -490,7 +521,7 @@ void scr_stream_st_trace(ScrStreamState *st, ScrTraceVisit visit, void *ctx) {
for (size_t i = 0; i < st->fin.n; i++) visit(st->fin.cb[i], ctx);
for (size_t i = st->w.head; i < st->w.n; i++) {
if (st->w.q[i].cb) visit(st->w.q[i].cb, ctx);
if (st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk))
if (!st->w.q[i].desc && st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk))
visit(st->w.q[i].chunk, ctx);
if (st->w.q[i].enc && scr_stream_dyn_entry_traced(st->w.q[i].enc)) visit(st->w.q[i].enc, ctx);
}
@@ -896,18 +927,20 @@ static void *scr_stream_rbuf_take(ScrStreamState *st, size_t n) {
* slot, so at most one of the first two is non-NULL; the compiler's
* stream-data listener thunks dispatch on them (typed listeners unwrap
* their declared side; dyn listeners box by tag or take the value). */
static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) {
static void scr_stream_emit_data_entry(ScrStream *s, void *chunk, ScrEntryKind kind) {
ScrStreamState *st = s->st;
ScrBytes *b = NULL;
ScrStr *str = NULL;
ScrDyn *value = NULL;
/* a typed record, delivered to pipes as one (no 'data' listeners) */
const ScrStreamTypedDesc *typed = kind == SCR_ENTRY_TYPED ? st->r.typed : NULL;
if (kind == SCR_ENTRY_DYN) {
value = chunk;
if (value->kind == SCR_DYN_BYTES) b = value->v.bytes;
else if (value->kind == SCR_DYN_STR) str = value->v.str;
} else if (kind == SCR_ENTRY_STR) {
str = chunk;
} else {
} else if (kind == SCR_ENTRY_BYTES) {
b = chunk;
}
/* emit('data') with no listeners does nothing; a piped stream delivers
@@ -932,9 +965,10 @@ static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) {
if (scr_exc_pending()) return;
continue; /* the pipes array shifted down over i */
}
bool ok = value ? scr_stream_write_dyn(dst, value, NULL)
: b ? scr_stream_write(dst, b, NULL)
: scr_stream_write_str(dst, str, NULL);
bool ok = typed ? scr_stream_write_typed(dst, chunk, typed, NULL)
: value ? scr_stream_write_dyn(dst, value, NULL)
: b ? scr_stream_write(dst, b, NULL)
: scr_stream_write_str(dst, str, NULL);
if (scr_exc_pending()) return;
i++;
if (!ok && !dst->st->destroyed) {
@@ -958,6 +992,20 @@ static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) {
}
}
/* A typed record reaches 'data' listeners as its capsule, and the pipes
* then receive that capsule too; without listeners the pipes receive the
* record. */
static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) {
ScrStreamState *st = s->st;
if (kind != SCR_ENTRY_TYPED || !st->r.data_listeners) {
scr_stream_emit_data_entry(s, chunk, kind);
return;
}
ScrDyn *v = st->r.typed->box(chunk);
scr_stream_emit_data_entry(s, v, SCR_ENTRY_DYN);
scr_dyn_release(v);
}
static void scr_stream_emit_readable_nt(ScrStream *s) {
ScrStreamState *st = s->st;
st->r.need_readable = false; /* Node's emitReadable clears the ask */
@@ -1076,7 +1124,7 @@ static bool scr_stream_add_chunk(ScrStream *s, ScrBytes *chunk, bool front) {
if (st->r.emitted_readable) st->r.emitted_readable = false;
ScrEntryKind kind = scr_stream_rkind(st);
scr_stream_emit_data(s, entry, kind);
scr_stream_entry_release_kind(entry, kind);
scr_stream_entry_release_kind(st, entry, kind);
if (scr_exc_pending()) return false;
} else {
scr_stream_rbuf_push(st, entry, scr_stream_rkind(st), front);
@@ -1188,7 +1236,7 @@ static void *scr_stream_read_n_kind(ScrStream *s, double size, ScrEntryKind *kin
if (ret != NULL && !st->error_emitted && !st->close_emitted) {
scr_stream_emit_data(s, ret, *kind);
if (scr_exc_pending()) {
scr_stream_entry_release_kind(ret, *kind);
scr_stream_entry_release_kind(st, ret, *kind);
return NULL;
}
}
@@ -1201,7 +1249,7 @@ static void scr_stream_flow(ScrStream *s) {
ScrEntryKind kind;
void *b = scr_stream_read_n_kind(s, -1, &kind);
if (b == NULL) break;
scr_stream_entry_release_kind(b, kind);
scr_stream_entry_release_kind(st, b, kind);
}
}
@@ -1376,6 +1424,10 @@ static void scr_stream_next_fulfill(ScrStreamState *st, ScrPromise *w /*moves*/,
void *chunk /*moves or NULL=EOF*/, ScrEntryKind kind) {
void *payload;
bool dyn = st->r.next_dyn;
if (chunk != NULL && kind == SCR_ENTRY_TYPED) {
chunk = scr_stream_entry_box(st, chunk, kind);
kind = SCR_ENTRY_DYN;
}
if (!dyn && chunk != NULL && kind == SCR_ENTRY_DYN) {
ScrDyn *d = chunk;
if (d->kind == SCR_DYN_BYTES) {
@@ -1400,11 +1452,11 @@ static void scr_stream_next_fulfill(ScrStreamState *st, ScrPromise *w /*moves*/,
ScrDyn *result = scr_dyn_new_obj();
scr_dyn_obj_set(result, "value", 5,
chunk == NULL ? scr_dyn_retain(scr_dyn_undefined())
: scr_stream_entry_box(chunk, kind));
: scr_stream_entry_box(st, chunk, kind));
scr_dyn_obj_set(result, "done", 4, scr_dyn_new_bool(chunk == NULL));
payload = result;
} else if (dyn) {
payload = chunk == NULL ? scr_stream_from_done() : scr_stream_entry_box(chunk, kind);
payload = chunk == NULL ? scr_stream_from_done() : scr_stream_entry_box(st, chunk, kind);
} else {
payload = chunk ? (ScrBytes *)chunk : scr_bytes_new(SCR_BYTES_U8, 0);
}
@@ -1438,7 +1490,7 @@ static void scr_stream_iter_step(ScrStream *s) {
void *chunk = scr_stream_read_n_kind(s, -1, &kind);
st->r.next_reading = false;
if (scr_exc_pending()) {
if (chunk) scr_stream_entry_release_kind(chunk, kind);
if (chunk) scr_stream_entry_release_kind(st, chunk, kind);
ScrPromise *w = st->r.next_waiter;
st->r.next_waiter = NULL;
if (w) {
@@ -1984,7 +2036,7 @@ ScrStream *scr_stream_new_object_readable(ScrClosure *read, ScrStreamReadInv rea
static void scr_stream_rbuf_to_dyn(ScrStreamState *st) {
ScrEntryKind kind = scr_stream_rkind(st);
for (size_t i = st->r.first; i < st->r.first + st->r.n; i++)
st->r.buf[i] = scr_stream_entry_box(st->r.buf[i], kind);
st->r.buf[i] = scr_stream_entry_box(st, st->r.buf[i], kind);
st->r.dyn_entries = true;
st->r.object_kind_set = true;
}
@@ -2005,7 +2057,7 @@ static void *scr_stream_entry_convert_kind(ScrStreamState *st, void *entry, ScrE
return dec;
}
if (want != SCR_ENTRY_DYN) scr_stream_rbuf_to_dyn(st);
ScrDyn *d = scr_stream_entry_box(entry, *kind);
ScrDyn *d = scr_stream_entry_box(st, entry, *kind);
*kind = SCR_ENTRY_DYN;
return d;
}
@@ -2027,14 +2079,14 @@ static int scr_stream_add_entry(ScrStream *s, void *entry /*moves*/, ScrEntryKin
}
entry = scr_stream_entry_convert(st, entry, &kind);
if (st->r.ended) {
scr_stream_entry_release_kind(entry, kind);
scr_stream_entry_release_kind(st, entry, kind);
ScrError *e = scr_stream_mkerr("ERR_STREAM_PUSH_AFTER_EOF", "stream.push() after EOF");
scr_stream_error_or_destroy(s, e);
scr_error_release(e);
return 0;
}
if (st->destroyed || st->errored) {
scr_stream_entry_release_kind(entry, kind);
scr_stream_entry_release_kind(st, entry, kind);
return 0;
}
st->r.reading = false;
@@ -2042,7 +2094,7 @@ static int scr_stream_add_entry(ScrStream *s, void *entry /*moves*/, ScrEntryKin
(st->r.data_listeners || st->pipes.n > 0)) {
if (st->r.emitted_readable) st->r.emitted_readable = false;
scr_stream_emit_data(s, entry, kind);
scr_stream_entry_release_kind(entry, kind);
scr_stream_entry_release_kind(st, entry, kind);
if (scr_exc_pending()) return 0;
} else {
scr_stream_rbuf_push(st, entry, kind, false);
@@ -2053,6 +2105,24 @@ static int scr_stream_add_entry(ScrStream *s, void *entry /*moves*/, ScrEntryKin
return !st->r.ended && (st->r.length < st->r.hwm || st->r.length == 0);
}
/* push(rec) of a typed record: an object-mode buffer that holds records
* of its type, or nothing yet and no encoding, keeps the record itself;
* any other stream takes its capsule, as push(value) takes it (a typed
* buffer turns to dyn entries before an encoding is set). Borrows rec. */
bool scr_stream_push_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc) {
ScrStreamState *st = s->st;
if (st->has_r && st->r.object_entries && !st->r.dyn_entries &&
(st->r.typed ? scr_stream_desc_same(st->r.typed, desc)
: !st->r.object_kind_set && st->r.length == 0 && !st->r.encoded)) {
if (!st->r.typed) st->r.typed = desc;
return scr_stream_add_entry(s, desc->retain(rec), SCR_ENTRY_TYPED) > 0;
}
ScrDyn *v = desc->box(rec);
bool ret = scr_stream_push_dyn(s, v);
scr_dyn_release(v);
return ret;
}
/* push(value) on an object-mode stream: a string or a Buffer keeps its
* kind, any other value is a dyn entry. Borrows v. */
int scr_stream_push_object(ScrStream *s, const ScrDyn *v) {
@@ -2110,7 +2180,7 @@ static void scr_stream_on_listener(ScrEmitter *em, ScrStr *name) {
/* ── writable internals ───────────────────────────────────────────────── */
static void scr_stream_wq_push(ScrStreamState *st, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/,
ScrDyn *enc /*moves*/) {
ScrDyn *enc /*moves*/, const ScrStreamTypedDesc *desc) {
if (st->w.n == st->w.cap) {
st->w.cap = st->w.cap ? st->w.cap * 2 : 4;
st->w.q = realloc(st->w.q, st->w.cap * sizeof *st->w.q);
@@ -2119,32 +2189,41 @@ static void scr_stream_wq_push(ScrStreamState *st, ScrBytes *chunk /*moves*/, Sc
st->w.q[st->w.n].chunk = chunk;
st->w.q[st->w.n].cb = cb;
st->w.q[st->w.n].enc = enc;
st->w.q[st->w.n].desc = desc;
st->w.n++;
}
static void scr_stream_transform_callback(ScrStream *s);
static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/);
static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk);
static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/,
const ScrStreamTypedDesc *desc);
static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk,
const ScrStreamTypedDesc *desc);
static ScrDyn *scr_stream_wentry_encoding(const ScrStreamState *st, ScrDyn *enc);
/* One write reaching _write/_transform; the callback reads the write's
* encoding argument through w.cur_enc while it runs. */
* encoding argument through w.cur_enc, and the type of a typed chunk
* through w.cur_desc, while it runs. */
static void scr_stream_do_write(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/,
ScrDyn *enc /*moves*/) {
ScrDyn *enc /*moves*/, const ScrStreamTypedDesc *desc) {
ScrStreamState *st = s->st;
const ScrStreamTypedDesc *outer_desc = st->w.cur_desc;
st->w.cur_desc = desc;
if (enc == NULL) {
scr_stream_do_write_inner(s, chunk, cb);
scr_stream_do_write_inner(s, chunk, cb, desc);
st->w.cur_desc = outer_desc;
return;
}
ScrDyn *outer = st->w.cur_enc;
st->w.cur_enc = enc;
scr_stream_do_write_inner(s, chunk, cb);
scr_stream_do_write_inner(s, chunk, cb, desc);
st->w.cur_enc = outer;
st->w.cur_desc = outer_desc;
scr_dyn_release(enc);
}
static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/) {
static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/,
const ScrStreamTypedDesc *desc) {
ScrStreamState *st = s->st;
st->w.writing = true;
st->w.wsync = true;
@@ -2153,16 +2232,17 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S
if (st->is_transform || st->passthrough) st->xf_length = st->r.length;
if (st->passthrough && st->transform_cb == NULL) {
/* identity: push through, complete synchronously */
if (st->w.values) scr_stream_push_dyn(s, (ScrDyn *)chunk);
if (desc) scr_stream_push_typed(s, chunk, desc);
else if (st->w.values) scr_stream_push_dyn(s, (ScrDyn *)chunk);
else scr_stream_add_chunk(s, chunk, false);
scr_stream_wchunk_release(st, chunk);
scr_stream_wchunk_release(st, chunk, desc);
if (!scr_exc_pending()) scr_stream_transform_callback(s);
st->w.wsync = false;
return;
}
if (st->is_transform || st->passthrough) {
if (st->transform_cb == NULL) {
scr_stream_wchunk_release(st, chunk);
scr_stream_wchunk_release(st, chunk, desc);
ScrError *e = scr_stream_mkerr("ERR_METHOD_NOT_IMPLEMENTED", "The _transform() method is not implemented");
st->w.wsync = false;
scr_stream_error_or_destroy(s, e);
@@ -2170,7 +2250,7 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S
return;
}
st->transform_inv(st->transform_cb, s, chunk);
scr_stream_wchunk_release(st, chunk);
scr_stream_wchunk_release(st, chunk, desc);
st->w.wsync = false;
return;
}
@@ -2178,17 +2258,17 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S
/* Writable.prototype._write: this._writev([{ chunk, encoding }], cb) */
ScrDyn *chunks = scr_dyn_new_arr();
ScrDyn *entry = scr_dyn_new_obj();
scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, chunk));
scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, chunk, desc));
scr_dyn_obj_set(entry, "encoding", 8, scr_stream_wentry_encoding(st, st->w.cur_enc));
scr_dyn_arr_push(chunks, entry);
scr_stream_wchunk_release(st, chunk);
scr_stream_wchunk_release(st, chunk, desc);
st->w.writev_inv(st->w.writev_cb, s, chunks);
scr_dyn_release(chunks);
st->w.wsync = false;
return;
}
if (st->w.write_cb == NULL) {
scr_stream_wchunk_release(st, chunk);
scr_stream_wchunk_release(st, chunk, desc);
ScrError *e = scr_stream_mkerr("ERR_METHOD_NOT_IMPLEMENTED", "The _write() method is not implemented");
st->w.wsync = false;
scr_stream_error_or_destroy(s, e);
@@ -2196,7 +2276,7 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S
return;
}
st->w.write_inv(st->w.write_cb, s, chunk);
scr_stream_wchunk_release(st, chunk);
scr_stream_wchunk_release(st, chunk, desc);
st->w.wsync = false;
}
@@ -2263,7 +2343,7 @@ static void scr_stream_error_buffer(ScrStream *s) {
while (st->w.head < st->w.n && !scr_exc_pending()) {
ScrWEntry e = st->w.q[st->w.head++];
st->w.length -= scr_stream_wchunk_len(st, e.chunk);
scr_stream_wchunk_release(st, e.chunk);
scr_stream_wchunk_release(st, e.chunk, e.desc);
if (e.enc) scr_dyn_release(e.enc);
if (e.cb) {
SCR_WCB_CALL_ERROF(e.cb, st);
@@ -2360,11 +2440,13 @@ static SCR_COLD void scr_stream_writev_buffer(ScrStream *s) {
}
st->w.head = 0;
st->w.n = 0;
scr_stream_do_write(s, all, cbs, NULL);
scr_stream_do_write(s, all, cbs, NULL, NULL);
}
/* A write's chunk and encoding as _writev receives them (+1 each). */
static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk) {
static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk,
const ScrStreamTypedDesc *desc) {
if (desc) return desc->box(chunk);
if (st->w.values) return scr_dyn_retain((ScrDyn *)chunk);
return chunk->is_buffer ? scr_dyn_new_buffer(chunk) : scr_dyn_new_bytes(chunk);
}
@@ -2451,11 +2533,11 @@ static void scr_stream_writev_batch(ScrStream *s) {
if (!(encoding->kind == SCR_DYN_STR && encoding->v.str->len == 6 &&
memcmp(encoding->v.str->data, "buffer", 6) == 0))
all_buffers = false;
scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, e.chunk));
scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, e.chunk, e.desc));
scr_dyn_obj_set(entry, "encoding", 8, encoding);
scr_dyn_obj_set(entry, "callback", 8, scr_stream_writev_entry_cb(e.cb));
scr_dyn_arr_push(chunks, entry);
scr_stream_wchunk_release(st, e.chunk);
scr_stream_wchunk_release(st, e.chunk, e.desc);
if (e.enc) scr_dyn_release(e.enc);
if (e.cb) {
cbs->caps[k] = scr_box_new_obj(&scr_closure_retain_v, &scr_closure_release_v,
@@ -2489,7 +2571,7 @@ static void scr_stream_clear_buffer_entries(ScrStream *s) {
st->w.head = 0;
st->w.n = 0;
}
scr_stream_do_write(s, e.chunk, e.cb, e.enc);
scr_stream_do_write(s, e.chunk, e.cb, e.enc, e.desc);
}
st->w.buffer_processing = false;
}
@@ -4851,7 +4933,9 @@ static void scr_compose_destroyed(ScrCompose *c, ScrError *err) {
/* d._write: head.write(chunk) ? callback() : (ondrain = callback) */
static void scr_compose_write_inv(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) {
ScrCompose *c = scr_box_get_ref(cb->caps[0]);
bool ok = scr_stream_writes_values(s) ? scr_stream_write_dyn(c->head, (ScrDyn *)chunk, NULL)
const ScrStreamTypedDesc *desc = s->st->w.cur_desc;
bool ok = desc ? scr_stream_write_typed(c->head, chunk, desc, NULL)
: scr_stream_writes_values(s) ? scr_stream_write_dyn(c->head, (ScrDyn *)chunk, NULL)
: scr_stream_write(c->head, chunk, NULL);
if (ok) {
if (!scr_exc_pending()) scr_stream_write_done(s, NULL);
@@ -5214,7 +5298,9 @@ static void scr_dupx_r_eos(ScrClosure *cb, ScrStream *s, ScrError *err) {
/* d._write: w.write(chunk) ? callback() : (ondrain = callback) */
static void scr_dupx_write_inv(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) {
ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
bool ok = scr_stream_writes_values(s) ? scr_stream_write_dyn(x->w, (ScrDyn *)chunk, NULL)
const ScrStreamTypedDesc *desc = s->st->w.cur_desc;
bool ok = desc ? scr_stream_write_typed(x->w, chunk, desc, NULL)
: scr_stream_writes_values(s) ? scr_stream_write_dyn(x->w, (ScrDyn *)chunk, NULL)
: scr_stream_write(x->w, chunk, NULL);
if (ok) {
if (!scr_exc_pending()) scr_stream_write_done(s, NULL);
@@ -5265,7 +5351,7 @@ static void scr_dupx_pull(ScrDuplexify *x) {
return;
}
bool more;
if (d->st->r.object_entries) {
if (d->st->r.object_entries && kind != SCR_ENTRY_TYPED) {
more = scr_stream_add_entry(d, chunk, kind) > 0;
} else if (kind == SCR_ENTRY_STR) {
more = scr_stream_push_str(d, chunk);
@@ -5273,6 +5359,9 @@ static void scr_dupx_pull(ScrDuplexify *x) {
} else if (kind == SCR_ENTRY_DYN) {
more = scr_stream_push_dyn(d, chunk);
scr_dyn_release(chunk);
} else if (kind == SCR_ENTRY_TYPED) {
more = scr_stream_push_typed(d, chunk, x->r->st->r.typed);
x->r->st->r.typed->release(chunk);
} else {
more = scr_stream_push(d, chunk);
scr_bytes_release(chunk);
@@ -6755,7 +6844,7 @@ bool scr_stream_emit_value(ScrStream *s, ScrStr *name, ScrDyn *v) {
ScrBytes *scr_stream_chunk_bytes(ScrStream *s, ScrBytes *chunk) {
if (!scr_stream_writes_values(s)) return scr_bytes_retain(chunk);
ScrDyn *v = (ScrDyn *)chunk;
if (v->kind == SCR_DYN_BYTES) return scr_bytes_raw_view(v->v.bytes);
if (!s->st->w.cur_desc && v->kind == SCR_DYN_BYTES) return scr_bytes_raw_view(v->v.bytes);
static const char msg[] =
"a stream callback declaring a Buffer chunk received an object-mode value that is not a Buffer";
scr_throw_error_msg(SCR_ERR_TYPE, msg, sizeof msg - 1);
@@ -6763,10 +6852,17 @@ ScrBytes *scr_stream_chunk_bytes(ScrStream *s, ScrBytes *chunk) {
}
ScrDyn *scr_stream_chunk_value(ScrStream *s, ScrBytes *chunk) {
const ScrStreamTypedDesc *desc = s->st->w.cur_desc;
if (desc) return desc->box(chunk);
if (scr_stream_writes_values(s)) return scr_dyn_retain((ScrDyn *)chunk);
return scr_dyn_new_buffer(chunk);
}
void *scr_stream_chunk_typed(ScrStream *s, ScrBytes *chunk, const ScrStreamTypedDesc *desc) {
const ScrStreamTypedDesc *cur = s->st->w.cur_desc;
return cur && scr_stream_desc_same(cur, desc) ? desc->retain(chunk) : NULL;
}
ScrStr *scr_stream_chunk_encoding(ScrStream *s) {
if (!scr_stream_writes_values(s)) return scr_str_new("buffer", 6);
ScrDyn *enc = s->st->w.cur_enc;
@@ -7018,7 +7114,7 @@ ScrDyn *scr_stream_read_dyn(ScrStream *s, ScrDyn *size) {
scr_bytes_release(chunk);
return result;
}
return scr_stream_entry_box(chunk, kind);
return scr_stream_entry_box(s->st, chunk, kind);
}
/* read() answers STRINGS once encoded: the static result type here is
@@ -7041,6 +7137,10 @@ ScrBytes *scr_stream_read(ScrStream *s, double size) {
}
ScrEntryKind kind;
void *chunk = scr_stream_read_n_kind(s, size, &kind);
if (chunk && kind == SCR_ENTRY_TYPED) {
chunk = scr_stream_entry_box(s->st, chunk, kind);
kind = SCR_ENTRY_DYN;
}
if (chunk && kind == SCR_ENTRY_DYN) {
/* an object-mode value: a Buffer is the result, anything else is not */
ScrDyn *d = chunk;
@@ -7074,6 +7174,8 @@ ScrBytes *scr_stream_read(ScrStream *s, double size) {
ScrStream *scr_stream_set_encoding(ScrStream *s, ScrStr *enc) {
ScrStreamState *st = s->st;
if (!st->has_r) return scr_stream_retain(s);
/* typed records are values like the dyn entries */
if (st->r.typed && !st->r.dyn_entries) scr_stream_rbuf_to_dyn(st);
if (st->r.dyn_entries) {
/* an object-mode buffer of values: Buffers pushed later decode */
if (st->r.enc) scr_str_release(st->r.enc);
@@ -7215,8 +7317,9 @@ ScrStream *scr_stream_unpipe(ScrStream *src, ScrStream *dst) {
/* ── the writable surface ─────────────────────────────────────────────── */
static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed*/,
ScrClosure *cb /*moves*/, ScrDyn *enc /*borrowed*/) {
static bool scr_stream_write_chunk_desc(ScrStream *s, ScrBytes *chunk /*borrowed*/,
ScrClosure *cb /*moves*/, ScrDyn *enc /*borrowed*/,
const ScrStreamTypedDesc *desc) {
ScrStreamState *st = s->st;
if (!st->has_w || st->destroyed || st->w.ending) {
ScrError *e = st->has_w && !st->destroyed
@@ -7236,9 +7339,9 @@ static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed*
st->w.length += scr_stream_wchunk_len(st, chunk);
if (enc) enc = scr_dyn_retain(enc);
if (st->w.writing || st->w.corked > 0 || st->constructing) {
scr_stream_wq_push(st, scr_stream_wchunk_retain(st, chunk), cb, enc);
scr_stream_wq_push(st, scr_stream_wchunk_retain(st, chunk, desc), cb, enc, desc);
} else {
scr_stream_do_write(s, scr_stream_wchunk_retain(st, chunk), cb, enc);
scr_stream_do_write(s, scr_stream_wchunk_retain(st, chunk, desc), cb, enc, desc);
}
/* Node computes the answer AFTER the (possibly synchronous) write:
* a sync completion has already drained its bytes from length. Only
@@ -7249,6 +7352,11 @@ static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed*
return below && !st->destroyed && !st->errored;
}
static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed*/,
ScrClosure *cb /*moves*/, ScrDyn *enc /*borrowed*/) {
return scr_stream_write_chunk_desc(s, chunk, cb, enc, NULL);
}
static bool scr_stream_write_chunk(ScrStream *s, ScrBytes *chunk /*borrowed*/, ScrClosure *cb /*moves*/) {
return scr_stream_write_chunk_enc(s, chunk, cb, NULL);
}
@@ -7264,6 +7372,19 @@ static bool scr_stream_write_value(ScrStream *s, ScrDyn *v, ScrClosure *cb) {
return scr_stream_write_chunk_enc(s, (ScrBytes *)v, cb, scr_stream_cb_enc(cb));
}
/* write(rec) of a typed record: on an object-mode side the record itself
* is the chunk; another side takes its capsule, as write(value) takes it.
* Borrows rec. */
bool scr_stream_write_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc,
ScrClosure *cb) {
if (s->st->has_w && s->st->w.object_mode)
return scr_stream_write_chunk_desc(s, rec, cb, scr_stream_cb_enc(cb), desc);
ScrDyn *v = desc->box(rec);
bool ret = scr_stream_write_dyn(s, v, cb);
scr_dyn_release(v);
return ret;
}
/* A Buffer write on a byte-mode side: with decodeStrings: false the
* queue holds values, so the Buffer is boxed with its 'buffer' encoding.
* Borrows chunk. */
@@ -7642,6 +7763,32 @@ void scr_stream_transform_done_value(ScrStream *s, ScrError *err /*moves*/, ScrD
else scr_stream_transform_callback(s);
}
void scr_stream_transform_done_typed(ScrStream *s, ScrError *err /*moves*/, void *rec,
const ScrStreamTypedDesc *desc) {
if (!err && rec) scr_stream_push_typed(s, rec, desc);
if (scr_exc_pending()) {
if (err) scr_error_release(err);
return;
}
if (err) scr_stream_write_done(s, err);
else scr_stream_transform_callback(s);
}
void scr_stream_flush_done_typed(ScrStream *s, ScrError *err /*moves*/, void *rec,
const ScrStreamTypedDesc *desc) {
s->st->flush_pending = false;
if (err) {
scr_stream_error_or_destroy(s, err);
scr_error_release(err);
return;
}
if (rec) scr_stream_push_typed(s, rec, desc);
if (scr_exc_pending()) return;
scr_stream_push_null(s);
if (scr_exc_pending()) return;
scr_stream_finish_maybe(s);
}
void scr_stream_flush_done_value(ScrStream *s, ScrError *err /*moves*/, ScrDyn *v) {
s->st->flush_pending = false;
if (err) {
@@ -8191,7 +8338,7 @@ static void scr_stream_run_tick(ScrStreamTick *t) {
if (st->has_r) {
ScrEntryKind kind;
void *chunk = scr_stream_read_n_kind(s, 0, &kind);
if (chunk) scr_stream_entry_release_kind(chunk, kind);
if (chunk) scr_stream_entry_release_kind(st, chunk, kind);
}
break;
}
packages/runtime/src/scr_stream.h
+31
View File
@@ -574,6 +574,37 @@ bool scr_stream_chunk_done(const struct ScrDyn *v);
void scr_stream_transform_done_value(ScrStream *s, ScrError *err, struct ScrDyn *v);
void scr_stream_flush_done_value(ScrStream *s, ScrError *err, struct ScrDyn *v);
/* Typed object-mode chunks. A compiled push, completion or write of a
* record whose type has a descriptor (one internal constant per type,
* emitted by the compiler) keeps the record itself in the stream's
* buffers. A consumer that reads the chunk as a checked-dynamic value
* gets the record's capsule from `box`, the conversion compiled code
* uses for the record (same referent, so the same JS object). A side
* that is not in object mode, or a buffer that holds other values, takes
* the chunk boxed, as push(value) takes it. */
typedef struct ScrStreamTypedDesc {
const char *key; /* the type key (the capsule's type key) */
size_t key_len;
void *(*retain)(void *);
void (*release)(void *);
struct ScrDyn *(*box)(void *rec); /* +1 capsule; borrows rec */
} ScrStreamTypedDesc;
/* push(rec): borrows rec. */
bool scr_stream_push_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc);
/* write(rec[, cb]): borrows rec; cb moves. */
bool scr_stream_write_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc,
ScrClosure *cb);
/* callback(err, rec) of a transform or flush: rec borrowed, NULL when the
* callback passed no data; err moves. */
void scr_stream_transform_done_typed(ScrStream *s, ScrError *err, void *rec,
const ScrStreamTypedDesc *desc);
void scr_stream_flush_done_typed(ScrStream *s, ScrError *err, void *rec,
const ScrStreamTypedDesc *desc);
/* A compiled write or transform callback's chunk declared with desc's
* type: the record (+1) when the write in progress carries one of that
* type, else NULL (the caller converts scr_stream_chunk_value). */
void *scr_stream_chunk_typed(ScrStream *s, ScrBytes *chunk, const ScrStreamTypedDesc *desc);
/* Runtime-backed streams (the fs streams in scr_fs_stream.c).
* - construct: Node's _construct. The hook runs on a tick after
* construction; until construct_done's tick, reads, buffered writes,
tests/corpus/stream-typed-chunk-boxed-kinds.ts
+89
View File
@@ -0,0 +1,89 @@
// Arrays, union-typed values, strings and Buffers in object-mode streams,
// alone and mixed with records in one stream, reach their consumers as
// the values that were pushed.
import { Readable, Transform, Writable } from "node:stream";
type Pair = { a: number; b: number };
type Shape = { kind: "circle"; r: number } | { kind: "square"; side: number };
const log: string[] = [];
const arrays: number[][] = [[1, 2], [3], []];
const shapes: Shape[] = [
{ kind: "circle", r: 2 },
{ kind: "square", side: 3 },
];
const first: Pair = { a: 1, b: 2 };
const second: Pair = { a: 3, b: 4 };
const pairs: Pair[] = [first, second];
const arraySource = new Readable({ objectMode: true, read() {} });
for (const arr of arrays) arraySource.push(arr);
arraySource.push(null);
const arraySink = new Writable({
objectMode: true,
write(arr: number[], _enc, cb) {
log.push(`array [${arr.join(",")}] same=${arrays.includes(arr)}`);
cb();
},
});
const shapeSource = new Readable({ objectMode: true, read() {} });
for (const shape of shapes) shapeSource.push(shape);
shapeSource.push(null);
const area = new Transform({
objectMode: true,
transform(shape: Shape, _enc, cb) {
const size = shape.kind === "circle" ? shape.r : shape.side;
log.push(`shape ${shape.kind} ${size} same=${shapes.includes(shape)}`);
cb(null, shape);
},
});
const shapeSink = new Writable({
objectMode: true,
write(shape: Shape, _enc, cb) {
log.push(`shape out ${shape.kind} same=${shapes.includes(shape)}`);
cb();
},
});
const mixed = new Readable({ objectMode: true, read() {} });
mixed.push(first);
mixed.push("text");
mixed.push(second);
mixed.push(Buffer.from("buf"));
mixed.push(null);
const mixedSink = new Writable({
objectMode: true,
write(value: any, _enc, cb) {
if (typeof value === "string") log.push(`mixed string ${value}`);
else if (Buffer.isBuffer(value)) log.push(`mixed buffer ${value.toString()}`);
else log.push(`mixed pair ${value.a},${value.b} same=${pairs.includes(value)}`);
cb();
},
});
const later = new Readable({ objectMode: true, read() {} });
later.push("first");
later.push(first);
later.push(null);
const laterSink = new Writable({
objectMode: true,
write(value: any, _enc, cb) {
log.push(`later ${typeof value} ${value === first}`);
cb();
},
});
let finished = 0;
const done = (): void => {
finished++;
if (finished === 4) for (const line of log) console.log(line);
};
arraySink.on("finish", done);
shapeSink.on("finish", done);
mixedSink.on("finish", done);
laterSink.on("finish", done);
arraySource.pipe(arraySink);
shapeSource.pipe(area).pipe(shapeSink);
mixed.pipe(mixedSink);
later.pipe(laterSink);
tests/corpus/stream-typed-chunk-data-and-pipe.ts
+55
View File
@@ -0,0 +1,55 @@
// A 'data' listener and a pipe on one object-mode stream receive the same
// object; a property the listener adds is visible to the untyped
// consumer downstream, and the typed consumer sees the same object.
import { Readable, Writable } from "node:stream";
type Point = { x: number; y: number };
const points: Point[] = [];
let i = 0;
const source = new Readable({
objectMode: true,
read() {
if (i < 3) {
const p = { x: i, y: i * i };
points.push(p);
i++;
this.push(p);
} else {
this.push(null);
}
},
});
const log: string[] = [];
let lastData: any = null;
source.on("data", (p: any) => {
lastData = p;
p.label = `p${p.x}`;
log.push(`data ${p.x},${p.y} same=${p === points[p.x]}`);
});
const typed = new Writable({
objectMode: true,
write(p: Point, _enc, cb) {
log.push(`typed ${p.x},${p.y} same=${p === points[p.x]} last=${p === lastData}`);
cb();
},
});
const untyped = new Writable({
objectMode: true,
write(p: any, _enc, cb) {
log.push(`untyped ${p.x} ${p.label} keys=${Object.keys(p).join("/")} same=${p === points[p.x]}`);
cb();
},
});
let finished = 0;
const done = (): void => {
finished++;
if (finished === 2) {
for (const line of log) console.log(line);
console.log(JSON.stringify(points));
}
};
typed.on("finish", done);
untyped.on("finish", done);
source.pipe(typed);
source.pipe(untyped);
tests/corpus/stream-typed-chunk-for-await.ts
+44
View File
@@ -0,0 +1,44 @@
// for await over an object-mode Readable and over a Transform's output
// yields the objects that were pushed.
import { Readable, Transform } from "node:stream";
type Entry = { key: string; value: number };
const entries: Entry[] = [];
let n = 0;
const source = new Readable({
objectMode: true,
read() {
if (n < 4) {
const entry = { key: `k${n}`, value: n * 10 };
entries.push(entry);
n++;
this.push(entry);
} else {
this.push(null);
}
},
});
const outputs: Entry[] = [];
const double = new Transform({
objectMode: true,
transform(entry: Entry, _enc, cb) {
const out = { key: entry.key.toUpperCase(), value: entry.value * 2 };
outputs.push(out);
cb(null, out);
},
});
async function main(): Promise<void> {
const copy = new Readable({ objectMode: true, read() {} });
for (const entry of [{ key: "a", value: 1 }, { key: "b", value: 2 }]) copy.push(entry);
copy.push(null);
for await (const entry of copy) console.log(`copy ${entry.key}=${entry.value}`);
let index = 0;
for await (const entry of source.pipe(double)) {
console.log(`${entry.key}=${entry.value} same=${entry === outputs[index]} source=${entries.includes(entry)}`);
index++;
}
console.log(`count ${index}`);
}
void main();
tests/corpus/stream-typed-chunk-identity.ts
+54
View File
@@ -0,0 +1,54 @@
// Object-mode chunks keep their identity through a Readable, a Transform
// and a Writable: each stage receives the object the previous stage
// pushed or passed to its callback.
import { Readable, Transform, Writable } from "node:stream";
type Row = { id: number; qty: number };
const pushed: Row[] = [];
const made: Row[] = [];
const seen = new Map<Row, number>();
let produced = 0;
const source = new Readable({
objectMode: true,
highWaterMark: 2,
read() {
while (produced < 6) {
produced++;
const row = { id: produced, qty: produced * 3 };
pushed.push(row);
seen.set(row, produced);
if (!this.push(row)) return;
}
this.push(null);
},
});
const lines: string[] = [];
const pass = new Transform({
objectMode: true,
transform(row: Row, _enc, cb) {
lines.push(`transform ${row.id} same=${row === pushed[row.id - 1]} map=${seen.get(row)}`);
if (row.id % 2 === 0) {
cb(null, row);
return;
}
const next = { id: row.id, qty: row.qty + 1 };
made.push(next);
cb(null, next);
},
});
const sink = new Writable({
objectMode: true,
highWaterMark: 1,
write(row: Row, _enc, cb) {
const original = row === pushed[row.id - 1];
const fresh = made.includes(row);
lines.push(`write ${row.id} qty=${row.qty} original=${original} fresh=${fresh}`);
cb();
},
});
sink.on("finish", () => {
for (const line of lines) console.log(line);
console.log(`pushed ${pushed.length} made ${made.length}`);
});
source.pipe(pass).pipe(sink);
tests/corpus/stream-typed-chunk-mismatch.ts
+41
View File
@@ -0,0 +1,41 @@
// A Transform that declares fewer fields than the chunks carry: the
// fields it does not declare stay on the chunks it passes on, so the
// Writable after it, which declares them, reads them, and an untyped
// 'data' listener on the Transform lists them.
import { Readable, Transform, Writable } from "node:stream";
type Full = { id: number; name: string; score: number };
type Narrow = { id: number };
let k = 0;
const source = new Readable({
objectMode: true,
read() {
if (k < 3) {
this.push({ id: k, name: `n${k}`, score: k * 1.5 });
k++;
} else {
this.push(null);
}
},
});
const log: string[] = [];
const narrow = new Transform({
objectMode: true,
transform(row: Narrow, _enc, cb) {
log.push(`narrow ${row.id} ${JSON.stringify(row)}`);
cb(null, row);
},
});
narrow.on("data", (row: any) => log.push(`data ${Object.keys(row).join("/")} ${row.name}`));
const full = new Writable({
objectMode: true,
write(row: Full, _enc, cb) {
log.push(`full ${row.id} ${row.name} ${row.score} ${JSON.stringify(row)}`);
cb();
},
});
full.on("finish", () => {
for (const line of log) console.log(line);
});
source.pipe(narrow).pipe(full);
tests/corpus/stream-typed-chunk-mutation.ts
+38
View File
@@ -0,0 +1,38 @@
// Chunks are shared, not copied: a change to an object after push() is
// visible to the consumer that reads it later, and a change a Transform
// makes reaches the next stage.
import { PassThrough, Transform, Writable } from "node:stream";
type Cell = { row: number; value: number };
const cells: Cell[] = [];
const log: string[] = [];
const bump = new Transform({
objectMode: true,
transform(cell: Cell, _enc, cb) {
log.push(`transform ${cell.row} value=${cell.value}`);
cell.value += 100;
cb(null, cell);
},
});
const sink = new Writable({
objectMode: true,
write(cell: Cell, _enc, cb) {
log.push(`write ${cell.row} value=${cell.value} same=${cell === cells[cell.row]}`);
cb();
},
});
const source = new PassThrough({ objectMode: true });
source.pause();
for (let r = 0; r < 3; r++) {
const cell = { row: r, value: r };
cells.push(cell);
source.write(cell);
cell.value = r * 10;
}
sink.on("finish", () => {
for (const line of log) console.log(line);
console.log(cells.map((c) => c.value).join(","));
});
source.pipe(bump).pipe(sink);
source.end();
tests/corpus/stream-typed-chunk-read-paused.ts
+26
View File
@@ -0,0 +1,26 @@
// read() on a paused object-mode stream answers the pushed objects
// themselves, in order, one per call.
import { Readable } from "node:stream";
type Job = { id: number; done: boolean };
const jobs: Job[] = [];
const source = new Readable({ objectMode: true, read() {} });
for (let i = 0; i < 3; i++) {
const job = { id: i, done: false };
jobs.push(job);
source.push(job);
}
console.log(`length ${source.readableLength}`);
const first = source.read();
console.log(`first ${first.id} same=${first === jobs[0]}`);
first.done = true;
console.log(`jobs[0].done ${jobs[0]!.done}`);
source.push(null);
source.on("readable", () => {
let job: any;
while ((job = source.read()) !== null) {
console.log(`read ${job.id} same=${job === jobs[job.id]} done=${job.done}`);
}
});
source.on("end", () => console.log("end"));
tests/corpus/stream-typed-chunk-weakmap.ts
+53
View File
@@ -0,0 +1,53 @@
// A WeakMap and a WeakSet keyed by the objects a Readable pushes find
// them again in a typed Writable, in an untyped 'data' listener, and in
// an untyped Writable downstream.
import { PassThrough, Readable, Writable } from "node:stream";
type Item = { name: string; n: number };
const tags = new WeakMap<Item, string>();
const marked = new WeakSet<object>();
const items: Item[] = [];
for (let i = 0; i < 4; i++) {
const item = { name: `item${i}`, n: i };
items.push(item);
tags.set(item, `tag-${i}`);
if (i % 2 === 0) marked.add(item);
}
let next = 0;
const source = new Readable({
objectMode: true,
read() {
const item = items[next++];
if (item === undefined) this.push(null);
else this.push(item);
},
});
const out: string[] = [];
const typed = new Writable({
objectMode: true,
write(item: Item, _enc, cb) {
out.push(`typed ${item.name} ${tags.get(item)} ${marked.has(item)}`);
cb();
},
});
const middle = new PassThrough({ objectMode: true });
middle.on("data", (item: any) => {
out.push(`data ${item.name} ${tags.get(item)} ${marked.has(item)}`);
});
const untyped = new Writable({
objectMode: true,
write(item: any, _enc, cb) {
out.push(`untyped ${item.name} ${tags.get(item)} ${items.indexOf(item)}`);
cb();
},
});
let finished = 0;
const done = (): void => {
finished++;
if (finished === 2) for (const line of out) console.log(line);
};
typed.on("finish", done);
untyped.on("finish", done);
source.pipe(typed);
source.pipe(middle).pipe(untyped);
tests/corpus/stream-typed-chunk-writev.ts
+39
View File
@@ -0,0 +1,39 @@
// writev on an object-mode Writable receives the written objects in its
// batch, with their encodings, and write() receives them one at a time.
import { Writable } from "node:stream";
type Msg = { seq: number; text: string };
const sent: Msg[] = [];
const lines: string[] = [];
const batched = new Writable({
objectMode: true,
write(msg: Msg, _enc, cb) {
lines.push(`write ${msg.seq} same=${msg === sent[msg.seq]}`);
cb();
},
writev(chunks: Array<{ chunk: Msg; encoding: any }>, cb) {
const parts = chunks.map(
(c) => `${c.chunk.seq}:${c.chunk.text}:${c.chunk === sent[c.chunk.seq]}:${String(c.encoding)}`,
);
lines.push(`writev ${parts.join(" ")}`);
cb();
},
});
const make = (seq: number): Msg => {
const msg = { seq, text: `m${seq}` };
sent.push(msg);
return msg;
};
batched.write(make(0));
batched.cork();
batched.write(make(1));
batched.write(make(2));
batched.write(make(3));
process.nextTick(() => {
batched.uncork();
batched.end(make(4));
});
batched.on("finish", () => {
for (const line of lines) console.log(line);
});
tests/harness/corpus-costs/stream-typed-chunk-boxed-kinds.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2797, "san": 4727 }
tests/harness/corpus-costs/stream-typed-chunk-data-and-pipe.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2910, "san": 4166 }
tests/harness/corpus-costs/stream-typed-chunk-for-await.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2833, "san": 5275 }
tests/harness/corpus-costs/stream-typed-chunk-identity.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2525, "san": 4060 }
tests/harness/corpus-costs/stream-typed-chunk-mismatch.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2953, "san": 4748 }
tests/harness/corpus-costs/stream-typed-chunk-mutation.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2493, "san": 4446 }
tests/harness/corpus-costs/stream-typed-chunk-read-paused.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2615, "san": 4274 }
tests/harness/corpus-costs/stream-typed-chunk-weakmap.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2399, "san": 4694 }
tests/harness/corpus-costs/stream-typed-chunk-writev.ts.json
+1
View File
@@ -0,0 +1 @@
{ "plain": 2486, "san": 4615 }