mirror of
https://github.com/vercel-labs/scriptc.git
synced 2026-10-31 07:23:06 +08:00
Run untyped pipeline function stages and later iterable stages
A pipeline stage that arrives as an untyped value now follows Node's
stage ladder for more shapes. As the first stage, a function is a source
function: the pipeline calls it with { signal } while it is set up and
pumps its result, which must be iterable (ERR_INVALID_RETURN_VALUE
otherwise), and a { readable, writable } pair pumps its readable side.
After the first stage, the pipeline call is lowered once per combination
of its first three untyped stages being functions or not, and a typeof
test picks the branch at run time. A function stage is called with the
previous stage and { signal }; before the last stage its result must be
async iterable, and as the last stage a promise resolves the pipeline's
value and an async iterable feeds the PassThrough pipeline() returns
(SCR_PIPE_DEST_DYN). Any other value that is not a stream becomes
Duplex.from of it, which the next stage reads from without the previous
stage being piped into it, and Duplex.from's ERR_INVALID_ARG_TYPE for
the values it rejects.
An untyped stage followed by an untyped function stage hands its value to
that stage through a cell created per pipeline call, so the function
receives the previous stage's own iterator, as Node's makeAsyncIterable
answers it, instead of an async generator over it.
Untyped web streams, pairs after the first stage, and function stages
that return a stream throw the not-supported error.
RUNTIME_ABI_VERSION goes from 45 to 46 for the new stream exports.
This commit is contained in:
Chris Tate
@@ -4,7 +4,13 @@
|
||||
When an untyped value reaches `NodeJS.ReadableStream` or `NodeJS.WritableStream`, those two interfaces hold any value, so JavaScript stream objects (constructor-function streams, plain objects with `on` and `pipe`) pass through TypeScript functions typed with them and come back to JavaScript as the same objects.
|
||||
Member calls on such a value use the native stream when it is one and call the object's own members otherwise; `readable.pipe()` into a JavaScript destination follows Node's pipe protocol.
|
||||
The first `pipeline` stage may arrive as an untyped value that is a native stream, a `Cipheriv` or `Decipheriv` value, or an iterable: a string, an array, a generator object, or any value with a `Symbol.asyncIterator` or `Symbol.iterator` method, which the pipeline pumps into the next stage as Node does.
|
||||
Later `pipeline` stages and `for await` sources that arrive as untyped values must be native streams, or `Cipheriv` and `Decipheriv` values, at run time; other values throw a not-supported error.
|
||||
It may also be a source function, which the pipeline calls with `{ signal }` and whose iterable result it pumps, or a `{ readable, writable }` pair, whose readable side it pumps.
|
||||
Later `pipeline` stages that arrive as untyped values may be native streams, `Cipheriv` and `Decipheriv` values, functions, and iterables.
|
||||
A function is called with the previous stage and `{ signal }` as a function stage: before the last stage it must return an async iterable, and as the last stage it may return a promise, whose value resolves the pipeline, or an async iterable, which feeds the PassThrough that `pipeline()` returns.
|
||||
An untyped source function, function stage, or destination that returns a stream is unsupported.
|
||||
An iterable becomes `Duplex.from` of it, which the next stage reads from while the stages before it are not read, as in Node.
|
||||
A pipeline branches at run time on whether each of its first three untyped stages after the first is a function; a function in a later untyped stage, an untyped web stream, and a `{ readable, writable }` pair after the first stage throw a not-supported error.
|
||||
`for await` sources that arrive as untyped values must be native streams, or `Cipheriv` and `Decipheriv` values, at run time; other values throw a not-supported error.
|
||||
Native streams expose no `emit` to JavaScript, so a JavaScript stream whose own `pipe()` emits on its destination (the legacy `Stream` pipe) cannot pipe into one.
|
||||
`Readable.from` accepts strings, Buffers, arrays, Sets, Maps, Map and Set iterators, generators, async generators, and class iterables and async iterables, and in JavaScript any iterable or async iterable.
|
||||
In JavaScript, a sync iterator value with a `then` method that is not a promise, and an object whose `[Symbol.asyncIterator]` or `[Symbol.iterator]` method is a generator method, are unsupported.
|
||||
|
||||
@@ -72,7 +72,7 @@ describe.runIf(supported)("macOS arm64 native link info", () => {
|
||||
minimum_os: "14.0",
|
||||
},
|
||||
program: { object, entry_symbol: "main" },
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" },
|
||||
runtime_pack: {
|
||||
kind: "precompiled",
|
||||
package: "@scriptc/runtime-darwin-arm64",
|
||||
@@ -165,6 +165,6 @@ describe.runIf(supported)("macOS arm64 native link info", () => {
|
||||
() => null,
|
||||
(failure: { stderr?: string }) => failure,
|
||||
);
|
||||
expect(error?.stderr).toContain("scr_runtime_abi_v45");
|
||||
expect(error?.stderr).toContain("scr_runtime_abi_v46");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -671,11 +671,13 @@ export function emitGenericLibCall(host: LlvmEmitterContext, e: LibCallExpr): Ll
|
||||
// which also takes those streams.
|
||||
e.fn === "stream.unboxDyn" && host.usesDynBuiltinStream
|
||||
? "scr_dyn_builtin_stream_unbox"
|
||||
: // for await over a checked-dynamic value iterates a native
|
||||
// stream behind it (scr_dyn_for_await_stream.c)
|
||||
e.fn === "dyn.asyncIteratorOpen" && host.dynForAwaitStreams
|
||||
? "scr_stream_dyn_async_iterator_open"
|
||||
: LIB_FN_SYMS[e.fn]);
|
||||
: e.fn === "stream.unboxDynStage" && host.usesDynBuiltinStream
|
||||
? "scr_dyn_builtin_stream_unbox_stage"
|
||||
: // for await over a checked-dynamic value iterates a native
|
||||
// stream behind it (scr_dyn_for_await_stream.c)
|
||||
e.fn === "dyn.asyncIteratorOpen" && host.dynForAwaitStreams
|
||||
? "scr_stream_dyn_async_iterator_open"
|
||||
: LIB_FN_SYMS[e.fn]);
|
||||
if (sym === undefined) throw new LlvmUnsupportedError(`libCall:${e.fn}`, e.loc);
|
||||
// The runtime's Date functions take and answer time values: a Date
|
||||
// operand passes its time value, read at once from the borrowed object,
|
||||
|
||||
@@ -818,6 +818,9 @@ export const LIB_FN_SYMS: Record<string, string> = {
|
||||
"stream.stdioWritable": "scr_stream_stdio_writable",
|
||||
"stream.unboxDyn": "scr_stream_unbox_dyn",
|
||||
"stream.dynIterableSource": "scr_stream_dyn_iterable_source",
|
||||
"stream.dynSourceValue": "scr_stream_dyn_source_value",
|
||||
"stream.unboxDynStage": "scr_stream_unbox_dyn_stage",
|
||||
"stream.dynTransformResult": "scr_stream_dyn_transform_result",
|
||||
"stream.handleStream": "scr_stream_handle",
|
||||
"stream.pipeDyn": "scr_stream_pipe_dyn",
|
||||
"stream.unpipeDyn": "scr_stream_unpipe_dyn",
|
||||
|
||||
@@ -743,7 +743,7 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
|
||||
const helpers = packed(5 + n + k, helperCount);
|
||||
let at = 5 + n + k + helperCount;
|
||||
const take = (present: boolean): string => (present ? args[at++]!.name : "null");
|
||||
const box = take(((flags >> 1) & 3) === 2);
|
||||
const box = take(((flags >> 1) & 3) >= 2);
|
||||
const signal = take((flags & 32) !== 0);
|
||||
const abort = take((flags & 32) !== 0);
|
||||
const head =
|
||||
|
||||
@@ -2,8 +2,8 @@
|
||||
* of the link contract: a program object keeps this symbol undefined and the
|
||||
* matching runtime defines it, so incompatible manual links fail before the
|
||||
* program can start. */
|
||||
export const RUNTIME_ABI_VERSION = 45 as const;
|
||||
export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v45" as const;
|
||||
export const RUNTIME_ABI_VERSION = 46 as const;
|
||||
export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v46" as const;
|
||||
|
||||
/** Object consumption is useful today, but the complete scr_* surface may
|
||||
* still change before 1.0. The versioned marker prevents accidental mixing;
|
||||
|
||||
@@ -52,7 +52,7 @@ async function fixture(target: NativeTargetSpec, workers = false) {
|
||||
object_format: target.objectFormat,
|
||||
minimum_os: target.minimumOs,
|
||||
},
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" },
|
||||
compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple },
|
||||
macros: { executable: [], excluded: ["SCR_LIB"], sanitizer: "external-toolchain-required" },
|
||||
flavors: {
|
||||
|
||||
@@ -129,7 +129,7 @@ async function fixture(target: NativeTargetSpec = MACOS_ARM64_TARGET) {
|
||||
object_format: target.objectFormat,
|
||||
minimum_os: target.minimumOs,
|
||||
},
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" },
|
||||
compiler: {
|
||||
command: "clang",
|
||||
identity: "fixture clang",
|
||||
|
||||
@@ -2837,20 +2837,24 @@ export function streamClassAliasDecl(
|
||||
const PIPELINE_STAGE_EXPECTED =
|
||||
"of type function or an instance of Blob, ReadableStream, WritableStream, Stream, Iterable, AsyncIterable, or Promise or { readable, writable } pair";
|
||||
|
||||
/** A checked-dynamic pipeline stage (an untyped source or destination):
|
||||
* a boxed native stream unwraps at run time; other values (functions,
|
||||
* pairs) throw the not-supported Error there. */
|
||||
/** A checked-dynamic pipeline stage taken as a stream: a boxed native
|
||||
* stream unwraps at run time, and a value that is not stream-like or a
|
||||
* function (an iterable after the first stage) becomes Duplex.from of it,
|
||||
* as pipelineImpl's last branch does. Stream-like objects, web streams and
|
||||
* { readable, writable } pairs throw the not-supported Error there, and so
|
||||
* do functions, which reach this lowering only past
|
||||
* MAX_DYN_FUNCTION_STAGES. */
|
||||
function unboxDynStage(lowerer: Lowerer, v: IrExpr, node: ts.Expression): IrExpr {
|
||||
noteUntypedStreamFlow(lowerer, null);
|
||||
return {
|
||||
kind: "libCall",
|
||||
fn: "stream.unboxDyn",
|
||||
fn: "stream.unboxDynStage",
|
||||
args: [
|
||||
v,
|
||||
strLit(
|
||||
lowerer.dynNodeStreams
|
||||
? "pipeline stages other than native streams (JavaScript stream objects, iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet"
|
||||
: "pipeline stages other than native streams (iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet",
|
||||
? "pipeline stages other than native streams, iterables and functions (JavaScript stream objects, web streams, { readable, writable } pairs after the first stage, and functions after the first three untyped stages) are not supported yet"
|
||||
: "pipeline stages other than native streams, iterables and functions (web streams, { readable, writable } pairs after the first stage, and functions after the first three untyped stages) are not supported yet",
|
||||
locOf(node),
|
||||
),
|
||||
],
|
||||
@@ -2930,6 +2934,102 @@ const BOXED_SOURCE_KINDS: ReadonlySet<IrType["kind"]> = new Set([
|
||||
"string",
|
||||
]);
|
||||
|
||||
/** The most checked-dynamic stages after the first that a pipeline
|
||||
* branches on as functions; each one doubles the lowered pipeline calls. */
|
||||
const MAX_DYN_FUNCTION_STAGES = 3;
|
||||
|
||||
/** A pipeline's arguments for one of the run-time branches
|
||||
* lowerDynStagePipeline lowers. */
|
||||
interface DynStagePlan {
|
||||
/** the first stage when it is checked-dynamic */
|
||||
source: DynPipelineSource | null;
|
||||
/** argument values evaluated before the branches, in argument order */
|
||||
bound: ReadonlyMap<number, IrExpr>;
|
||||
/** a typed iterable first stage's source, lowered before the branches */
|
||||
preSource?: { closure: IrExpr; out: SyncGenType } | undefined;
|
||||
/** the checked-dynamic stages this branch takes as functions */
|
||||
fns: ReadonlySet<number>;
|
||||
}
|
||||
|
||||
/** pipeline() whose stages after the first include untyped values:
|
||||
* pipelineImpl calls a stage that is a function with the previous stage
|
||||
* and { signal }, and pipes into or takes any other value, so the call is
|
||||
* lowered once per combination of those stages being functions or not,
|
||||
* up to MAX_DYN_FUNCTION_STAGES of them, and a typeof test on each
|
||||
* picks the branch at run time. The arguments other than function stages
|
||||
* are evaluated first, in argument order. */
|
||||
function lowerDynStagePipeline(
|
||||
lowerer: Lowerer,
|
||||
stageArgs: readonly ts.Expression[],
|
||||
src: DynPipelineSource | null,
|
||||
loc: SrcLoc,
|
||||
build: (plan: DynStagePlan) => IrExpr,
|
||||
): IrExpr {
|
||||
const candidates: number[] = [];
|
||||
for (let j = 1; j < stageArgs.length; j++) {
|
||||
const arg = stageArgs[j]!;
|
||||
if (pipelineStageFunction(lowerer, arg) !== null || isValueStream(lowerer, arg)) continue;
|
||||
if ((lowerer.typeOf(arg).flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown)) !== 0) {
|
||||
candidates.push(j);
|
||||
}
|
||||
}
|
||||
if (candidates.length === 0) return build({ source: src, bound: new Map(), fns: new Set() });
|
||||
const stmts: IrStmt[] = [];
|
||||
const bind = (init: IrExpr): IrExpr => {
|
||||
const local = lowerer.declareHiddenLocal("%pipelineArg", init.type);
|
||||
stmts.push({ kind: "varDecl", localId: local.id, init, loc: init.loc });
|
||||
return varRef(local.id, init.type, init.loc);
|
||||
};
|
||||
const source = src !== null && !src.iterable ? { expr: bind(src.expr), iterable: false } : src;
|
||||
const bound = new Map<number, IrExpr>();
|
||||
let preSource: DynStagePlan["preSource"];
|
||||
for (let j = src !== null ? 1 : 0; j < stageArgs.length; j++) {
|
||||
const arg = stageArgs[j]!;
|
||||
if (pipelineStageFunction(lowerer, arg) !== null) continue;
|
||||
if (j === 0 && iterableSourceType(lowerer, arg) !== null) {
|
||||
const lowered = lowerIterableSource(lowerer, arg, null, loc);
|
||||
if (lowered === null) lowerer.noLowering("pipeline over this source", arg);
|
||||
preSource = { closure: bind(lowered!.closure), out: lowered!.out };
|
||||
continue;
|
||||
}
|
||||
let value = lowerer.lowerExpr(arg);
|
||||
// an untyped array, string, Set, Map or generator object that keeps
|
||||
// its static type is taken through its checked-dynamic value
|
||||
if (
|
||||
candidates.includes(j) &&
|
||||
BOXED_SOURCE_KINDS.has(value.type.kind) &&
|
||||
lowerer.dynConvertible(value.type)
|
||||
) {
|
||||
value = { kind: "dynFrom", value, type: DYN, loc: value.loc };
|
||||
}
|
||||
bound.set(j, bind(value));
|
||||
}
|
||||
const dynStages = candidates
|
||||
.filter((j) => bound.get(j)!.type.kind === "dyn")
|
||||
.slice(0, MAX_DYN_FUNCTION_STAGES);
|
||||
const branch = (i: number, fns: ReadonlySet<number>): IrExpr => {
|
||||
if (i === dynStages.length) return build({ source, bound, preSource, fns });
|
||||
const j = dynStages[i]!;
|
||||
const asStream = branch(i + 1, fns);
|
||||
const asFunction = branch(i + 1, new Set([...fns, j]));
|
||||
// a function destination's promise resolves to its value
|
||||
const type =
|
||||
asFunction.type.kind === "promise" && asFunction.type.inner.kind === "dyn"
|
||||
? asFunction.type
|
||||
: asStream.type;
|
||||
return {
|
||||
kind: "ternary",
|
||||
cond: { kind: "dynTest", test: "function", value: bound.get(j)!, type: BOOL, loc },
|
||||
then: lowerer.coerceToExpected(asFunction, type),
|
||||
else_: lowerer.coerceToExpected(asStream, type),
|
||||
type,
|
||||
loc,
|
||||
};
|
||||
};
|
||||
const result = branch(0, new Set());
|
||||
return { kind: "seqExpr", stmts, result, type: result.type, loc };
|
||||
}
|
||||
|
||||
/** lowerPipelineStages' source arguments for a checked-dynamic first stage. */
|
||||
function stagesSource(
|
||||
src: DynPipelineSource | null,
|
||||
@@ -2938,6 +3038,22 @@ function stagesSource(
|
||||
return src.iterable ? [undefined, src.expr] : [new Map([[0, src.expr]]), undefined];
|
||||
}
|
||||
|
||||
/** A pipeline stage taken as a stream under a DynStagePlan: the
|
||||
* checked-dynamic first stage's stream, a value bound before the
|
||||
* branches, or the argument lowered here. */
|
||||
function planStreamArg(
|
||||
lowerer: Lowerer,
|
||||
plan: DynStagePlan,
|
||||
node: ts.Expression,
|
||||
i: number,
|
||||
): IrExpr {
|
||||
if (i === 0 && plan.source !== null) return plan.source.expr;
|
||||
const bound = plan.bound.get(i);
|
||||
return bound !== undefined
|
||||
? streamArgOf(lowerer, node, bound, "pipeline")
|
||||
: lowerStreamArg(lowerer, node, "pipeline");
|
||||
}
|
||||
|
||||
/** A stream-rooted argument, lowered — or the pointed fence (pipeline's
|
||||
* iterable/generator/function stages have no lowering yet). `what` picks
|
||||
* the validation ladder: finished() rejects EVERY non-stream through
|
||||
@@ -3304,7 +3420,9 @@ export function lowerStreamModuleCall(
|
||||
: null;
|
||||
const stageArgs = opts ? args.slice(0, -1) : args;
|
||||
const build = (src: DynPipelineSource | null): IrExpr =>
|
||||
lowerPromisePipeline(lowerer, stageArgs, opts, src, loc);
|
||||
lowerDynStagePipeline(lowerer, stageArgs, src, loc, (plan) =>
|
||||
lowerPromisePipeline(lowerer, stageArgs, opts, plan, loc),
|
||||
);
|
||||
return lowerDynSourcePipeline(lowerer, stageArgs[0]!, loc, build) ?? build(null);
|
||||
}
|
||||
return null;
|
||||
@@ -3312,17 +3430,23 @@ export function lowerStreamModuleCall(
|
||||
return lowerStreamModuleCallbackCall(lowerer, call, bi, loc);
|
||||
}
|
||||
|
||||
/** stream/promises pipeline(...stages[, options]), with the first stage
|
||||
* pre-lowered when it is checked-dynamic (lowerDynSourcePipeline). */
|
||||
/** stream/promises pipeline(...stages[, options]), for one branch of
|
||||
* lowerDynStagePipeline's plan. */
|
||||
function lowerPromisePipeline(
|
||||
lowerer: Lowerer,
|
||||
stageArgs: readonly ts.Expression[],
|
||||
opts: ReturnType<typeof pipelineOptions> | null,
|
||||
src: DynPipelineSource | null,
|
||||
plan: DynStagePlan,
|
||||
loc: SrcLoc,
|
||||
): IrExpr {
|
||||
if (src?.iterable === true || hasPipelineFunctionStage(lowerer, stageArgs)) {
|
||||
const staged = lowerPipelineStages(lowerer, stageArgs, loc, ...stagesSource(src));
|
||||
const src = plan.source;
|
||||
if (
|
||||
src?.iterable === true ||
|
||||
plan.fns.size > 0 ||
|
||||
plan.preSource !== undefined ||
|
||||
hasPipelineFunctionStage(lowerer, stageArgs)
|
||||
) {
|
||||
const staged = lowerPipelineStages(lowerer, stageArgs, loc, ...stagesSource(src), plan);
|
||||
const type = staged.destPromise ?? { kind: "promise", inner: VOID };
|
||||
// A promise destination's value settles the result (same type).
|
||||
if (!opts) {
|
||||
@@ -3346,9 +3470,7 @@ function lowerPromisePipeline(
|
||||
}
|
||||
let streams: IrExpr[];
|
||||
try {
|
||||
streams = stageArgs.map((a, i) =>
|
||||
i === 0 && src !== null ? src.expr : lowerStreamArg(lowerer, a, "pipeline"),
|
||||
);
|
||||
streams = stageArgs.map((a, i) => planStreamArg(lowerer, plan, a, i));
|
||||
} catch (e) {
|
||||
if (e instanceof StreamArgTypeThrow) return e.expr;
|
||||
throw e;
|
||||
@@ -3530,7 +3652,9 @@ function lowerStreamModuleCallbackCall(
|
||||
);
|
||||
}
|
||||
const build = (src: DynPipelineSource | null): IrExpr =>
|
||||
lowerCallbackPipeline(lowerer, call, args, src, loc);
|
||||
lowerDynStagePipeline(lowerer, args.slice(0, -1), src, loc, (plan) =>
|
||||
lowerCallbackPipeline(lowerer, call, args, plan, loc),
|
||||
);
|
||||
return lowerDynSourcePipeline(lowerer, args[0]!, loc, build) ?? build(null);
|
||||
}
|
||||
|
||||
@@ -3544,23 +3668,32 @@ function lowerStreamModuleCallbackCall(
|
||||
return null;
|
||||
}
|
||||
|
||||
/** pipeline(...stages, callback), with the first stage pre-lowered when it
|
||||
* is checked-dynamic (lowerDynSourcePipeline). */
|
||||
/** pipeline(...stages, callback), for one branch of lowerDynStagePipeline's
|
||||
* plan. */
|
||||
function lowerCallbackPipeline(
|
||||
lowerer: Lowerer,
|
||||
call: ts.CallExpression,
|
||||
args: readonly ts.Expression[],
|
||||
src: DynPipelineSource | null,
|
||||
plan: DynStagePlan,
|
||||
loc: SrcLoc,
|
||||
): IrExpr {
|
||||
if (src?.iterable === true || hasPipelineFunctionStage(lowerer, args.slice(0, -1))) {
|
||||
const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc, ...stagesSource(src));
|
||||
const cbNode = args[args.length - 1]!;
|
||||
const cbParams =
|
||||
ts.isArrowFunction(cbNode) || ts.isFunctionExpression(cbNode)
|
||||
? cbNode.parameters.nodes.length
|
||||
: 0;
|
||||
if (staged.dest === 2 || (staged.dest === 1 && cbParams > 1)) {
|
||||
const src = plan.source;
|
||||
const cbNode = args[args.length - 1]!;
|
||||
const cbParams =
|
||||
ts.isArrowFunction(cbNode) || ts.isFunctionExpression(cbNode)
|
||||
? cbNode.parameters.nodes.length
|
||||
: 0;
|
||||
// A branch over untyped stages whose destination is a stream: the
|
||||
// callback still receives (err, value), value undefined, as in Node.
|
||||
const dynCallback = plan.bound.size > 0 && cbParams > 1;
|
||||
if (
|
||||
src?.iterable === true ||
|
||||
plan.fns.size > 0 ||
|
||||
plan.preSource !== undefined ||
|
||||
hasPipelineFunctionStage(lowerer, args.slice(0, -1))
|
||||
) {
|
||||
const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc, ...stagesSource(src), plan);
|
||||
if (staged.dest >= 2 || (staged.dest === 1 && cbParams > 1) || dynCallback) {
|
||||
// callback(err, value): the destination's value (undefined for
|
||||
// an async generator) reaches the callback through the
|
||||
// checked-dynamic path.
|
||||
@@ -3595,15 +3728,15 @@ function lowerCallbackPipeline(
|
||||
}
|
||||
let streams: IrExpr[];
|
||||
try {
|
||||
streams = args
|
||||
.slice(0, -1)
|
||||
.map((a, i) => (i === 0 && src !== null ? src.expr : lowerStreamArg(lowerer, a, "pipeline")));
|
||||
streams = args.slice(0, -1).map((a, i) => planStreamArg(lowerer, plan, a, i));
|
||||
} catch (e) {
|
||||
if (e instanceof StreamArgTypeThrow) return e.expr;
|
||||
throw e;
|
||||
}
|
||||
const last = streams[streams.length - 1]!;
|
||||
const { cb, dyn } = lowerEosCallback(lowerer, "pipeline", args[args.length - 1]!, last.type);
|
||||
const { cb, dyn } = dynCallback
|
||||
? { cb: lowerer.lowerExprExpecting(cbNode, DYN), dyn: true }
|
||||
: lowerEosCallback(lowerer, "pipeline", cbNode, last.type);
|
||||
return {
|
||||
kind: "libCall",
|
||||
fn: dyn ? "stream.pipelineDyn" : "stream.pipeline",
|
||||
@@ -4213,6 +4346,397 @@ function lowerDynIterableSource(
|
||||
};
|
||||
}
|
||||
|
||||
/** The async generator a checked-dynamic stage's iterable is pumped
|
||||
* through. */
|
||||
function dynStageGenType(): SyncGenType {
|
||||
return { kind: "generator", async: true, yieldT: DYN, retT: VOID, nextT: UNDEFINED_T };
|
||||
}
|
||||
|
||||
/** A checked-dynamic pipeline stage after the first that is a function at
|
||||
* run time, as the stage closure the runtime calls with the previous
|
||||
* stage and { signal } (pipelineImpl's `typeof stream === 'function'`
|
||||
* branch): it calls the function with the previous stream or generator
|
||||
* as an untyped value. Before the last stage the result must be async
|
||||
* iterable (ERR_INVALID_RETURN_VALUE otherwise) and is pumped on through
|
||||
* lowerDynIterableSource's generator; the last stage answers the result,
|
||||
* which the runtime classifies (SCR_PIPE_DEST_DYN). A previous stage
|
||||
* whose type has no untyped form throws the not-supported Error. */
|
||||
function lowerDynFunctionStage(
|
||||
lowerer: Lowerer,
|
||||
fnValue: IrExpr,
|
||||
srcT: IrType,
|
||||
j: number,
|
||||
isLast: boolean,
|
||||
loc: SrcLoc,
|
||||
handoff?: DynStageHandoff,
|
||||
): { closure: IrExpr; out: IrType } {
|
||||
const out: IrType = isLast ? DYN : dynStageGenType();
|
||||
const stageT = funcOf([srcT, DYN], out);
|
||||
const take = handoff?.take === true;
|
||||
const give = handoff?.give === true && !isLast;
|
||||
const key = `dynFunctionStage:${isLast ? "dest" : "transform"}:${typeKey(srcT)}:${take ? 1 : 0}${give ? 1 : 0}`;
|
||||
let maker = lowerer.valueHelpers.get(key);
|
||||
if (!maker) {
|
||||
maker = `%stream.dynFunctionStage.${lowerer.valueHelpers.size}`;
|
||||
lowerer.valueHelpers.set(key, maker);
|
||||
const stageName = `${maker}.stage`;
|
||||
const frame = new LiftedFrame(loc);
|
||||
const fnId = frame.local("fn", DYN, false);
|
||||
const labelId = frame.local("label", STRING, false);
|
||||
const cellId = frame.local("cell", DYN, false);
|
||||
const srcId = frame.param("src", srcT);
|
||||
const optsId = frame.param("opts", DYN);
|
||||
const inputId = frame.local("input", DYN, false);
|
||||
const resultId = frame.local("result", DYN, false);
|
||||
const nextId = frame.local("next", DYN, false);
|
||||
const cell = (): IrExpr => varRef(cellId, DYN, loc);
|
||||
const body: IrStmt[] = [];
|
||||
if (take) {
|
||||
// the previous untyped stage's value, handed over as it is
|
||||
// (makeAsyncIterable answers an iterable itself)
|
||||
body.push(
|
||||
{ kind: "varDecl", localId: inputId, init: handoffTake(cell(), loc), loc },
|
||||
handoffPut(cell(), dynUndefinedExpr(loc), loc),
|
||||
);
|
||||
}
|
||||
if (!take && !lowerer.dynConvertible(srcT)) {
|
||||
body.push({
|
||||
kind: "return",
|
||||
value: nodeThrowExpr(
|
||||
0,
|
||||
"SC2020",
|
||||
`pipeline function stages after a stage of type '${lowerer.fmt(srcT)}' are not supported yet`,
|
||||
out,
|
||||
loc,
|
||||
),
|
||||
loc,
|
||||
});
|
||||
} else {
|
||||
body.push({
|
||||
kind: "varDecl",
|
||||
localId: resultId,
|
||||
init: {
|
||||
kind: "dynCall",
|
||||
callee: varRef(fnId, DYN, loc),
|
||||
receiver: dynUndefinedExpr(loc),
|
||||
calleeName: "transform",
|
||||
args: [
|
||||
take
|
||||
? varRef(inputId, DYN, loc)
|
||||
: { kind: "dynFrom", value: varRef(srcId, srcT, loc), type: DYN, loc },
|
||||
varRef(optsId, DYN, loc),
|
||||
],
|
||||
type: DYN,
|
||||
loc,
|
||||
},
|
||||
loc,
|
||||
});
|
||||
if (isLast) {
|
||||
body.push({ kind: "return", value: varRef(resultId, DYN, loc), loc });
|
||||
} else {
|
||||
body.push({
|
||||
kind: "varDecl",
|
||||
localId: nextId,
|
||||
init: {
|
||||
kind: "libCall",
|
||||
fn: "stream.dynTransformResult",
|
||||
args: [varRef(resultId, DYN, loc), varRef(labelId, STRING, loc)],
|
||||
type: DYN,
|
||||
loc,
|
||||
},
|
||||
loc,
|
||||
});
|
||||
// the next untyped function stage takes the result itself; the
|
||||
// generator returned for the runtime is then never read
|
||||
if (give) body.push(handoffPut(cell(), varRef(nextId, DYN, loc), loc));
|
||||
const iterable = lowerDynIterableSource(lowerer, varRef(nextId, DYN, loc), loc);
|
||||
if (!iterable)
|
||||
throw new InternalCompilerError("dyn iterable generator without a result record");
|
||||
body.push({
|
||||
kind: "return",
|
||||
value: { kind: "callValue", callee: iterable.closure, args: [], type: out, loc },
|
||||
loc,
|
||||
});
|
||||
}
|
||||
}
|
||||
const captured = take || give ? [fnId, labelId, cellId] : [fnId, labelId];
|
||||
for (const id of captured) frame.locals.find((l) => l.id === id)!.boxed = true;
|
||||
lowerer.liftedFns.push({
|
||||
name: stageName,
|
||||
params: frame.params,
|
||||
returnType: out,
|
||||
captures: [
|
||||
{ localId: fnId, name: "fn", type: DYN },
|
||||
{ localId: labelId, name: "label", type: STRING },
|
||||
...(take || give ? [{ localId: cellId, name: "cell", type: DYN }] : []),
|
||||
],
|
||||
locals: frame.locals,
|
||||
body,
|
||||
loc,
|
||||
});
|
||||
const makerFn: IrLocal = { id: "fn.0", name: "fn", type: DYN, mutable: false, boxed: true };
|
||||
const makerLabel: IrLocal = {
|
||||
id: "label.1",
|
||||
name: "label",
|
||||
type: STRING,
|
||||
mutable: false,
|
||||
boxed: true,
|
||||
};
|
||||
const makerCell: IrLocal = {
|
||||
id: "cell.2",
|
||||
name: "cell",
|
||||
type: DYN,
|
||||
mutable: false,
|
||||
boxed: true,
|
||||
};
|
||||
const makerLocals = take || give ? [makerFn, makerLabel, makerCell] : [makerFn, makerLabel];
|
||||
lowerer.liftedFns.push({
|
||||
name: maker,
|
||||
params: makerLocals.map((l) => ({ localId: l.id, name: l.name, type: l.type })),
|
||||
returnType: stageT,
|
||||
locals: makerLocals,
|
||||
body: [
|
||||
{
|
||||
kind: "return",
|
||||
value: {
|
||||
kind: "closure",
|
||||
fnName: stageName,
|
||||
captures: makerLocals.map((l) => l.id),
|
||||
type: stageT,
|
||||
loc,
|
||||
},
|
||||
loc,
|
||||
},
|
||||
],
|
||||
loc,
|
||||
});
|
||||
}
|
||||
return {
|
||||
closure: {
|
||||
kind: "call",
|
||||
callee: maker,
|
||||
args: [fnValue, strLit(`transform[${j - 1}]`, loc), ...(take || give ? [handoff!.cell] : [])],
|
||||
type: stageT,
|
||||
loc,
|
||||
},
|
||||
out,
|
||||
};
|
||||
}
|
||||
|
||||
/** The cell that carries an untyped stage's value to the untyped function
|
||||
* stage after it, created per pipeline call. A stage that gives stores its
|
||||
* value; the next stage takes it instead of the generator the runtime
|
||||
* passes, as Node's makeAsyncIterable answers an iterable itself. */
|
||||
interface DynStageHandoff {
|
||||
cell: IrExpr;
|
||||
take: boolean;
|
||||
give: boolean;
|
||||
}
|
||||
|
||||
function handoffPut(cell: IrExpr, value: IrExpr, loc: SrcLoc): IrStmt {
|
||||
return {
|
||||
kind: "exprStmt",
|
||||
expr: {
|
||||
kind: "libCall",
|
||||
fn: "dyn.keySet",
|
||||
args: [cell, strLit("v", loc), value],
|
||||
type: VOID,
|
||||
loc,
|
||||
},
|
||||
loc,
|
||||
};
|
||||
}
|
||||
|
||||
function handoffTake(cell: IrExpr, loc: SrcLoc): IrExpr {
|
||||
return { kind: "dynKeyGet", value: cell, key: strLit("v", loc), type: DYN, loc };
|
||||
}
|
||||
|
||||
/** `(value) => generator over value`: the async generator an untyped
|
||||
* destination function's async iterable result is pumped through
|
||||
* (SCR_PIPE_DEST_DYN's box), interned. */
|
||||
function dynStageIterHelper(lowerer: Lowerer, loc: SrcLoc): IrExpr {
|
||||
const genT = dynStageGenType();
|
||||
const type = funcOf([DYN], genT);
|
||||
const key = "dynStageIter";
|
||||
let name = lowerer.valueHelpers.get(key);
|
||||
if (!name) {
|
||||
name = `%stream.dynStageIter.${lowerer.valueHelpers.size}`;
|
||||
lowerer.valueHelpers.set(key, name);
|
||||
const v: IrLocal = { id: "value.0", name: "value", type: DYN, mutable: false };
|
||||
const iterable = lowerDynIterableSource(lowerer, varRef(v.id, DYN, loc), loc);
|
||||
if (!iterable)
|
||||
throw new InternalCompilerError("dyn iterable generator without a result record");
|
||||
lowerer.liftedFns.push({
|
||||
name,
|
||||
params: [{ localId: v.id, name: v.name, type: DYN }],
|
||||
returnType: genT,
|
||||
locals: [v],
|
||||
body: [
|
||||
{
|
||||
kind: "return",
|
||||
value: { kind: "callValue", callee: iterable.closure, args: [], type: genT, loc },
|
||||
loc,
|
||||
},
|
||||
],
|
||||
loc,
|
||||
});
|
||||
}
|
||||
return { kind: "closure", fnName: name, captures: [], type, loc };
|
||||
}
|
||||
|
||||
/** A checked-dynamic first pipeline stage pumped as an iterable
|
||||
* (pipelineImpl's i === 0 branch), as the stage closure the runtime calls
|
||||
* with { signal } while it sets the pipeline up: a function is called
|
||||
* with { signal } and its result must be iterable, a { readable, writable }
|
||||
* pair pumps its readable side, and any other value is pumped itself,
|
||||
* through lowerDynIterableSource's generator. */
|
||||
function lowerDynSourceStage(
|
||||
lowerer: Lowerer,
|
||||
value: IrExpr,
|
||||
loc: SrcLoc,
|
||||
cell?: IrExpr,
|
||||
): { closure: IrExpr; out: SyncGenType } | null {
|
||||
const out: SyncGenType = {
|
||||
kind: "generator",
|
||||
async: true,
|
||||
yieldT: DYN,
|
||||
retT: VOID,
|
||||
nextT: UNDEFINED_T,
|
||||
};
|
||||
const stageT = funcOf([DYN], out);
|
||||
const give = cell !== undefined;
|
||||
const key = give ? "dynSourceStage:give" : "dynSourceStage";
|
||||
let maker = lowerer.valueHelpers.get(key);
|
||||
if (!maker) {
|
||||
maker = `%stream.dynSourceStage.${lowerer.valueHelpers.size}`;
|
||||
const stageName = `${maker}.stage`;
|
||||
const frame = new LiftedFrame(loc);
|
||||
const srcId = frame.local("src", DYN, false);
|
||||
const cellId = frame.local("cell", DYN, false);
|
||||
const optsId = frame.param("opts", DYN);
|
||||
const valueId = frame.local("value", DYN, false);
|
||||
const genId = frame.local("gen", out, false);
|
||||
const src = (): IrExpr => varRef(srcId, DYN, loc);
|
||||
const sourceValue = (v: IrExpr, returned: boolean): IrExpr => ({
|
||||
kind: "libCall",
|
||||
fn: "stream.dynSourceValue",
|
||||
args: [v, boolLit(returned, loc)],
|
||||
type: DYN,
|
||||
loc,
|
||||
});
|
||||
const iterable = lowerDynIterableSource(lowerer, varRef(valueId, DYN, loc), loc);
|
||||
if (!iterable) return null;
|
||||
lowerer.valueHelpers.set(key, maker);
|
||||
const body: IrStmt[] = [
|
||||
{
|
||||
kind: "varDecl",
|
||||
localId: valueId,
|
||||
init: {
|
||||
kind: "ternary",
|
||||
cond: { kind: "dynTest", test: "function", value: src(), type: BOOL, loc },
|
||||
then: sourceValue(
|
||||
{
|
||||
kind: "dynCall",
|
||||
callee: src(),
|
||||
receiver: dynUndefinedExpr(loc),
|
||||
calleeName: "source",
|
||||
args: [varRef(optsId, DYN, loc)],
|
||||
type: DYN,
|
||||
loc,
|
||||
},
|
||||
true,
|
||||
),
|
||||
else_: sourceValue(src(), false),
|
||||
type: DYN,
|
||||
loc,
|
||||
},
|
||||
loc,
|
||||
},
|
||||
{
|
||||
kind: "varDecl",
|
||||
localId: genId,
|
||||
init: { kind: "callValue", callee: iterable.closure, args: [], type: out, loc },
|
||||
loc,
|
||||
},
|
||||
];
|
||||
if (give) {
|
||||
// the untyped function stage after the source takes the iterable
|
||||
// itself, as Node's makeAsyncIterable answers it; a stream (a pair's
|
||||
// readable side) goes as the generator over it, as Node iterates a
|
||||
// stream there
|
||||
const v = varRef(valueId, DYN, loc);
|
||||
body.push(
|
||||
handoffPut(
|
||||
varRef(cellId, DYN, loc),
|
||||
{
|
||||
kind: "ternary",
|
||||
cond: { kind: "libCall", fn: "stream.dynIterableSource", args: [v], type: BOOL, loc },
|
||||
then: v,
|
||||
else_: { kind: "dynFrom", value: varRef(genId, out, loc), type: DYN, loc },
|
||||
type: DYN,
|
||||
loc,
|
||||
},
|
||||
loc,
|
||||
),
|
||||
);
|
||||
}
|
||||
body.push({ kind: "return", value: varRef(genId, out, loc), loc });
|
||||
const captured = give ? [srcId, cellId] : [srcId];
|
||||
for (const id of captured) frame.locals.find((l) => l.id === id)!.boxed = true;
|
||||
lowerer.liftedFns.push({
|
||||
name: stageName,
|
||||
params: frame.params,
|
||||
returnType: out,
|
||||
captures: [
|
||||
{ localId: srcId, name: "src", type: DYN },
|
||||
...(give ? [{ localId: cellId, name: "cell", type: DYN }] : []),
|
||||
],
|
||||
locals: frame.locals,
|
||||
body,
|
||||
loc,
|
||||
});
|
||||
const makerSrc: IrLocal = { id: "src.0", name: "src", type: DYN, mutable: false, boxed: true };
|
||||
const makerCell: IrLocal = {
|
||||
id: "cell.1",
|
||||
name: "cell",
|
||||
type: DYN,
|
||||
mutable: false,
|
||||
boxed: true,
|
||||
};
|
||||
const makerLocals = give ? [makerSrc, makerCell] : [makerSrc];
|
||||
lowerer.liftedFns.push({
|
||||
name: maker,
|
||||
params: makerLocals.map((l) => ({ localId: l.id, name: l.name, type: l.type })),
|
||||
returnType: stageT,
|
||||
locals: makerLocals,
|
||||
body: [
|
||||
{
|
||||
kind: "return",
|
||||
value: {
|
||||
kind: "closure",
|
||||
fnName: stageName,
|
||||
captures: makerLocals.map((l) => l.id),
|
||||
type: stageT,
|
||||
loc,
|
||||
},
|
||||
loc,
|
||||
},
|
||||
],
|
||||
loc,
|
||||
});
|
||||
}
|
||||
return {
|
||||
closure: {
|
||||
kind: "call",
|
||||
callee: maker,
|
||||
args: give ? [value, cell!] : [value],
|
||||
type: stageT,
|
||||
loc,
|
||||
},
|
||||
out,
|
||||
};
|
||||
}
|
||||
|
||||
function lowerIterableSource(
|
||||
lowerer: Lowerer,
|
||||
node: ts.Expression,
|
||||
@@ -4356,7 +4880,7 @@ function lowerIterableSource(
|
||||
interface PipelineStagesLowering {
|
||||
args: IrExpr[];
|
||||
last: IrType;
|
||||
dest: 0 | 1 | 2;
|
||||
dest: 0 | 1 | 2 | 3;
|
||||
destPromise: (IrType & { kind: "promise" }) | null;
|
||||
wrap: (call: IrExpr) => IrExpr;
|
||||
}
|
||||
@@ -4377,6 +4901,7 @@ function lowerPipelineStages(
|
||||
loc: SrcLoc,
|
||||
preLowered?: ReadonlyMap<number, IrExpr>,
|
||||
dynSource?: IrExpr,
|
||||
plan?: DynStagePlan,
|
||||
): PipelineStagesLowering {
|
||||
if (stageArgs.length > 52) {
|
||||
lowerer.noLowering(
|
||||
@@ -4390,7 +4915,7 @@ function lowerPipelineStages(
|
||||
const helperArgs: IrExpr[] = [];
|
||||
let positions = 0;
|
||||
let mask = 0;
|
||||
let dest: 0 | 1 | 2 = 0;
|
||||
let dest: 0 | 1 | 2 | 3 = 0;
|
||||
let destPromise: (IrType & { kind: "promise" }) | null = null;
|
||||
// What the next function stage receives: null before the first stage,
|
||||
// else the previous stream's type or the previous stage's generator.
|
||||
@@ -4399,17 +4924,69 @@ function lowerPipelineStages(
|
||||
// before all stages, so when a stage precedes a stream the arguments
|
||||
// are bound to locals first, keeping JavaScript's evaluation order.
|
||||
const argOrder: { list: IrExpr[]; index: number }[] = [];
|
||||
// An untyped stage followed by an untyped function stage hands its value
|
||||
// over through one cell per call (DynStageHandoff).
|
||||
let handoffCell: IrLocal | null = null;
|
||||
const cellRef = (): IrExpr => {
|
||||
handoffCell ??= lowerer.declareHiddenLocal("%pipelineHandoff", DYN);
|
||||
return varRef(handoffCell.id, DYN, loc);
|
||||
};
|
||||
let takes = false;
|
||||
for (let j = 0; j < stageArgs.length; j++) {
|
||||
const arg = stageArgs[j]!;
|
||||
const givesToNext = plan?.fns.has(j + 1) === true;
|
||||
if (j === 0 && dynSource !== undefined) {
|
||||
const source = lowerDynIterableSource(lowerer, dynSource, loc);
|
||||
const source = lowerDynSourceStage(
|
||||
lowerer,
|
||||
dynSource,
|
||||
loc,
|
||||
givesToNext ? cellRef() : undefined,
|
||||
);
|
||||
takes = givesToNext;
|
||||
if (source === null) lowerer.noLowering("pipeline over this source", arg);
|
||||
// a source function takes { signal }
|
||||
mask |= 1 << stages.length;
|
||||
positions += 1;
|
||||
argOrder.push({ list: stages, index: stages.length });
|
||||
stages.push(source!.closure);
|
||||
srcT = source!.out;
|
||||
continue;
|
||||
}
|
||||
if (plan?.fns.has(j) === true) {
|
||||
// a checked-dynamic stage that is a function at run time
|
||||
const isLast = j === stageArgs.length - 1;
|
||||
const take = takes;
|
||||
const give = givesToNext && !isLast;
|
||||
takes = give;
|
||||
const lowered = lowerDynFunctionStage(
|
||||
lowerer,
|
||||
plan.bound.get(j)!,
|
||||
srcT!,
|
||||
j,
|
||||
isLast,
|
||||
loc,
|
||||
take || give ? { cell: cellRef(), take, give } : undefined,
|
||||
);
|
||||
mask |= 1 << stages.length;
|
||||
positions += 2 ** j;
|
||||
argOrder.push({ list: stages, index: stages.length });
|
||||
stages.push(lowered.closure);
|
||||
if (isLast) {
|
||||
// its result is classified at run time: a promise settles the
|
||||
// pipeline's value, and an async iterable is pumped into the
|
||||
// PassThrough
|
||||
const genT = dynStageGenType();
|
||||
const helpers = fromGenHelpers(lowerer, genT, loc);
|
||||
if (!helpers) lowerer.noLowering("a pipeline destination with this generator type", arg);
|
||||
helperArgs.push(helpers![0]!, helpers![1]!, helpers![2]!);
|
||||
dest = 3;
|
||||
destPromise = { kind: "promise", inner: DYN };
|
||||
srcT = null;
|
||||
} else {
|
||||
srcT = lowered.out;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// compose's head and tail functions arrive as Duplex.from streams
|
||||
const fn = preLowered?.has(j) ? null : pipelineStageFunction(lowerer, arg);
|
||||
const syncSourceFn =
|
||||
@@ -4423,7 +5000,9 @@ function lowerPipelineStages(
|
||||
!preLowered?.has(j) &&
|
||||
(syncSourceFn || (fn === null && iterableSourceType(lowerer, arg) !== null))
|
||||
) {
|
||||
const source = lowerIterableSource(lowerer, arg, syncSourceFn ? fn : null, loc);
|
||||
const source =
|
||||
(syncSourceFn ? undefined : plan?.preSource) ??
|
||||
lowerIterableSource(lowerer, arg, syncSourceFn ? fn : null, loc);
|
||||
if (source === null) lowerer.noLowering("pipeline over this source", arg);
|
||||
positions += 1;
|
||||
argOrder.push({ list: stages, index: stages.length });
|
||||
@@ -4448,7 +5027,12 @@ function lowerPipelineStages(
|
||||
}
|
||||
let stream: IrExpr;
|
||||
try {
|
||||
stream = preLowered?.get(j) ?? lowerStreamArg(lowerer, arg, "pipeline");
|
||||
const bound = plan?.bound.get(j);
|
||||
stream =
|
||||
preLowered?.get(j) ??
|
||||
(bound !== undefined
|
||||
? streamArgOf(lowerer, arg, bound, "pipeline")
|
||||
: lowerStreamArg(lowerer, arg, "pipeline"));
|
||||
} catch (e) {
|
||||
if (e instanceof StreamArgTypeThrow) lowerer.noLowering("pipeline over this value", arg);
|
||||
throw e;
|
||||
@@ -4527,6 +5111,8 @@ function lowerPipelineStages(
|
||||
);
|
||||
}
|
||||
helperArgs.push(pipelineBoxHelper(lowerer, destPromise!, loc));
|
||||
} else if (dest === 3) {
|
||||
helperArgs.push(dynStageIterHelper(lowerer, loc));
|
||||
}
|
||||
const argDecls: IrStmt[] = [];
|
||||
const firstStage = argOrder.findIndex((a) => a.list === stages);
|
||||
@@ -4538,6 +5124,15 @@ function lowerPipelineStages(
|
||||
a.list[a.index] = varRef(local.id, init.type, init.loc);
|
||||
}
|
||||
}
|
||||
if (handoffCell !== null) {
|
||||
const cell: IrLocal = handoffCell;
|
||||
argDecls.unshift({
|
||||
kind: "varDecl",
|
||||
localId: cell.id,
|
||||
init: { kind: "dynObjLit", fields: [], type: DYN, loc },
|
||||
loc,
|
||||
});
|
||||
}
|
||||
const bindArgs = (call: IrExpr): IrExpr =>
|
||||
argDecls.length === 0
|
||||
? call
|
||||
|
||||
@@ -92,6 +92,9 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
|
||||
"child.stdioStream",
|
||||
"stream.unboxDyn",
|
||||
"stream.dynIterableSource",
|
||||
"stream.dynSourceValue",
|
||||
"stream.unboxDynStage",
|
||||
"stream.dynTransformResult",
|
||||
"stream.handleStream",
|
||||
"stream.pipeDyn",
|
||||
"stream.unpipeDyn",
|
||||
|
||||
@@ -2870,9 +2870,27 @@ function buildLibFnSignatures() {
|
||||
result: { kind: "object", className: "%Duplex" },
|
||||
},
|
||||
/** Whether pipeline() pumps a checked-dynamic first stage as an
|
||||
* iterable (a string, or a value with an iterator method that is not a
|
||||
* stream with pipe() and on()) instead of piping it as a stream. */
|
||||
* iterable (a string, a value with an iterator method that is not a
|
||||
* stream with pipe() and on(), a { readable, writable } pair, or a
|
||||
* source function's result) instead of piping it as a stream. */
|
||||
"stream.dynIterableSource": { argTypes: [DYN], result: BOOL },
|
||||
/** The value pipeline() pumps for a checked-dynamic first stage: a
|
||||
* source function's result (the flag set), which must be iterable, else
|
||||
* a { readable, writable } pair's readable side or the value itself. */
|
||||
"stream.dynSourceValue": { argTypes: [DYN, BOOL], result: DYN },
|
||||
/** An untyped pipeline function stage's result before the last stage,
|
||||
* which must be async iterable: ERR_INVALID_RETURN_VALUE naming the
|
||||
* stage ("transform[i]") otherwise. Answers the value (+1). */
|
||||
"stream.dynTransformResult": { argTypes: [DYN, STRING], result: DYN },
|
||||
/** A checked-dynamic pipeline() stage taken as a stream: a native stream
|
||||
* (+1), or Duplex.from of a value that is not stream-like or a function
|
||||
* (an iterable), which the pipeline uses without piping into it. Other
|
||||
* values throw: Duplex.from's ERR_INVALID_ARG_TYPE, or the given
|
||||
* not-supported message. */
|
||||
"stream.unboxDynStage": {
|
||||
argTypes: [DYN, STRING],
|
||||
result: { kind: "object", className: "%Duplex" },
|
||||
},
|
||||
/** The Transform a Cipheriv or Decipheriv reads and writes as a stream
|
||||
* (+1, created on first use and owned by the cipher). */
|
||||
"stream.handleStream": {
|
||||
@@ -5237,10 +5255,12 @@ export type IrLibFn = keyof ReturnType<typeof buildLibFnSignatures>;
|
||||
/** The function-stage runs of a pipeline of `total` stages that a stream
|
||||
* follows (positions bit j: stage j is a function), and an async
|
||||
* generator destination's run (flags bits 1-2 = 1, pumped into the
|
||||
* PassThrough pipeline() answers): each pumps its last generator through
|
||||
* one (next, read, return) triple. */
|
||||
* PassThrough pipeline() answers, or 3, an untyped destination whose
|
||||
* result may be pumped there): each pumps its last generator through one
|
||||
* (next, read, return) triple. */
|
||||
export function pipelinePumpedRuns(total: number, positions: number, flags: number): number {
|
||||
let runs = ((flags >> 1) & 3) === 1 ? 1 : 0;
|
||||
const dest = (flags >> 1) & 3;
|
||||
let runs = dest === 1 || dest === 3 ? 1 : 0;
|
||||
for (let j = 0; j + 1 < total; j++) {
|
||||
if (Math.floor(positions / 2 ** j) % 2 === 1 && Math.floor(positions / 2 ** (j + 1)) % 2 === 0)
|
||||
runs++;
|
||||
|
||||
@@ -278,6 +278,9 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
|
||||
"stream.stdioWritable",
|
||||
"stream.unboxDyn",
|
||||
"stream.dynIterableSource",
|
||||
"stream.dynSourceValue",
|
||||
"stream.dynTransformResult",
|
||||
"stream.unboxDynStage",
|
||||
"stream.handleStream",
|
||||
"stream.unpipeDyn",
|
||||
"stream.workerStdio",
|
||||
@@ -314,6 +317,7 @@ const READABLE_DYN_ASYNC_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
|
||||
"stream.duplexFromAsyncFn",
|
||||
"stream.duplexFromGenSignal",
|
||||
"stream.duplexFromAsyncFnSignal",
|
||||
"stream.unboxDynStage",
|
||||
]);
|
||||
|
||||
/** libCall name prefixes of the Readable entry points that await through the dyn-async unit. */
|
||||
|
||||
@@ -5761,7 +5761,7 @@ function validateFunction(
|
||||
}
|
||||
const lastIsFunction = Math.floor(positions! / 2 ** (total - 1)) % 2 === 1;
|
||||
const helperCount = 3 * pipelinePumpedRuns(total, positions!, flags!);
|
||||
const groups = helperCount + (destKind === 2 ? 1 : 0) + (flags! & 32 ? 2 : 0);
|
||||
const groups = helperCount + (destKind >= 2 ? 1 : 0) + (flags! & 32 ? 2 : 0);
|
||||
const cbArgs =
|
||||
e.fn === "sp.pipelineStagesOpts"
|
||||
? 1 + (flags! & 64 ? 1 : 0)
|
||||
@@ -5776,7 +5776,7 @@ function validateFunction(
|
||||
total > 52 ||
|
||||
positions! >= 2 ** total ||
|
||||
functions !== k ||
|
||||
destKind === 3 ||
|
||||
(destKind === 3 && e.fn === "stream.composeStages") ||
|
||||
(destKind !== 0) !== lastIsFunction ||
|
||||
total < 2 ||
|
||||
e.args.length !== 5 + n! + k! + groups + cbArgs
|
||||
@@ -5795,7 +5795,7 @@ function validateFunction(
|
||||
err(`libCall ${e.fn} stage function ${i} must be a func`, e.loc);
|
||||
}
|
||||
at2 += k!;
|
||||
const helpers = helperCount + (destKind === 2 ? 1 : 0);
|
||||
const helpers = helperCount + (destKind >= 2 ? 1 : 0);
|
||||
for (let i = at2; i < at2 + helpers; i++) {
|
||||
if (e.args[i]?.type.kind !== "func")
|
||||
err(`libCall ${e.fn} helper ${i} must be a func`, e.loc);
|
||||
|
||||
@@ -378,7 +378,7 @@ done:
|
||||
expect(await symbols(helperObject, ["-u"])).toEqual(await symbols(clangObject, ["-u"]));
|
||||
expect(await symbols(helperObject, ["-gU"])).toEqual(await symbols(clangObject, ["-gU"]));
|
||||
expect(await symbols(helperObject, ["-gU"])).toEqual(["0000000000000000 T _main"]);
|
||||
expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v45");
|
||||
expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v46");
|
||||
|
||||
const clangExe = join(dir, "clang-program");
|
||||
const linkDriver = join(dir, "clang-driver.c");
|
||||
@@ -712,7 +712,7 @@ done:
|
||||
.split("\n");
|
||||
expect(undefinedSymbols).toContain("_sf_scale");
|
||||
expect(undefinedSymbols).toContain("_sf_callback_mix");
|
||||
expect(undefinedSymbols).toContain("_scr_runtime_abi_v45");
|
||||
expect(undefinedSymbols).toContain("_scr_runtime_abi_v46");
|
||||
});
|
||||
|
||||
test("helper and clang object paths retain the same outbound FFI C ABI", async () => {
|
||||
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7508-pipeline-untyped-source-functions.mjs"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7509-pipeline-untyped-function-stages.mjs"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7510-pipeline-untyped-later-iterables.mjs"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7511-pipeline-untyped-stages.cjs"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7512-pipeline-any-stages.ts"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"order": [
|
||||
"<repo>/tests/corpus/7513-pipeline-untyped-stage-errors.mjs"
|
||||
],
|
||||
"diags": []
|
||||
}
|
||||
@@ -366,7 +366,7 @@ async function build() {
|
||||
package: packageManifest.name,
|
||||
version: packageManifest.version,
|
||||
target: RUNTIME_PACK_MATRIX.target,
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" },
|
||||
compiler: {
|
||||
command: compiler,
|
||||
identity: compilerVersion,
|
||||
|
||||
@@ -469,7 +469,7 @@ async function build() {
|
||||
package: packageManifest.name,
|
||||
version: packageManifest.version,
|
||||
target: matrix.target,
|
||||
runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" },
|
||||
runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" },
|
||||
compiler: {
|
||||
command: compiler,
|
||||
identity: compilerVersion,
|
||||
|
||||
@@ -148,7 +148,7 @@ static void scr_enable_utf8_console_output(void) {
|
||||
}
|
||||
#endif
|
||||
|
||||
void scr_runtime_abi_v45(void) {}
|
||||
void scr_runtime_abi_v46(void) {}
|
||||
|
||||
void scr_init(void) {
|
||||
#ifdef SCR_WORKERS
|
||||
|
||||
@@ -81,7 +81,7 @@ void scr_init(void);
|
||||
/* Program objects emitted by the bundled LLVM helper reference this symbol.
|
||||
* Its versioned spelling makes a mismatched manual runtime link fail before
|
||||
* the program can start. */
|
||||
void scr_runtime_abi_v45(void);
|
||||
void scr_runtime_abi_v46(void);
|
||||
void scr_runtime_workers_v8(void);
|
||||
|
||||
/* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ──────
|
||||
|
||||
@@ -1548,6 +1548,9 @@ ScrDyn *scr_dyn_builtin_stream_object(struct ScrStream *s);
|
||||
ScrDyn *scr_dyn_builtin_stream_view(struct ScrStream *s);
|
||||
struct ScrStream *scr_dyn_builtin_stream_check(const ScrDyn *value, double kind);
|
||||
struct ScrStream *scr_dyn_builtin_stream_unbox(const ScrDyn *value, const ScrStr *message);
|
||||
/* stream.unboxDynStage in those programs (scr_stream_unbox_dyn_stage's
|
||||
* contract, also taking constructor-function streams). */
|
||||
struct ScrStream *scr_dyn_builtin_stream_unbox_stage(const ScrDyn *value, const ScrStr *message);
|
||||
/* stream.dynNative in those programs: scr_stream_dyn_native's contract,
|
||||
* also taking constructor-function streams that keep the runtime's write
|
||||
* and end (+1 or NULL). */
|
||||
|
||||
@@ -1752,3 +1752,11 @@ ScrStream *scr_dyn_builtin_stream_unbox(const ScrDyn *value, const ScrStr *messa
|
||||
}
|
||||
return scr_stream_unbox_dyn(value, message);
|
||||
}
|
||||
|
||||
/* stream.unboxDynStage in programs that link this unit: a
|
||||
* constructor-function stream as scr_dyn_builtin_stream_unbox takes it,
|
||||
* any other value as scr_stream_unbox_dyn_stage does. */
|
||||
ScrStream *scr_dyn_builtin_stream_unbox_stage(const ScrDyn *value, const ScrStr *message) {
|
||||
if (bcs_of(value) && value->kind == SCR_DYN_OBJ) return scr_dyn_builtin_stream_unbox(value, message);
|
||||
return scr_stream_unbox_dyn_stage(value, message);
|
||||
}
|
||||
|
||||
@@ -93,6 +93,10 @@ struct ScrStreamState {
|
||||
bool is_transform; /* the writable side routes through transform/flush */
|
||||
bool allow_half_open;
|
||||
bool auto_destroy, emit_close;
|
||||
/* Duplex.from of a pipeline() argument that is not a stream: the
|
||||
* pipeline takes it as that stage's stream without wiring the stage
|
||||
* (scr_stream_unbox_dyn_stage) */
|
||||
bool pipeline_body;
|
||||
|
||||
/* shared lifecycle */
|
||||
bool destroyed, destroy_calling, error_emitted, close_emitted, error_scheduled, close_scheduled;
|
||||
@@ -3850,7 +3854,9 @@ typedef ScrPromise *(*ScrPipeNextFn)(ScrClosure *, ScrGen *);
|
||||
typedef ScrDyn *(*ScrPipeReadFn)(ScrClosure *, void *);
|
||||
typedef ScrDyn *(*ScrPipeBoxFn)(ScrClosure *, ScrPromise *);
|
||||
|
||||
enum { SCR_PIPE_DEST_STREAM, SCR_PIPE_DEST_GEN, SCR_PIPE_DEST_PROMISE };
|
||||
/* SCR_PIPE_DEST_DYN: an untyped destination function's result, a
|
||||
* checked-dynamic value the pipeline classifies (scr_pipe_dest_dyn_kind) */
|
||||
enum { SCR_PIPE_DEST_STREAM, SCR_PIPE_DEST_GEN, SCR_PIPE_DEST_PROMISE, SCR_PIPE_DEST_DYN };
|
||||
/* flags bit 7: the callback is typed code's (stream.pipelineStagesDyn) */
|
||||
enum { SCR_PIPE_TYPED_CB = 128 };
|
||||
|
||||
@@ -3868,7 +3874,9 @@ typedef struct {
|
||||
* async generator destination's PassThrough included), in pipeline
|
||||
* order, or NULL */
|
||||
ScrClosure **helpers;
|
||||
ScrClosure *box; /* SCR_PIPE_DEST_PROMISE: boxes its promise as a dyn promise */
|
||||
/* SCR_PIPE_DEST_PROMISE: boxes its promise as a dyn promise;
|
||||
* SCR_PIPE_DEST_DYN: answers the async generator over a dyn iterable */
|
||||
ScrClosure *box;
|
||||
} ScrPipeStages;
|
||||
|
||||
static bool scr_pipe_is_stage(const ScrPipeStages *fs, size_t j) {
|
||||
@@ -4234,6 +4242,71 @@ static void scr_pipe_dest_promise(ScrStreamPipeCtx *ctx, ScrPromise *p /*moves*/
|
||||
scr_promise_release(scr_async_spawn_after(d->boxed->v.promise, &scr_pipe_dest_entry, d));
|
||||
}
|
||||
|
||||
/* An untyped destination's promise: scr_pipe_dest_promise over a promise
|
||||
* that already holds dyn values. */
|
||||
static void scr_pipe_dest_dyn_promise(ScrStreamPipeCtx *ctx, ScrDyn *boxed /*moves*/, size_t pi,
|
||||
bool end) {
|
||||
ScrPipeDest *d = malloc(sizeof *d);
|
||||
if (!d) scr_stream_oom();
|
||||
d->ctx = scr_stream_pipectx_retain_v(ctx);
|
||||
d->typed = scr_promise_retain(boxed->v.promise);
|
||||
d->boxed = boxed;
|
||||
d->pi = pi;
|
||||
d->end = end;
|
||||
scr_promise_release(scr_async_spawn_after(d->boxed->v.promise, &scr_pipe_dest_entry, d));
|
||||
}
|
||||
|
||||
typedef ScrGen *(*ScrPipeIterFn)(ScrClosure *, ScrDyn *);
|
||||
static bool scr_stream_dyn_has_method(const ScrDyn *value, const char *key);
|
||||
static bool scr_stream_dyn_transform_like(const ScrDyn *value);
|
||||
static void scr_stream_throw_web_stage(void);
|
||||
static bool scr_stream_dyn_stage_returned_stream(const ScrDyn *value);
|
||||
static void scr_stream_throw_invalid_return(const char *expected, const char *name,
|
||||
const ScrDyn *got);
|
||||
|
||||
/* What pipelineImpl does with an untyped destination function's result:
|
||||
* a promise settles the pipeline's value (the `then` branch), and an
|
||||
* async iterable, or a { readable, writable } pair's readable side, is
|
||||
* pumped into the PassThrough. *out answers the promise or that
|
||||
* iterable's async generator (+1). Other values throw
|
||||
* ERR_INVALID_RETURN_VALUE, and thenables that are not promises and web
|
||||
* streams throw the not-supported Error; -1 then. */
|
||||
enum { SCR_PIPE_DYN_PROMISE, SCR_PIPE_DYN_ITER };
|
||||
static int scr_pipe_dest_dyn_kind(const ScrPipeStages *fs, ScrDyn *r, void **out) {
|
||||
if (r->kind == SCR_DYN_PROMISE) {
|
||||
*out = scr_dyn_retain(r);
|
||||
return SCR_PIPE_DYN_PROMISE;
|
||||
}
|
||||
if (r->kind == SCR_DYN_OBJ && scr_stream_dyn_has_method(r, "then")) {
|
||||
static const char msg[] =
|
||||
"pipeline destinations that return a thenable other than a promise are not supported yet";
|
||||
scr_throw_error_msg_code(SCR_ERR_ERROR, msg, sizeof msg - 1, "SC2020");
|
||||
return -1;
|
||||
}
|
||||
if (scr_exc_pending() || scr_stream_dyn_stage_returned_stream(r)) return -1;
|
||||
ScrDyn *read = NULL;
|
||||
int kind = scr_dyn_iterable_kind(r);
|
||||
if (kind < 0) return -1;
|
||||
if (r->kind == SCR_DYN_HANDLE &&
|
||||
(r->v.handle.tag == SCR_DYNH_WEB_STREAM || r->v.handle.tag == SCR_DYNH_WEB_TRANSFORM)) {
|
||||
scr_stream_throw_web_stage();
|
||||
return -1;
|
||||
}
|
||||
if (kind == 2) {
|
||||
read = scr_dyn_retain(r);
|
||||
} else if (scr_stream_dyn_transform_like(r)) {
|
||||
read = scr_dyn_obj_read(r, "readable", 8);
|
||||
}
|
||||
if (!read) {
|
||||
if (!scr_exc_pending()) scr_stream_throw_invalid_return("AsyncIterable or Promise", "destination", r);
|
||||
return -1;
|
||||
}
|
||||
ScrGen *g = ((ScrPipeIterFn)fs->box->fn)(fs->box, read);
|
||||
if (!g) return -1;
|
||||
*out = g;
|
||||
return SCR_PIPE_DYN_ITER;
|
||||
}
|
||||
|
||||
static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosure *cb,
|
||||
ScrStreamErrInv inv, bool end_opt, ScrDyn *signal,
|
||||
ScrAbortAddFn add, const ScrPipeStages *fs,
|
||||
@@ -4294,8 +4367,18 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu
|
||||
: (void *)scr_stream_retain(ret);
|
||||
ret_gen = NULL;
|
||||
void *out = scr_pipe_call_stage(fs, k++, src);
|
||||
int dyn_kind = -1;
|
||||
if (!scr_exc_pending() && dest && fs->dest == SCR_PIPE_DEST_DYN) {
|
||||
ScrDyn *r = out ? out : scr_dyn_retain(scr_dyn_undefined());
|
||||
out = NULL;
|
||||
dyn_kind = scr_pipe_dest_dyn_kind(fs, r, &out);
|
||||
scr_dyn_release(r);
|
||||
}
|
||||
if (scr_exc_pending()) {
|
||||
if (out && dest) scr_promise_release(out);
|
||||
if (out && dest && fs->dest == SCR_PIPE_DEST_DYN) {
|
||||
if (dyn_kind == SCR_PIPE_DYN_ITER) scr_gen_release(out);
|
||||
else scr_dyn_release(out);
|
||||
} else if (out && dest) scr_promise_release(out);
|
||||
else if (out) scr_gen_release(out);
|
||||
scr_stream_pipectx_drop(ctx);
|
||||
scr_stream_pipectx_release_v(ctx);
|
||||
@@ -4315,9 +4398,12 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu
|
||||
size_t pi = ctx->n++;
|
||||
ctx->streams[pi] = pt;
|
||||
ctx->finish_count++;
|
||||
if (fs->dest == SCR_PIPE_DEST_GEN) {
|
||||
if (fs->dest == SCR_PIPE_DEST_GEN ||
|
||||
(fs->dest == SCR_PIPE_DEST_DYN && dyn_kind == SCR_PIPE_DYN_ITER)) {
|
||||
/* pumpToNode(ret, pt, finish, { end }) */
|
||||
scr_pump_start(ctx, out, fs->helpers + 3 * h++, pt, end);
|
||||
} else if (fs->dest == SCR_PIPE_DEST_DYN) {
|
||||
scr_pipe_dest_dyn_promise(ctx, out, pi, end);
|
||||
} else {
|
||||
scr_pipe_dest_promise(ctx, out, fs->box, pi, end);
|
||||
}
|
||||
@@ -4332,6 +4418,20 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu
|
||||
continue;
|
||||
}
|
||||
ScrStream *s = streams[i];
|
||||
if (s->st->pipeline_body) {
|
||||
/* ret = Duplex.from(stream) for an argument that is not a stream:
|
||||
* no destroyer or 'error' listener, and the previous stage is
|
||||
* neither piped nor pumped into it */
|
||||
s->st->pipeline_body = false;
|
||||
if (ret_gen) {
|
||||
scr_gen_release(ret_gen);
|
||||
ret_gen = NULL;
|
||||
h++;
|
||||
}
|
||||
ret = s;
|
||||
i++;
|
||||
continue;
|
||||
}
|
||||
bool writing = j > 0;
|
||||
if (reading && !scr_pipe_is_stage(fs, j + 1) &&
|
||||
(streams[i + 1]->st->close_emitted || streams[i + 1]->st->destroyed)) {
|
||||
@@ -8593,10 +8693,90 @@ ScrStream *scr_stream_unbox_dyn(const ScrDyn *value, const ScrStr *message) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* pipeline.js's first stage: a value with pipe() and on() methods is a
|
||||
* readable node stream, which the stage ladder tests before isIterable;
|
||||
* a string, or a value with a Symbol.asyncIterator or Symbol.iterator
|
||||
* method, is an iterable the pipeline pumps into the next stage. */
|
||||
/* Whether a checked-dynamic object's member is of type "object". */
|
||||
static bool scr_stream_dyn_member_object(const ScrDyn *value, const char *key) {
|
||||
ScrDyn *m = scr_dyn_obj_read(value, key, strlen(key));
|
||||
if (!m) return false;
|
||||
ScrStr *t = scr_dyn_typeof(m);
|
||||
bool object = t->len == 6 && memcmp(t->data, "object", 6) == 0;
|
||||
scr_str_release(t);
|
||||
scr_dyn_release(m);
|
||||
return object;
|
||||
}
|
||||
|
||||
/* pipeline.js's isTransformStream for a value that is not a stream: an
|
||||
* object whose readable and writable members are both of type "object". */
|
||||
static bool scr_stream_dyn_transform_like(const ScrDyn *value) {
|
||||
if (value->kind == SCR_DYN_HANDLE) return value->v.handle.tag == SCR_DYNH_WEB_TRANSFORM;
|
||||
return value->kind == SCR_DYN_OBJ && scr_stream_dyn_member_object(value, "readable") &&
|
||||
!scr_exc_pending() && scr_stream_dyn_member_object(value, "writable");
|
||||
}
|
||||
|
||||
/* Whether a checked-dynamic object has a function member `key`. */
|
||||
static bool scr_stream_dyn_has_method(const ScrDyn *value, const char *key) {
|
||||
ScrDyn *m = scr_dyn_obj_read(value, key, strlen(key));
|
||||
if (!m) return false;
|
||||
bool fn = m->kind == SCR_DYN_FUNC;
|
||||
scr_dyn_release(m);
|
||||
return fn;
|
||||
}
|
||||
|
||||
/* pipeline.js's isNodeStream and isWebStream for a checked-dynamic value
|
||||
* that is not a native stream: a stream-like object or a web stream, which
|
||||
* the pipeline would pipe or pump into and has no lowering for. */
|
||||
static bool scr_stream_dyn_stream_like(const ScrDyn *value) {
|
||||
if (value->kind == SCR_DYN_HANDLE) {
|
||||
int tag = value->v.handle.tag;
|
||||
return tag == SCR_DYNH_WEB_STREAM || tag == SCR_DYNH_WEB_TRANSFORM || tag == SCR_DYNH_WEB_WRITABLE;
|
||||
}
|
||||
if (value->kind != SCR_DYN_OBJ) return false;
|
||||
static const char *const state_keys[] = {"_readableState", "_writableState"};
|
||||
for (size_t i = 0; i < 2; i++) {
|
||||
ScrDyn *v = scr_dyn_obj_read(value, state_keys[i], strlen(state_keys[i]));
|
||||
bool set = v && scr_dyn_truthy(v);
|
||||
if (v) scr_dyn_release(v);
|
||||
if (set) return true;
|
||||
}
|
||||
bool on = scr_stream_dyn_has_method(value, "on");
|
||||
if (on && (scr_stream_dyn_has_method(value, "write") || scr_stream_dyn_has_method(value, "pipe")))
|
||||
return true;
|
||||
if (scr_stream_dyn_has_method(value, "pipeThrough") &&
|
||||
scr_stream_dyn_has_method(value, "getReader") && scr_stream_dyn_has_method(value, "cancel"))
|
||||
return true;
|
||||
if (scr_stream_dyn_has_method(value, "getWriter") && scr_stream_dyn_has_method(value, "abort"))
|
||||
return true;
|
||||
return scr_stream_dyn_transform_like(value);
|
||||
}
|
||||
|
||||
/* A checked-dynamic pipeline() stage that the pipeline takes as a stream
|
||||
* (the stream.unboxDynStage libCall): a boxed native stream or a stream
|
||||
* handle answers its stream (+1). Any other value that is not stream-like
|
||||
* or a function takes pipelineImpl's last branch, Duplex.from(stream):
|
||||
* an iterable answers a readable Duplex the pipeline uses without piping
|
||||
* the previous stage into it, and values Duplex.from rejects throw its
|
||||
* ERR_INVALID_ARG_TYPE. Stream-like objects, web streams and functions
|
||||
* throw `message` as the not-supported Error. */
|
||||
ScrStream *scr_stream_unbox_dyn_stage(const ScrDyn *value, const ScrStr *message) {
|
||||
ScrStream *stream = scr_stream_of_handle(value);
|
||||
if (!stream) stream = scr_stream_boxed(value);
|
||||
if (stream) return stream;
|
||||
if (value->kind == SCR_DYN_FUNC || scr_stream_dyn_stream_like(value)) {
|
||||
if (!scr_exc_pending())
|
||||
scr_throw_error_msg_code(SCR_ERR_ERROR, message->data, message->len, "SC2020");
|
||||
return NULL;
|
||||
}
|
||||
if (scr_exc_pending()) return NULL;
|
||||
ScrStream *d = scr_stream_duplex_from_dyn((ScrDyn *)value);
|
||||
if (d) d->st->pipeline_body = true;
|
||||
return d;
|
||||
}
|
||||
|
||||
/* pipeline.js's first stage: a function is a source function, which the
|
||||
* pipeline calls; a value with pipe() and on() methods is a readable node
|
||||
* stream, which the stage ladder tests before isIterable; a string, a
|
||||
* value with a Symbol.asyncIterator or Symbol.iterator method, or a
|
||||
* { readable, writable } pair is an iterable the pipeline pumps into the
|
||||
* next stage. */
|
||||
bool scr_stream_dyn_iterable_source(const ScrDyn *value) {
|
||||
ScrStream *stream = scr_stream_of_handle(value);
|
||||
if (!stream) stream = scr_stream_boxed(value);
|
||||
@@ -8604,7 +8784,8 @@ bool scr_stream_dyn_iterable_source(const ScrDyn *value) {
|
||||
scr_stream_release(stream);
|
||||
return false;
|
||||
}
|
||||
if (value->kind == SCR_DYN_STR) return true;
|
||||
if (value->kind == SCR_DYN_STR || value->kind == SCR_DYN_FUNC) return true;
|
||||
if (value->kind == SCR_DYN_HANDLE && value->v.handle.tag == SCR_DYNH_WEB_TRANSFORM) return true;
|
||||
if (value->kind == SCR_DYN_OBJ) {
|
||||
ScrDyn *pipe = scr_dyn_obj_read(value, "pipe", 4);
|
||||
if (!pipe) return false;
|
||||
@@ -8617,6 +8798,79 @@ bool scr_stream_dyn_iterable_source(const ScrDyn *value) {
|
||||
scr_dyn_release(on);
|
||||
if (streamish) return false;
|
||||
}
|
||||
if (scr_stream_dyn_transform_like(value)) return true;
|
||||
if (scr_exc_pending()) return false;
|
||||
}
|
||||
return scr_dyn_iterable_kind(value) > 0;
|
||||
}
|
||||
|
||||
/* Untyped web streams as pipeline stages have no lowering. */
|
||||
static void scr_stream_throw_web_stage(void) {
|
||||
static const char msg[] = "pipeline stages that are untyped web streams are not supported yet";
|
||||
scr_throw_error_msg_code(SCR_ERR_ERROR, msg, sizeof msg - 1, "SC2020");
|
||||
}
|
||||
|
||||
/* A native stream returned by an untyped pipeline function, which Node
|
||||
* pipes or passes on as the stream: throws the not-supported Error. */
|
||||
static bool scr_stream_dyn_stage_returned_stream(const ScrDyn *value) {
|
||||
ScrStream *stream = scr_stream_of_handle(value);
|
||||
if (!stream) stream = scr_stream_boxed(value);
|
||||
if (!stream) return false;
|
||||
scr_stream_release(stream);
|
||||
static const char msg[] = "pipeline function stages that return a stream are not supported yet";
|
||||
scr_throw_error_msg_code(SCR_ERR_ERROR, msg, sizeof msg - 1, "SC2020");
|
||||
return true;
|
||||
}
|
||||
|
||||
/* Node's ERR_INVALID_RETURN_VALUE TypeError: "Expected <expected> to be
|
||||
* returned from the "<name>" function but got <the value's type>." */
|
||||
static void scr_stream_throw_invalid_return(const char *expected, const char *name,
|
||||
const ScrDyn *got) {
|
||||
char detail[160];
|
||||
const char *type = scr_dyn_specific_type(got, detail, sizeof detail);
|
||||
char msg[384];
|
||||
int n = snprintf(msg, sizeof msg, "Expected %s to be returned from the \"%s\" function but got %s.",
|
||||
expected, name, type);
|
||||
if (n < 0) n = 0;
|
||||
if ((size_t)n >= sizeof msg) n = (int)sizeof msg - 1;
|
||||
scr_throw_error_msg_code(SCR_ERR_TYPE, msg, (size_t)n, "ERR_INVALID_RETURN_VALUE");
|
||||
}
|
||||
|
||||
/* An untyped pipeline function stage's result before the last stage
|
||||
* (pipelineImpl's `isIterable(ret, true)` check): an async iterable
|
||||
* answers itself (+1); other values throw ERR_INVALID_RETURN_VALUE naming
|
||||
* the stage, as `label` ("transform[i]") does. */
|
||||
ScrDyn *scr_stream_dyn_transform_result(const ScrDyn *value, const ScrStr *label) {
|
||||
if (scr_stream_dyn_stage_returned_stream(value)) return NULL;
|
||||
int kind = scr_dyn_iterable_kind(value);
|
||||
if (kind < 0) return NULL;
|
||||
if (kind != 2) {
|
||||
scr_stream_throw_invalid_return("AsyncIterable", label->data, value);
|
||||
return NULL;
|
||||
}
|
||||
return scr_dyn_retain((ScrDyn *)value);
|
||||
}
|
||||
|
||||
/* The value a checked-dynamic first stage pumps (pipelineImpl's i === 0
|
||||
* branch): for `returned`, a source function's result, which must be
|
||||
* iterable; otherwise a { readable, writable } pair's readable side, or
|
||||
* the value itself. +1, or NULL with an exception pending. */
|
||||
ScrDyn *scr_stream_dyn_source_value(const ScrDyn *value, bool returned) {
|
||||
if (!returned) {
|
||||
if (value->kind == SCR_DYN_HANDLE && value->v.handle.tag == SCR_DYNH_WEB_TRANSFORM) {
|
||||
scr_stream_throw_web_stage();
|
||||
return NULL;
|
||||
}
|
||||
if (scr_stream_dyn_transform_like(value)) return scr_dyn_obj_read(value, "readable", 8);
|
||||
if (scr_exc_pending()) return NULL;
|
||||
return scr_dyn_retain((ScrDyn *)value);
|
||||
}
|
||||
if (scr_stream_dyn_stage_returned_stream(value)) return NULL;
|
||||
int kind = scr_dyn_iterable_kind(value);
|
||||
if (kind < 0) return NULL;
|
||||
if (kind == 0) {
|
||||
scr_stream_throw_invalid_return("Iterable, AsyncIterable or Stream", "source", value);
|
||||
return NULL;
|
||||
}
|
||||
return scr_dyn_retain((ScrDyn *)value);
|
||||
}
|
||||
|
||||
@@ -149,6 +149,18 @@ ScrStream *scr_stream_unbox_dyn(const struct ScrDyn *value, const ScrStr *messag
|
||||
* to pump rather than as a stream (false with an exception pending when
|
||||
* reading the value's members throws). */
|
||||
bool scr_stream_dyn_iterable_source(const struct ScrDyn *value);
|
||||
/* A checked-dynamic pipeline() stage the pipeline takes as a stream: a
|
||||
* native stream (+1), or Duplex.from of a value that is not stream-like
|
||||
* or a function, which the pipeline uses without piping into it; NULL
|
||||
* with an exception otherwise (`message` for the unsupported shapes). */
|
||||
ScrStream *scr_stream_unbox_dyn_stage(const struct ScrDyn *value, const ScrStr *message);
|
||||
/* The iterable pipeline() pumps for a checked-dynamic first stage: a
|
||||
* source function's result when `returned`, else a { readable, writable }
|
||||
* pair's readable side or the value (+1, or NULL with an exception). */
|
||||
struct ScrDyn *scr_stream_dyn_source_value(const struct ScrDyn *value, bool returned);
|
||||
/* An untyped function stage's result before the last pipeline stage: the
|
||||
* async iterable itself (+1), or NULL with ERR_INVALID_RETURN_VALUE. */
|
||||
struct ScrDyn *scr_stream_dyn_transform_result(const struct ScrDyn *value, const ScrStr *label);
|
||||
/* Handles that are Transforms (ScrDynHandleOps.transform: Cipheriv and
|
||||
* Decipheriv). scr_stream_of_handle: the handle's Transform (+1), built on
|
||||
* first use, or NULL when the value is not such a handle.
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
// pipeline() over untyped first stages that are functions or
|
||||
// { readable, writable } pairs: a source function is called with
|
||||
// { signal } while the pipeline is set up and its iterable result is
|
||||
// pumped; a result that is not iterable throws ERR_INVALID_RETURN_VALUE
|
||||
// from the callback form and rejects the promise form; a pair pumps its
|
||||
// readable side.
|
||||
import { pipeline, Writable, Readable, PassThrough } from "node:stream";
|
||||
import { pipeline as pipelineP } from "node:stream/promises";
|
||||
|
||||
function pick(v) {
|
||||
return v;
|
||||
}
|
||||
function sink(label) {
|
||||
return new Writable({
|
||||
objectMode: true,
|
||||
write(chunk, _enc, cb) {
|
||||
console.log(label, "write", JSON.stringify(String(chunk)));
|
||||
cb();
|
||||
},
|
||||
final(cb) {
|
||||
console.log(label, "final");
|
||||
cb();
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const arraySource = pick(function (opts) {
|
||||
console.log("array source called", typeof opts, typeof opts.signal, opts.signal.aborted);
|
||||
return ["a1", "a2"];
|
||||
});
|
||||
const before = pipelineP(arraySource, sink("array"));
|
||||
console.log("after pipeline() returned");
|
||||
await before;
|
||||
console.log("array resolved");
|
||||
|
||||
await pipelineP(pick(() => "xy"), sink("string"));
|
||||
|
||||
await pipelineP(
|
||||
pick(async function* ({ signal }) {
|
||||
console.log("generator source", typeof signal.aborted);
|
||||
yield "g1";
|
||||
await new Promise((r) => setTimeout(r, 1));
|
||||
yield "g2";
|
||||
}),
|
||||
pick(async function* (source) {
|
||||
for await (const chunk of source) yield `[${chunk}]`;
|
||||
}),
|
||||
sink("generator"),
|
||||
);
|
||||
|
||||
await new Promise((resolve) => {
|
||||
pipeline(pick(() => new Set(["s1", "s2"])), sink("set"), (err) => {
|
||||
console.log("set done", err);
|
||||
resolve(undefined);
|
||||
});
|
||||
});
|
||||
|
||||
// a result that is not iterable
|
||||
try {
|
||||
pipeline(pick(() => 5), sink("five"), () => console.log("five callback"));
|
||||
} catch (err) {
|
||||
console.log("five threw", err.name, err.code, err.message);
|
||||
}
|
||||
try {
|
||||
await pipelineP(pick(() => ({ value: 1 })), sink("object"));
|
||||
} catch (err) {
|
||||
console.log("object rejected", err.name, err.code, err.message);
|
||||
}
|
||||
|
||||
// a source function that throws
|
||||
try {
|
||||
await pipelineP(
|
||||
pick(() => {
|
||||
throw new Error("source failed");
|
||||
}),
|
||||
sink("throws"),
|
||||
);
|
||||
} catch (err) {
|
||||
console.log("throws rejected", err.message);
|
||||
}
|
||||
|
||||
// a { readable, writable } pair pumps its readable side
|
||||
const side = new PassThrough();
|
||||
await pipelineP(pick({ readable: Readable.from(["p1", "p2"]), writable: side }), sink("pair"));
|
||||
console.log("done");
|
||||
@@ -0,0 +1,112 @@
|
||||
// pipeline() with untyped stages after the first that are functions: a
|
||||
// transform is called with the previous stream or generator and
|
||||
// { signal } while the pipeline is set up, and must return an async
|
||||
// iterable; an async function destination's value resolves the promise
|
||||
// form and reaches the callback; an async generator destination feeds
|
||||
// the PassThrough that pipeline() returns. Results of the wrong kind throw
|
||||
// ERR_INVALID_RETURN_VALUE naming the stage. A stage after an untyped
|
||||
// source function or transform receives that stage's own iterator, as
|
||||
// Node's makeAsyncIterable answers an iterable itself.
|
||||
import { pipeline, Writable, Readable, PassThrough } from "node:stream";
|
||||
import { pipeline as pipelineP } from "node:stream/promises";
|
||||
|
||||
function pick(v) {
|
||||
return v;
|
||||
}
|
||||
function sink(label) {
|
||||
return new Writable({
|
||||
objectMode: true,
|
||||
write(chunk, _enc, cb) {
|
||||
console.log(label, "write", JSON.stringify(String(chunk)));
|
||||
cb();
|
||||
},
|
||||
});
|
||||
}
|
||||
const upper = pick(async function* (source, { signal }) {
|
||||
console.log("upper called", source instanceof Readable, typeof signal.aborted);
|
||||
for await (const chunk of source) yield String(chunk).toUpperCase();
|
||||
});
|
||||
const join = pick(async (source) => {
|
||||
let text = "";
|
||||
for await (const chunk of source) text += chunk;
|
||||
return text;
|
||||
});
|
||||
|
||||
await pipelineP(Readable.from(["a", "b"]), upper, sink("transform"));
|
||||
console.log("joined", await pipelineP(Readable.from(["x", "y"]), upper, join));
|
||||
console.log("two transforms", await pipelineP(pick(["m", "n"]), upper, upper, join));
|
||||
|
||||
await new Promise((resolve) => {
|
||||
pipeline(Readable.from(["c1", "c2"]), upper, join, (err, value) => {
|
||||
console.log("callback", err, value);
|
||||
resolve(undefined);
|
||||
});
|
||||
});
|
||||
|
||||
// an async generator destination
|
||||
const wrap = pick(async function* (source) {
|
||||
for await (const chunk of source) yield `<${chunk}>`;
|
||||
});
|
||||
const out = pipeline(Readable.from(["p1", "p2"]), wrap, (err, value) => {
|
||||
console.log("generator destination done", err, value);
|
||||
});
|
||||
console.log("returns a PassThrough", out instanceof PassThrough);
|
||||
for await (const chunk of out) console.log("passthrough", chunk);
|
||||
|
||||
// a destination whose promise rejects
|
||||
try {
|
||||
await pipelineP(
|
||||
Readable.from(["r"]),
|
||||
pick(async (source) => {
|
||||
for await (const chunk of source) console.log("rejecting after", chunk);
|
||||
throw new Error("destination failed");
|
||||
}),
|
||||
);
|
||||
} catch (err) {
|
||||
console.log("rejected", err.message);
|
||||
}
|
||||
|
||||
// results of the wrong kind
|
||||
try {
|
||||
await pipelineP(Readable.from(["i"]), pick(() => 5), sink("number"));
|
||||
} catch (err) {
|
||||
console.log("transform result", err.name, err.code, err.message);
|
||||
}
|
||||
try {
|
||||
await pipelineP(Readable.from(["i"]), upper, pick(() => ["sync"]), sink("array"));
|
||||
} catch (err) {
|
||||
console.log("second transform result", err.code, err.message);
|
||||
}
|
||||
try {
|
||||
pipeline(Readable.from(["i"]), pick(() => 7), () => console.log("never"));
|
||||
} catch (err) {
|
||||
console.log("destination result", err.name, err.code, err.message);
|
||||
}
|
||||
// the iterators handed from stage to stage
|
||||
let transformIter = null;
|
||||
const handed = await pipelineP(
|
||||
pick(function* () {
|
||||
yield "s1";
|
||||
yield "s2";
|
||||
}),
|
||||
pick((source) => {
|
||||
// a sync generator, not an async iterator over it
|
||||
console.log(
|
||||
"transform gets the source's iterator",
|
||||
typeof source[Symbol.asyncIterator],
|
||||
typeof source[Symbol.iterator],
|
||||
);
|
||||
transformIter = (async function* () {
|
||||
for await (const chunk of source) yield `${chunk}!`;
|
||||
})();
|
||||
return transformIter;
|
||||
}),
|
||||
pick(async (source) => {
|
||||
console.log("destination gets the transform's iterator", source === transformIter);
|
||||
let text = "";
|
||||
for await (const chunk of source) text += chunk;
|
||||
return text;
|
||||
}),
|
||||
);
|
||||
console.log("handed", handed);
|
||||
console.log("done");
|
||||
@@ -0,0 +1,64 @@
|
||||
// pipeline() with an untyped stage after the first that is not a stream or
|
||||
// a function: Node takes Duplex.from of it, so an iterable becomes the
|
||||
// readable stage the next stream is piped from, the stages before it are
|
||||
// not read and are destroyed when the pipeline finishes, and a value
|
||||
// Duplex.from rejects throws ERR_INVALID_ARG_TYPE.
|
||||
import { pipeline, Writable, Readable, Duplex } from "node:stream";
|
||||
import { pipeline as pipelineP } from "node:stream/promises";
|
||||
|
||||
function pick(v) {
|
||||
return v;
|
||||
}
|
||||
function sink(label) {
|
||||
return new Writable({
|
||||
objectMode: true,
|
||||
write(chunk, _enc, cb) {
|
||||
console.log(label, "write", JSON.stringify(String(chunk)));
|
||||
cb();
|
||||
},
|
||||
});
|
||||
}
|
||||
function watched(label, values) {
|
||||
const source = Readable.from(values);
|
||||
source.on("error", (err) => console.log(label, "error", err.code));
|
||||
source.on("close", () => console.log(label, "close", source.destroyed));
|
||||
return source;
|
||||
}
|
||||
|
||||
await pipelineP(watched("first", ["unread"]), pick(["z1", "z2"]), sink("array"));
|
||||
console.log("array resolved");
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
|
||||
await new Promise((resolve) => {
|
||||
pipeline(watched("second", ["unread"]), pick("text"), sink("string"), (err) => {
|
||||
console.log("string done", err);
|
||||
resolve(undefined);
|
||||
});
|
||||
});
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
|
||||
// a generator stage before the iterable is never read
|
||||
await pipelineP(
|
||||
Readable.from(["s"]),
|
||||
pick(async function* (source) {
|
||||
console.log("generator stage called");
|
||||
for await (const chunk of source) yield chunk;
|
||||
}),
|
||||
pick(new Set(["set1", "set2"])),
|
||||
sink("set"),
|
||||
);
|
||||
console.log("set resolved");
|
||||
|
||||
// the iterable as the last stage is what pipeline() returns
|
||||
const last = pipeline(Readable.from(["u"]), pick(["tail"]), () => console.log("never"));
|
||||
console.log(last instanceof Duplex, last.readable, last.writable);
|
||||
for await (const chunk of last) console.log("last", chunk);
|
||||
|
||||
for (const value of [42, { plain: true }, null]) {
|
||||
try {
|
||||
await pipelineP(Readable.from(["i"]), pick(value), sink("rejected"));
|
||||
} catch (err) {
|
||||
console.log(err.name, err.code, err.message);
|
||||
}
|
||||
}
|
||||
console.log("done");
|
||||
@@ -0,0 +1,50 @@
|
||||
// A CommonJS helper that passes its untyped parameters to pipeline(): the
|
||||
// same call runs over streams, function stages and iterables, whichever
|
||||
// each argument is at run time.
|
||||
const { pipeline, Transform, Writable, Readable } = require("node:stream");
|
||||
|
||||
function run(label, source, middle, destination) {
|
||||
return new Promise((resolve) => {
|
||||
pipeline(source, middle, destination, (err, value) => {
|
||||
console.log(label, "done", err ? err.code || err.message : err, value);
|
||||
resolve(undefined);
|
||||
});
|
||||
});
|
||||
}
|
||||
function sink(label) {
|
||||
return new Writable({
|
||||
objectMode: true,
|
||||
write(chunk, _enc, cb) {
|
||||
console.log(label, "write", String(chunk));
|
||||
cb();
|
||||
},
|
||||
});
|
||||
}
|
||||
const double = new Transform({
|
||||
objectMode: true,
|
||||
transform(chunk, _enc, cb) {
|
||||
cb(null, String(chunk) + String(chunk));
|
||||
},
|
||||
});
|
||||
async function* exclaim(source) {
|
||||
for await (const chunk of source) yield `${chunk}!`;
|
||||
}
|
||||
async function count(source) {
|
||||
let n = 0;
|
||||
for await (const _ of source) n++;
|
||||
return n;
|
||||
}
|
||||
|
||||
(async () => {
|
||||
await run("streams", Readable.from(["a", "b"]), double, sink("streams"));
|
||||
await run("function middle", Readable.from(["c"]), exclaim, sink("function middle"));
|
||||
await run("function ends", () => ["d", "e", "f"], exclaim, count);
|
||||
await run("iterable middle", Readable.from(["skipped"]), ["g"], sink("iterable middle"));
|
||||
await run("generator into stream", ["h"], exclaim, new Transform({
|
||||
objectMode: true,
|
||||
transform(chunk, _enc, cb) {
|
||||
console.log("transform saw", chunk);
|
||||
cb(null, chunk);
|
||||
},
|
||||
}).on("data", (d) => console.log("data", d)));
|
||||
})();
|
||||
@@ -0,0 +1,45 @@
|
||||
// pipeline() in TypeScript with stages after the first typed any or
|
||||
// unknown: functions run as function stages, streams are piped, and an
|
||||
// iterable is taken as Duplex.from of it.
|
||||
import { pipeline, Readable, Writable, PassThrough } from "node:stream";
|
||||
import { pipeline as pipelineP } from "node:stream/promises";
|
||||
|
||||
function sink(label: string): Writable {
|
||||
return new Writable({
|
||||
objectMode: true,
|
||||
write(chunk: unknown, _enc, cb) {
|
||||
console.log(label, JSON.stringify(chunk));
|
||||
cb();
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const shout: any = async function* (source: any) {
|
||||
for await (const chunk of source) yield String(chunk).toUpperCase();
|
||||
};
|
||||
await pipelineP(Readable.from(["a", "b"]), shout, sink("function"));
|
||||
|
||||
const total: unknown = async (source: any) => {
|
||||
let sum = 0;
|
||||
for await (const chunk of source) sum += Number(chunk);
|
||||
return sum;
|
||||
};
|
||||
const sum = await pipelineP(Readable.from(["1", "2", "3"]), total as any);
|
||||
console.log("sum", sum);
|
||||
|
||||
const relay: any = new PassThrough({ objectMode: true });
|
||||
await pipelineP(Readable.from([{ k: 1 }]), relay, sink("stream"));
|
||||
|
||||
const values: any = ["v1", "v2"];
|
||||
await pipelineP(Readable.from(["unread"]), values, sink("iterable"));
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
pipeline(Readable.from(["x"]), shout, sink("callback"), (err: NodeJS.ErrnoException | null) => {
|
||||
console.log("callback done", err == null);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
main().then(() => console.log("done"));
|
||||
@@ -0,0 +1,86 @@
|
||||
// Untyped pipeline function stages next to typed stages: an untyped stage
|
||||
// after a typed generator stage or a Transform, the { signal } an untyped
|
||||
// stage receives aborting when the pipeline fails, and an untyped stage
|
||||
// that throws while the pipeline is set up.
|
||||
import { pipeline, Writable, Readable, Transform } from "node:stream";
|
||||
import { pipeline as pipelineP } from "node:stream/promises";
|
||||
|
||||
function pick(v) {
|
||||
return v;
|
||||
}
|
||||
function sink(label) {
|
||||
return new Writable({
|
||||
objectMode: true,
|
||||
write(chunk, _enc, cb) {
|
||||
console.log(label, "write", JSON.stringify(String(chunk)));
|
||||
cb();
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
await pipelineP(
|
||||
Readable.from(["a", "b"]),
|
||||
async function* (source) {
|
||||
for await (const chunk of source) yield `${chunk}1`;
|
||||
},
|
||||
pick(async function* (source) {
|
||||
for await (const chunk of source) yield `${chunk}2`;
|
||||
}),
|
||||
sink("after generator"),
|
||||
);
|
||||
|
||||
await pipelineP(
|
||||
Readable.from(["q"]),
|
||||
new Transform({
|
||||
transform(chunk, _enc, cb) {
|
||||
cb(null, chunk);
|
||||
},
|
||||
}),
|
||||
pick(async (source) => {
|
||||
for await (const chunk of source) console.log("destination saw", String(chunk));
|
||||
}),
|
||||
);
|
||||
|
||||
let seen;
|
||||
try {
|
||||
await pipelineP(
|
||||
Readable.from(["x", "y"]),
|
||||
pick(async function* (source, { signal }) {
|
||||
seen = signal;
|
||||
signal.addEventListener("abort", () => console.log("signal abort event"));
|
||||
for await (const chunk of source) yield chunk;
|
||||
}),
|
||||
new Writable({
|
||||
write(_chunk, _enc, cb) {
|
||||
cb(new Error("write failed"));
|
||||
},
|
||||
}),
|
||||
);
|
||||
} catch (err) {
|
||||
console.log("rejected", err.message, seen.aborted);
|
||||
}
|
||||
|
||||
try {
|
||||
pipeline(
|
||||
Readable.from(["t"]),
|
||||
pick(() => {
|
||||
throw new Error("stage threw");
|
||||
}),
|
||||
sink("never"),
|
||||
() => console.log("never called"),
|
||||
);
|
||||
} catch (err) {
|
||||
console.log("callback form threw", err.message);
|
||||
}
|
||||
try {
|
||||
await pipelineP(
|
||||
Readable.from(["t"]),
|
||||
pick(() => {
|
||||
throw new Error("stage threw again");
|
||||
}),
|
||||
sink("never"),
|
||||
);
|
||||
} catch (err) {
|
||||
console.log("promise form rejected", err.message);
|
||||
}
|
||||
console.log("done");
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 4900, "san": 6400 }
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 6100, "san": 8500 }
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 5100, "san": 8600 }
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 5300, "san": 6900 }
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 4500, "san": 6100 }
|
||||
@@ -0,0 +1 @@
|
||||
{ "plain": 4700, "san": 6300 }
|
||||
Reference in New Issue
Block a user