diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index f3883fe8d..52bf534e8 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -65,4 +65,7 @@ Their underlying sources, sinks, and transformers, queuing strategies with a `size` function, readers and writers, `tee`, `pipeTo` and `pipeThrough` with their options and an abort signal, `ReadableStream.from`, and async iteration settle promises in Node's order, and the streams take own properties. `TextDecoderStream` decodes UTF-8; its other encodings are unsupported. A JavaScript `ReadableStream` given as a `Response` or `Request` body, or as `fetch`'s body, is read through a stream of its own: `response.body` is that stream rather than the one passed in, and the stream passed in is locked from the start. + In JavaScript, a `Response` or `Request` body has the members of a `ReadableStream`, including `tee`, `pipeTo`, `pipeThrough`, `values`, and `for await`, and its lock and `bodyUsed` follow them. + After one of those members is used, a body given to a new `Response` or `Request` is locked from the start, as a JavaScript `ReadableStream` is. + In a program that names none of the web stream classes, those members are available when the program's JavaScript names one of them or iterates a body with `for await`; reached only through a computed key, they are not. In JavaScript, a byte stream (`type: 'bytes'`), a BYOB reader, a class that extends one of these classes, `Object.getPrototypeOf` on these values, their use as `WeakMap` and `WeakSet` keys, and `ReadableStream.from` of a function that is not iterable are unsupported. diff --git a/packages/compiler/src/backend/llvm/lib-dispatch.ts b/packages/compiler/src/backend/llvm/lib-dispatch.ts index 823853ec8..57a09d85d 100644 --- a/packages/compiler/src/backend/llvm/lib-dispatch.ts +++ b/packages/compiler/src/backend/llvm/lib-dispatch.ts @@ -1274,7 +1274,7 @@ export function emitLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue { case "webstream": // A stream's start, pull and cancel steps run as promise jobs, so // a program with the classes runs its event loop. - if (e.fn === "webstream.class") host.usesTimers = true; + if (e.fn === "webstream.class" || e.fn === "webstream.install") host.usesTimers = true; return host.emitGenericLibCall(e); case "test": return emitTestLibCall(host, e); diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 4af0484f4..14476833f 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -120,6 +120,7 @@ export const LIB_FN_SYMS: Record = { "util.throttle": "scr_util_throttle", "webstream.class": "scr_webstream_class", "webstream.instanceOf": "scr_webstream_instanceof", + "webstream.install": "scr_webstream_install", "mime.typeNew": "scr_mime_type_new", "mime.typeParse": "scr_mime_type_parse", "mime.paramsNew": "scr_mime_params_new", diff --git a/packages/compiler/src/compat/fetch-profile.ts b/packages/compiler/src/compat/fetch-profile.ts index 03d261aff..83d809ee1 100644 --- a/packages/compiler/src/compat/fetch-profile.ts +++ b/packages/compiler/src/compat/fetch-profile.ts @@ -938,6 +938,9 @@ export interface StaticFetchMembers { headersCalls: ReadonlySet; readableStreamReads: ReadonlySet; readableStreamCalls: ReadonlySet; + /** ReadableStream prototype members the native stream does not + * implement, symbol-keyed ones as `[Symbol.name]`. */ + readableStreamUnsupported: ReadonlySet; } let staticFetchMemberSets: StaticFetchMembers | undefined; @@ -961,6 +964,16 @@ export function staticFetchMembers(): StaticFetchMembers { ), readableStreamReads: new Set(profile.members.readableStreamReads), readableStreamCalls: new Set(profile.members.readableStreamCalls), + readableStreamUnsupported: new Set( + profile.inventory.entries + .filter( + (entry) => + entry.owner === "ReadableStream" && + (entry.placement === "prototype" || entry.placement === "prototype-symbol") && + entry.status === "unsupported", + ) + .map((entry) => entry.member), + ), }; return staticFetchMemberSets; } diff --git a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts index 24f603198..765d3beae 100644 --- a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts +++ b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts @@ -1,9 +1,19 @@ +import { staticFetchMembers } from "../../../compat/fetch-profile.js"; import { numLit } from "../../../ir/build.js"; -import { BOOL, DYN, STRING, type IrExpr, type SrcLoc } from "../../../ir/ir.js"; +import { everyStmt } from "../../../ir/traverse.js"; +import { + BOOL, + DYN, + STRING, + VOID, + type IrExpr, + type IrFunction, + type SrcLoc, +} from "../../../ir/ir.js"; import * as ts from "../../ts7/adapter.js"; import { isJsSourceFile, locOf } from "../../program.js"; import { constituentTypes } from "../../ts7/checker.js"; -import type { Lowerer } from "../lowerer.js"; +import { ENTRY_NAME, type Lowerer } from "../lowerer.js"; import { stdlibGlobalNameOf } from "../surfaces.js"; import { streamWebImportOf } from "./module-bindings.js"; @@ -221,3 +231,98 @@ export function isWebStreamValueInJavaScript(lowerer: Lowerer, node: ts.Expressi return true; return typeInvolvesWebStream(lowerer, lowerer.typeOf(node)); } + +/** Whether a checker type is the stdlib ReadableStream, or a union with it. */ +function isReadableStreamType(lowerer: Lowerer, type: ts.Type): boolean { + if ((type.flags & ts.TypeFlags.Union) !== 0) + return constituentTypes(type).some((t) => isReadableStreamType(lowerer, t)); + const symbol = type.getSymbol(); + return ( + symbol !== undefined && + ts.symbolName(symbol) === "ReadableStream" && + lowerer.isStdlibSymbol(symbol) + ); +} + +/** Whether a member access resolves to a member of the stdlib + * ReadableStream interface. */ +function isReadableStreamMember(lowerer: Lowerer, access: ts.PropertyAccessExpression): boolean { + const member = lowerer.checker.getSymbolAtLocation(access.name); + if (member === undefined || !lowerer.isStdlibSymbol(member)) return false; + let owner: ts.Node | undefined = member.declarations[0]?.parent; + while (owner !== undefined && !ts.isInterfaceDeclaration(owner)) owner = owner.parent; + return owner !== undefined && owner.name.text === "ReadableStream"; +} + +/** Whether a checker type is any or unknown, which a fetch body can be. */ +function isUntypedType(type: ts.Type): boolean { + return (type.flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown)) !== 0; +} + +/** Whether the program links the fetch unit (a fetch. entry point), whose + * Response and Request bodies can reach JavaScript as untyped values. */ +function programLinksFetch(functions: IrFunction[]): boolean { + let found = false; + const visitor = { + expr: (e: IrExpr): boolean => { + if (e.kind === "libCall" && e.fn.startsWith("fetch.")) found = true; + return !found; + }, + stmt: (): boolean => !found, + }; + for (const fn of functions) { + for (const s of fn.body) everyStmt(s, visitor); + if (found) return true; + } + return false; +} + +/** Whether a JavaScript source of the program uses a ReadableStream member + * that the fetch unit's own stream does not implement (the fetch profile's + * readableStreamUnsupported): a member access the checker resolves to the + * interface, or `for await` over a ReadableStream value. The same member + * names on an untyped receiver, or `for await` over an untyped value, + * count in a program that links the fetch unit, since the value can be a + * fetch body. */ +function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction[]): boolean { + const members = staticFetchMembers().readableStreamUnsupported; + const iterates = members.has("[Symbol.asyncIterator]"); + let untyped = false; + for (const file of lowerer.program.getSourceFiles()) { + if (file.isDeclarationFile || !isJsSourceFile(file)) continue; + let found = false; + ts.walkPreorder(file, (node) => { + let receiver: ts.Type | null = null; + if (ts.isPropertyAccessExpression(node) && members.has(node.name.text)) { + if (isReadableStreamMember(lowerer, node)) found = true; + else receiver = lowerer.typeOf(node.expression); + } else if (iterates && ts.isForOfStatement(node) && node.awaitModifier !== undefined) { + receiver = lowerer.typeOf(node.expression); + if (isReadableStreamType(lowerer, receiver)) found = true; + } + if (found) return "stop"; + if (receiver !== null && isUntypedType(receiver)) untyped = true; + return undefined; + }); + if (found) return true; + } + return untyped && programLinksFetch(functions); +} + +/** Links the web stream classes into a program whose JavaScript uses a + * ReadableStream member the fetch unit's own stream lacks: a fetch body's + * handle stands for a web ReadableStream once such a member is used + * (scr_fetch.c, sf_stream_web_view). %main installs the classes first. A + * program that names one of the classes links them already. */ +export function installFetchBodyWebStreams(lowerer: Lowerer, functions: IrFunction[]): void { + if (programNamesWebStreamClass(lowerer) || !programUsesWebStreamOnlyMembers(lowerer, functions)) + return; + const main = functions.find((fn) => fn.name === ENTRY_NAME); + if (!main) return; + const loc = main.loc; + main.body.unshift({ + kind: "exprStmt", + expr: { kind: "libCall", fn: "webstream.install", args: [], type: VOID, loc }, + loc, + }); +} diff --git a/packages/compiler/src/frontend/lowering/lowerer.ts b/packages/compiler/src/frontend/lowering/lowerer.ts index 2c3864964..e6b6a4653 100644 --- a/packages/compiler/src/frontend/lowering/lowerer.ts +++ b/packages/compiler/src/frontend/lowering/lowerer.ts @@ -118,7 +118,11 @@ import { fenceWebCryptoMember, lowerWebCryptoCall } from "./builtins/webcrypto.j import { valueStreamOf } from "./builtins/hash-stream.js"; import { lowerNetAddressCall, lowerNetAddressNew } from "./builtins/net-address.js"; import { lowerMimeCall, lowerMimeNew } from "./builtins/mime.js"; -import { lowerWebStreamNew, lowerWebStreamStaticCall } from "./builtins/web-streams-dyn.js"; +import { + installFetchBodyWebStreams, + lowerWebStreamNew, + lowerWebStreamStaticCall, +} from "./builtins/web-streams-dyn.js"; import { dynUndefinedExpr, nodeThrowExpr, numLit, varRef } from "../../ir/build.js"; import { iteratorSlotAdapter } from "./iterator-adapters.js"; import { iterableSlotValue } from "./iterable-values.js"; @@ -9459,6 +9463,7 @@ export class Lowerer { guardFrozenWrites(this, functions); applyHandleProperties(this, functions); installBufferValue(functions); + installFetchBodyWebStreams(this, functions); // Globals typed by a class that never REGISTERED (a JS class whose // collection fenced — Symbol-keyed fields, an unsupported base): the diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 69b157748..ba35a92e3 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -1211,6 +1211,7 @@ function buildLibFnSignatures() { "util.throttle": { argTypes: [DYN, DYN, DYN, DYN], result: DYN }, "webstream.class": { argTypes: [F64], result: DYN }, "webstream.instanceOf": { argTypes: [DYN, F64], result: BOOL }, + "webstream.install": { argTypes: [], result: VOID }, "mime.typeNew": { argTypes: [DYN, DYN], result: DYN }, "mime.typeParse": { argTypes: [DYN], result: DYN }, "mime.paramsNew": { argTypes: [], result: DYN }, diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index 04392fe48..58fce50b6 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -83,6 +83,7 @@ const DYN_ASYNC_LIB_FNS: ReadonlySet = new Set([ // Web streams settle their promises with dyn values. "webstream.class", "webstream.instanceOf", + "webstream.install", "dyn.promiseAll", "dyn.promiseAllSettled", "dyn.promiseAny", @@ -390,6 +391,7 @@ const SYMBOL_LIB_FNS: ReadonlySet = new Set([ // Web streams answer async iteration through Symbol.asyncIterator. "webstream.class", "webstream.instanceOf", + "webstream.install", // EventEmitter's kShapeMode and kCapture properties ...BUILTIN_CLASS_LIB_FNS, "dyn.usingDispose", @@ -954,6 +956,7 @@ const RUNTIME_FEATURE_RULES = { // scr_webstreams.c calls the iterators ReadableStream.from reads. "webstream.class", "webstream.instanceOf", + "webstream.install", ], // Constant-key reads of Object.prototype member names answer the // inherited member through scr_dyn_invoke.c. @@ -1076,6 +1079,7 @@ const RUNTIME_FEATURE_RULES = { "global.native", // ERR_ARG_NOT_ITERABLE renders the value as util.format's %s does. "webstream.class", + "webstream.install", "util.styleText", "crypto.cipherNew", "webcrypto.encrypt", diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json new file mode 100644 index 000000000..502ef427a --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-helpers.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json new file mode 100644 index 000000000..eaed983e9 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-linked.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json new file mode 100644 index 000000000..44d798239 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-network.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json new file mode 100644 index 000000000..1b37cd0c0 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-unlinked.mjs" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_dyn.h b/packages/runtime/src/scr_dyn.h index fafa16eda..c2aed501e 100644 --- a/packages/runtime/src/scr_dyn.h +++ b/packages/runtime/src/scr_dyn.h @@ -2084,6 +2084,16 @@ ScrDyn *scr_dyn_handle_class_prototype(ScrDynHandleTag tag); /* Handles of `tag` (another unit's native representation of the same * interface) are instances of the class of `klass_tag`. */ void scr_dyn_handle_class_alias(ScrDynHandleTag tag, ScrDynHandleTag klass_tag); +/* The instance of the class a handle of an alias tag stands for, which + * the class's methods and getters run on (+1, or NULL with the exception + * pending). Without one, they throw their ERR_INVALID_THIS TypeError on + * the alias's handles. */ +void scr_dyn_handle_class_alias_view(ScrDynHandleTag tag, ScrDyn *(*view)(ScrDyn *handle)); +/* The receiver a method or getter of the class of `klass_tag` runs on for + * `self`: true with *out set to self, or to the instance an alias handle + * stands for (+1, or NULL with the exception pending); false when self is + * neither. */ +bool scr_dyn_handle_class_receiver(ScrDynHandleTag klass_tag, ScrDyn *self, ScrDyn **out); /* The class tag a handle is an instance of, or SCR_DYNH_CAPACITY for a * handle of no class. */ ScrDynHandleTag scr_dyn_handle_class_tag_of(const ScrDyn *handle); diff --git a/packages/runtime/src/scr_dyn_handle_class.c b/packages/runtime/src/scr_dyn_handle_class.c index cf2a04f76..30e2b5500 100644 --- a/packages/runtime/src/scr_dyn_handle_class.c +++ b/packages/runtime/src/scr_dyn_handle_class.c @@ -20,6 +20,8 @@ static SCR_TL const ScrDynHandleClass *hc_classes[SCR_DYNH_CAPACITY]; /* The class tag of another tag's handles + 1 (scr_dyn_handle_class_alias); * 0 = none. */ static SCR_TL uint16_t hc_aliases[SCR_DYNH_CAPACITY]; +/* scr_dyn_handle_class_alias_view, per alias tag. */ +static SCR_TL ScrDyn *(*hc_alias_views[SCR_DYNH_CAPACITY])(ScrDyn *handle); static SCR_TL ScrDyn *hc_ctors[SCR_DYNH_CAPACITY]; static SCR_TL ScrDyn *hc_prototypes[SCR_DYNH_CAPACITY]; static SCR_TL bool hc_teardown_registered; @@ -73,9 +75,13 @@ static ScrDyn *hc_getter(ScrClosure *closure, ScrDyn *const *args, size_t argc) const ScrDynHandleOps *ops = hc_ops(tag); ScrDyn *self = scr_dyn_this_get(); ScrDyn *result = NULL; - if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) { - result = ops->get(self->v.handle.ptr, accessor->name, strlen(accessor->name)); - if (!result && !scr_exc_pending()) result = scr_dyn_retain(scr_dyn_undefined()); + ScrDyn *receiver = NULL; + if (scr_dyn_handle_class_receiver(tag, self, &receiver)) { + if (receiver) { + result = ops->get(receiver->v.handle.ptr, accessor->name, strlen(accessor->name)); + if (!result && !scr_exc_pending()) result = scr_dyn_retain(scr_dyn_undefined()); + scr_dyn_release(receiver); + } } else { hc_invalid_this(ops); if (accessor->rejects && ops->rejected) result = ops->rejected(); @@ -245,6 +251,23 @@ void scr_dyn_handle_class_alias(ScrDynHandleTag tag, ScrDynHandleTag klass_tag) hc_aliases[tag] = (uint16_t)(klass_tag + 1); } +void scr_dyn_handle_class_alias_view(ScrDynHandleTag tag, ScrDyn *(*view)(ScrDyn *handle)) { + hc_alias_views[tag] = view; +} + +bool scr_dyn_handle_class_receiver(ScrDynHandleTag klass_tag, ScrDyn *self, ScrDyn **out) { + *out = NULL; + if (self->kind != SCR_DYN_HANDLE) return false; + ScrDynHandleTag tag = self->v.handle.tag; + if (tag == klass_tag) { + *out = scr_dyn_retain(self); + return true; + } + if (hc_aliases[tag] != (uint16_t)(klass_tag + 1) || !hc_alias_views[tag]) return false; + *out = hc_alias_views[tag](self); + return true; +} + ScrDynHandleTag scr_dyn_handle_class_tag_of(const ScrDyn *handle) { ScrDynHandleTag tag = handle->v.handle.tag; if (hc_classes[tag]) return tag; diff --git a/packages/runtime/src/scr_fetch.c b/packages/runtime/src/scr_fetch.c index 00dcc9b41..7dde9c6cb 100644 --- a/packages/runtime/src/scr_fetch.c +++ b/packages/runtime/src/scr_fetch.c @@ -164,6 +164,20 @@ struct SfStream { /* Consuming the body takes Uint8Array chunks only: set on a body from a * web ReadableStream and on the tee branches of such a body. */ bool u8_chunks; + /* The web ReadableStream this stream's checked-dynamic handles stand + * for once a member the fetch stream lacks is used (+1, or NULL; see + * sf_stream_web_view). It reads this stream; from then on every member + * and every use of the body go through it. */ + ScrDyn *web_view; + /* A reader of web_view (+1) that keeps it locked while this stream was + * locked when web_view was made: by a reader of its own, released + * later, or by reading the body. */ + ScrDyn *web_view_lock; + /* web_view's reader of this stream, taken at its first pull (owned + * through `reader`). */ + SfReader *view_reader; + /* web_view's source holds a reference to this stream. */ + bool view_source_live; #endif ScrArr *from_array; ScrBytes *from_bytes; @@ -917,6 +931,18 @@ static void sf_stream_drop_source_callbacks( static void sf_stream_release(SfStream *s) { if (!s) return; if (--s->rc > 0) { +#ifdef SCR_HANDLE_CLASSES + /* Only the web view's source still holds the stream: the view no + * longer needs to be found from it, and the view, which reads it, + * keeps it while anything else holds the view. */ + if (s->rc == 1 && s->view_source_live && s->web_view) { + ScrDyn *view = s->web_view, *lock = s->web_view_lock; + s->web_view = s->web_view_lock = NULL; + scr_dyn_release(lock); + scr_dyn_release(view); /* may release s */ + return; + } +#endif sf_response_discard_if_unobserved(s); return; } @@ -933,6 +959,8 @@ static void sf_stream_release(SfStream *s) { scr_dyn_release(s->from_dyn); #ifdef SCR_HANDLE_CLASSES scr_dyn_release(s->web_reader); + scr_dyn_release(s->web_view_lock); + scr_dyn_release(s->web_view); #endif scr_arr_release(s->from_array); scr_bytes_release(s->from_bytes); @@ -1629,6 +1657,97 @@ static ScrDyn *sf_web_body(const ScrDyn *body) { static bool sf_is_web_body(const ScrDyn *body) { return body && body->kind == SCR_DYN_HANDLE && body->v.handle.tag == SCR_DYNH_WS_READABLE; } + +static SfReader *sf_stream_get_reader(SfStream *s); +static bool sf_reader_read_request(SfReader *r, ScrPromise *p, const ScrFetchReadSink *sink, void *sink_ctx); +static ScrPromise *sf_stream_cancel(SfStream *s, ScrDyn *reason, bool through_reader); + +/* The web view's source (scr_webstream_readable_over): it reads the stream + * through a reader of its own, taken at the first pull. */ +static bool sf_view_read_into(void *ctx, const ScrFetchReadSink *sink, void *sink_ctx) { + SfStream *s = ctx; + if (!s->view_reader) { + SfReader *r = sf_stream_get_reader(s); + if (!r) return false; + s->view_reader = r; /* the stream owns it as its reader */ + sf_reader_release(r); + } + /* A chunk the request takes at once can close the view, whose source + * then lets the stream go while it still drains. */ + sf_stream_retain(s); + bool ok = sf_reader_read_request(s->view_reader, NULL, sink, sink_ctx); + sf_stream_release(s); + return ok; +} + +static ScrPromise *sf_view_cancel(void *ctx, ScrDyn *reason) { + SfStream *s = ctx; + return sf_stream_cancel(s, reason, s->view_reader != NULL && s->reader == s->view_reader); +} + +static void sf_view_release(void *ctx) { + SfStream *s = ctx; + s->view_source_live = false; + sf_stream_release(s); +} + +static const ScrWebStreamSource sf_view_source = {sf_view_read_into, sf_view_cancel, sf_view_release}; + +/* The web ReadableStream the stream's checked-dynamic handles stand for + * (borrowed), made on first use. Node's Response.body is one; the fetch + * stream answers getReader, cancel and locked itself until a member it + * lacks is used, and a lock it holds then keeps the view locked. */ +static ScrDyn *sf_stream_web_view(SfStream *s) { + if (s->web_view) return s->web_view; + s->web_view = scr_webstream_readable_over(&sf_view_source, sf_stream_retain(s)); + s->view_source_live = true; + if (s->reader || s->internal_lock) { + s->web_view_lock = scr_dyn_invoke(s->web_view, "getReader", NULL, 0, "ReadableStream.getReader"); + if (!s->web_view_lock) return NULL; + } + return s->web_view; +} + +/* Whether `key` names one of the fetch stream's own methods (its ops' + * method table). */ +static bool sf_stream_own_method(const char *key, size_t len) { + ScrDyn probe; + memset(&probe, 0, sizeof probe); + probe.kind = SCR_DYN_HANDLE; + probe.v.handle.tag = SCR_DYNH_WEB_STREAM; + for (const ScrDynHandleMethod *m = scr_dyn_handle_ops_of(&probe)->methods; m && m->name; m++) + if (sf_name(key, len, m->name)) return true; + return false; +} + +/* scr_dyn_handle_class_alias_view for the stream's handles (+1). */ +static ScrDyn *sf_stream_view_of(ScrDyn *handle) { + ScrDyn *view = sf_stream_web_view((SfStream *)handle->v.handle.ptr); + return view ? scr_dyn_retain(view) : NULL; +} + +/* A body whose stream has a web view is read through the view, which + * locks it as reading a body locks Node's stream: a native stream that + * reads the view (+1), or NULL with the TypeError pending. Otherwise the + * stream itself (+1). */ +static SfStream *sf_stream_body_source(SfStream *s) { + if (!s->web_view) return sf_stream_retain(s); + ScrDyn *reading = sf_web_body(s->web_view); + if (!reading) return NULL; + SfStream *out = sf_stream_retain((SfStream *)reading->v.handle.ptr); + scr_dyn_release(reading); + return out; +} +#endif + +/* Whether a body's web view is locked or disturbed, which makes the body + * unusable as one, as its own lock and disturbed state do. */ +#ifdef SCR_HANDLE_CLASSES +static bool sf_view_unusable(const SfStream *s) { + return s->web_view && scr_web_stream_bridge->unusable(s->web_view); +} +#else +#define sf_view_unusable(s) false #endif static void sf_stream_pull_from_source(SfStream *s) { @@ -1882,10 +2001,22 @@ static void sf_reader_release_lock(SfReader *r) { sf_reader_reject_message_all(r, "Invalid state: Releasing reader"); +#ifdef SCR_HANDLE_CLASSES + /* The web view was made while this reader held the lock. */ + ScrDyn *view_lock = r != s->view_reader ? s->web_view_lock : NULL; + if (view_lock) s->web_view_lock = NULL; +#endif size_t handle_refs = r->handle_refs; s->reader = NULL; r->stream = NULL; sf_reader_release(r); /* stream's ownership */ +#ifdef SCR_HANDLE_CLASSES + if (view_lock) { + ScrDyn *released = scr_dyn_invoke(view_lock, "releaseLock", NULL, 0, "ReadableStreamDefaultReader.releaseLock"); + if (released) scr_dyn_release(released); + scr_dyn_release(view_lock); + } +#endif while (handle_refs-- > 0) sf_stream_release(s); } @@ -1999,8 +2130,21 @@ static ScrPromise *sf_stream_cancel(SfStream *s, ScrDyn *reason, } static ScrPromise *sf_stream_collect(SfStream *s, int mode) { +#ifdef SCR_HANDLE_CLASSES + if (s->web_view && !(s->disturbed || s->reader || s->internal_lock || sf_view_unusable(s))) { + SfStream *source = sf_stream_body_source(s); + if (!source) { + ScrPromise *p = scr_promise_new(); + scr_promise_reject_pending(p); + return p; + } + ScrPromise *p = sf_stream_collect(source, mode); + sf_stream_release(source); + return p; + } +#endif ScrPromise *p = scr_promise_new(); - if (s->disturbed || s->reader || s->internal_lock) { + if (s->disturbed || s->reader || s->internal_lock || sf_view_unusable(s)) { sf_type_error("Body is unusable: Body has already been read"); scr_promise_reject_pending(p); return p; @@ -2456,6 +2600,14 @@ static ScrDyn *sf_stream_invoke(void *ptr, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc, const char *what) { SfStream *s = ptr; +#ifdef SCR_HANDLE_CLASSES + /* Any other member, and every member once the web view exists, is the + * view's. */ + if (s->web_view || !sf_stream_own_method(method, strlen(method))) { + ScrDyn *view = sf_stream_web_view(s); + return view ? scr_dyn_invoke(view, method, args, argc, what) : NULL; + } +#endif if (strcmp(method, "getReader") == 0) { ScrDyn *options = argc ? args[0] : NULL; ScrDyn *materialized = NULL; @@ -2494,12 +2646,35 @@ static ScrDyn *sf_stream_invoke(void *ptr, ScrDyn *self, const char *method, static ScrDyn *sf_stream_get(void *ptr, const char *key, size_t len) { SfStream *s = ptr; +#ifdef SCR_HANDLE_CLASSES + /* The stream's own members are its own until the web view exists; any + * other member makes the view and is the view's. */ + if (s->web_view || !(sf_name(key, len, "locked") || sf_stream_own_method(key, len))) { + ScrDyn *view = sf_stream_web_view(s); + if (!view) return NULL; + ScrStr *name = scr_str_new(key, len); + ScrDyn *value = scr_dyn_handle_key_get(view, name); + scr_str_release(name); + return value; + } +#endif if (sf_name(key, len, "locked")) { return scr_dyn_new_bool(s->reader != NULL || s->internal_lock); } return NULL; } +#ifdef SCR_HANDLE_CLASSES +/* Symbol.asyncIterator is the web view's. */ +static ScrDyn *sf_stream_symbol_get(void *ptr, const ScrDyn *key) { + if (!scr_dyn_is_async_iterator_symbol(key)) return NULL; + ScrDyn *view = sf_stream_web_view(ptr); + if (!view) return NULL; + const ScrDynHandleOps *ops = scr_dyn_handle_ops_of(view); + return ops->symbol_get ? ops->symbol_get(view->v.handle.ptr, key) : NULL; +} +#endif + static ScrDyn *sf_reader_invoke(void *ptr, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc, const char *what) { @@ -4098,10 +4273,30 @@ static ScrPromise *sf_response_collect(SfResponse *r, int mode) { * guard. NULL with a TypeError pending when the body is unusable. */ static SfResponse *sf_response_clone(SfResponse *r, const char *unusable) { if (!r->null_body && - (r->body->disturbed || r->body->reader || r->body->internal_lock)) { + (r->body->disturbed || r->body->reader || r->body->internal_lock || sf_view_unusable(r->body))) { sf_type_error(unusable); return NULL; } +#ifdef SCR_HANDLE_CLASSES + /* A body with a web view tees through it; each side reads one branch. */ + SfStream *viewed[2] = {NULL, NULL}; + if (!r->null_body && r->body->web_view) { + ScrDyn *branches = scr_dyn_invoke(r->body->web_view, "tee", NULL, 0, "ReadableStream.tee"); + if (!branches) return NULL; + for (int i = 0; i < 2; i++) { + ScrDyn *reading = sf_web_body(branches->v.arr.items[i]); + if (!reading) { + sf_stream_release(viewed[0]); + scr_dyn_release(branches); + return NULL; + } + viewed[i] = sf_stream_retain((SfStream *)reading->v.handle.ptr); + viewed[i]->u8_chunks = r->body->u8_chunks; + scr_dyn_release(reading); + } + scr_dyn_release(branches); + } +#endif SfResponse *c = calloc(1, sizeof *c); if (!c) sf_oom(); c->rc = 1; @@ -4110,7 +4305,16 @@ static SfResponse *sf_response_clone(SfResponse *r, const char *unusable) { sf_stream_close(c->body); } else { SfStream *branches[2]; +#ifdef SCR_HANDLE_CLASSES + if (viewed[0]) { + branches[0] = viewed[0]; + branches[1] = viewed[1]; + } else { + sf_stream_tee(r->body, branches); + } +#else sf_stream_tee(r->body, branches); +#endif sf_stream_release(r->body); r->body = branches[0]; c->body = branches[1]; @@ -4177,6 +4381,12 @@ static ScrDyn *sf_response_get(void *ptr, const char *key, size_t len) { sf_name(key, len, "statusText")) sf_response_materialize(r); if (sf_name(key, len, "bodyUsed")) { +#ifdef SCR_HANDLE_CLASSES + /* A read, cancel or pipe started on the web view disturbs it at once, + * before its pull reaches this stream. */ + if (!r->null_body && r->body->web_view && scr_web_stream_bridge->disturbed(r->body->web_view)) + return scr_dyn_new_bool(true); +#endif return scr_dyn_new_bool(!r->null_body && r->body->disturbed); } if (sf_name(key, len, "body")) { @@ -6153,13 +6363,28 @@ static bool sf_start_hop(SfTransfer *t) { stream = sf_stream_of(body); } if (!stream || stream->reader || stream->internal_lock || - stream->disturbed) { + stream->disturbed || sf_view_unusable(stream)) { scr_dyn_release(seeded); if (t->client) scr_http_client_destroy(t->client); sf_reject(t, "Response body object should not be disturbed or locked"); return false; } +#ifdef SCR_HANDLE_CLASSES + /* A stream with a web view uploads through it; the transfer holds the + * stream that reads the view. */ + SfStream *through = NULL; + if (stream->web_view) { + through = sf_stream_body_source(stream); + if (!through) { + scr_dyn_release(seeded); + if (t->client) scr_http_client_destroy(t->client); + sf_reject(t, "Response body object should not be disturbed or locked"); + return false; + } + stream = through; + } +#endif stream->disturbed = true; stream->internal_lock = true; stream->request_owner = t; @@ -6169,6 +6394,9 @@ static bool sf_start_hop(SfTransfer *t) { !stream->error && !t->done && t->client) { sf_stream_pull(stream); } +#ifdef SCR_HANDLE_CLASSES + sf_stream_release(through); +#endif scr_dyn_release(seeded); } else { scr_http_client_end(client); @@ -6429,12 +6657,17 @@ static ScrDyn *sf_response_construct(ScrDyn *body, ScrDyn *init, } else if (body->kind == SCR_DYN_HANDLE && body->v.handle.tag == SCR_DYNH_WEB_STREAM) { stream = (SfStream *)body->v.handle.ptr; - if (stream->reader || stream->internal_lock || stream->disturbed) { + if (stream->reader || stream->internal_lock || stream->disturbed || sf_view_unusable(stream)) { stream = NULL; /* borrowed until the successful retain below */ sf_type_error("Response body object should not be disturbed or locked"); goto fail; } +#ifdef SCR_HANDLE_CLASSES + stream = sf_stream_body_source(stream); + if (!stream) goto fail; +#else sf_stream_retain(stream); +#endif #ifdef SCR_HANDLE_CLASSES } else if (sf_is_web_body(body)) { ScrDyn *web = sf_web_body(body); @@ -6676,7 +6909,8 @@ ScrDyn *scr_fetch_request_new(ScrDyn *input, ScrDyn *init) { ScrDyn *body = sf_response_init_get(init, "body", 4); bool supplied = body && body->kind != SCR_DYN_UNDEF && body->kind != SCR_DYN_NULL; if (!supplied && source && !source->payload->null_body) { - if (source->payload->body->disturbed || source->payload->body->reader || source->payload->body->internal_lock) { + if (source->payload->body->disturbed || source->payload->body->reader || source->payload->body->internal_lock || + sf_view_unusable(source->payload->body)) { scr_dyn_release(body); scr_str_release(method_text); scr_url_release(url); sf_type_error("Cannot construct a Request with a Request object that has already been used."); return NULL; } @@ -7195,6 +7429,9 @@ static const ScrDynHandleOps sf_abort_controller_ops = { static const ScrDynHandleOps sf_stream_ops = { "ReadableStream", &sf_stream_retain_v, &sf_stream_release_v, &sf_stream_invoke, &sf_stream_get, &sf_no_set, NULL, +#ifdef SCR_HANDLE_CLASSES + .symbol_get = &sf_stream_symbol_get, +#endif .methods = sf_stream_methods, .invalid_this = "Value of \"this\" must be of type ReadableStream", .rejected = &sf_rejected}; @@ -7273,6 +7510,9 @@ void scr_fetch_install(void) { scr_dyn_handle_install(SCR_DYNH_ABORT_CONTROLLER, &sf_abort_controller_ops); scr_dyn_handle_install(SCR_DYNH_WEB_STREAM, &sf_stream_ops); +#ifdef SCR_HANDLE_CLASSES + scr_dyn_handle_class_alias_view(SCR_DYNH_WEB_STREAM, &sf_stream_view_of); +#endif scr_dyn_handle_install(SCR_DYNH_WEB_READER, &sf_reader_ops); scr_dyn_handle_install(SCR_DYNH_WEB_CONTROLLER, &sf_controller_ops); scr_dyn_handle_install(SCR_DYNH_FETCH_REQUEST, &sf_request_ops); diff --git a/packages/runtime/src/scr_generator.h b/packages/runtime/src/scr_generator.h index 6b66e8f9d..bbb75d90d 100644 --- a/packages/runtime/src/scr_generator.h +++ b/packages/runtime/src/scr_generator.h @@ -231,8 +231,29 @@ typedef struct ScrWebStreamBridge { /* Whether a ReadableStream is locked or disturbed, which makes it * unusable as a body. */ bool (*unusable)(const ScrDyn *stream); + /* Whether a ReadableStream is disturbed: a read, a cancel or a pipe + * has started on it. */ + bool (*disturbed)(const ScrDyn *stream); } ScrWebStreamBridge; extern SCR_TL const ScrWebStreamBridge *scr_web_stream_bridge; +/* The read side of a stream another unit implements, which + * scr_webstream_readable_over reads: read_into queues one read request on + * it (false with the exception pending), cancel cancels it (+1 promise), + * and release drops the context. */ +typedef struct ScrWebStreamSource { + bool (*read_into)(void *ctx, const ScrFetchReadSink *sink, void *sink_ctx); + ScrPromise *(*cancel)(void *ctx, ScrDyn *reason); + void (*release)(void *ctx); +} ScrWebStreamSource; +/* A checked-dynamic ReadableStream (+1 handle) with a zero high-water mark + * that reads one chunk of `source` per pull (scr_webstreams.c). ctx's + * reference moves in; the stream releases it when it goes away. */ +ScrDyn *scr_webstream_readable_over(const ScrWebStreamSource *source, void *ctx); +/* Installs the web stream classes (webstream.install): a program whose + * JavaScript uses a ReadableStream member the fetch unit's streams lack + * calls it at startup, which links the classes that fetch bodies then + * stand for. */ +void scr_webstream_install(void); /* The web stream class constructors for checked-dynamic code, by kind * (scr_webstreams.c: ReadableStream, ReadableStreamDefaultReader, * ReadableStreamDefaultController, CountQueuingStrategy, diff --git a/packages/runtime/src/scr_json.c b/packages/runtime/src/scr_json.c index 1177231ee..76bce14a4 100644 --- a/packages/runtime/src/scr_json.c +++ b/packages/runtime/src/scr_json.c @@ -7015,16 +7015,20 @@ static ScrDyn *scr_dynh_method_call(ScrClosure *closure, ScrDyn *const *args, si const ScrDynHandleOps *ops = scr_dyn_handle_ops(tag); ScrDyn *self = scr_dyn_this_get(); ScrDyn *result; - if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) { #ifdef SCR_HANDLE_CLASSES - /* A class's method value reaches the class's method past the handle's - * own properties; this build is the one programs with such classes - * link. */ - result = scr_dyn_handle_class_method_call(ops, self, method->name, args, argc); -#else - result = ops->invoke(self->v.handle.ptr, self, method->name, args, argc, method->name); -#endif + /* A class's method value reaches the class's method past the handle's + * own properties, and runs on the instance a handle of an alias tag + * stands for; this build is the one programs with such classes link. */ + ScrDyn *receiver = NULL; + if (scr_dyn_handle_class_receiver(tag, self, &receiver)) { + result = receiver ? scr_dyn_handle_class_method_call(ops, receiver, method->name, args, argc) : NULL; + scr_dyn_release(receiver); } else { +#else + if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) { + result = ops->invoke(self->v.handle.ptr, self, method->name, args, argc, method->name); + } else { +#endif const char *message = method->invalid_this ? method->invalid_this : ops->invalid_this ? ops->invalid_this : "Illegal invocation"; diff --git a/packages/runtime/src/scr_webstreams.c b/packages/runtime/src/scr_webstreams.c index 1ea9c754e..f3294faa6 100644 --- a/packages/runtime/src/scr_webstreams.c +++ b/packages/runtime/src/scr_webstreams.c @@ -1691,6 +1691,96 @@ static ScrDyn *ws_readable_from_thunk(ScrClosure *closure, ScrDyn *const *args, return ws_readable_from(argc > 0 ? args[0] : scr_dyn_undefined()); } +/* ── streams over another unit's reads (scr_webstream_readable_over) ─ */ + +typedef struct WsOver { + WsObj obj; + WsReadable *stream; /* +1 */ + const ScrWebStreamSource *source; + void *ctx; /* released through source->release */ +} WsOver; + +static void ws_over_trace(void *obj, ScrTraceVisit visit, void *ctx) { ws_visit(((WsOver *)obj)->stream, visit, ctx); } + +static void ws_over_drop(void *obj, bool gc) { + WsOver *o = obj; + ws_drop_obj(o->stream, gc); + o->stream = NULL; + /* The context is the other unit's, outside the collector. */ + void *ctx = o->ctx; + o->ctx = NULL; + if (ctx) o->source->release(ctx); +} + +static const WsObjType ws_over_type = {ws_over_trace, ws_over_drop}; + +/* The stream's controller while it still holds this source. */ +static WsRController *ws_over_controller(WsOver *o) { + WsRController *c = o->stream ? o->stream->controller : NULL; + return c && !c->algorithms_cleared ? c : NULL; +} + +static void ws_over_chunk(void *ctx, ScrDyn *value) { + WsOver *o = ctx; + WsRController *c = ws_over_controller(o); + if (c) { + if (ws_rcontroller_enqueue(c, value)) { + ws_pull_fulfilled(c, NULL); + } else { + ScrDyn *error = ws_take_pending(); + ws_rcontroller_error(c, error); + scr_dyn_release(error); + } + } + ws_release(o); +} + +static void ws_over_close(void *ctx) { + WsOver *o = ctx; + WsRController *c = ws_over_controller(o); + if (c) ws_rcontroller_close(c); + ws_release(o); +} + +static void ws_over_error(void *ctx, ScrDyn *reason) { + WsOver *o = ctx; + WsRController *c = ws_over_controller(o); + if (c) ws_rcontroller_error(c, reason ? reason : scr_dyn_undefined()); + ws_release(o); +} + +static const ScrFetchReadSink ws_over_sink = {ws_over_chunk, ws_over_close, ws_over_error}; + +static ScrDyn *ws_over_pull(void *ctx, WsRController *c) { + (void)c; + WsOver *o = ctx; + if (!o->source->read_into(o->ctx, &ws_over_sink, ws_retain(o))) { + ws_release(o); + return NULL; + } + return WS_PARKED; +} + +static ScrPromise *ws_over_cancel(void *ctx, WsRController *c, ScrDyn *reason) { + (void)c; + WsOver *o = ctx; + return o->source->cancel(o->ctx, reason); +} + +static const WsSourceOps ws_over_source_ops = {ws_over_pull, ws_over_cancel}; + +ScrDyn *scr_webstream_readable_over(const ScrWebStreamSource *source, void *ctx) { + ws_install(); + WsOver *o = ws_obj_new(sizeof *o, &ws_over_type); + o->source = source; + o->ctx = ctx; + WsReadable *s = ws_readable_create_native(&ws_over_source_ops, o, 0, 0); + o->stream = ws_readable_retain(s); + ScrDyn *box = ws_stream_box(s); + ws_readable_release(s); + return box; +} + /* ── async iteration (ReadableStream.prototype.values) ────────────── */ typedef struct WsIter { @@ -2047,7 +2137,12 @@ static bool ws_readable_unusable(const ScrDyn *stream) { return !s || ws_readable_locked(s) || s->disturbed; } -static const ScrWebStreamBridge ws_fetch_bridge = {ws_reader_read_into, ws_readable_unusable}; +static bool ws_readable_disturbed(const ScrDyn *stream) { + WsReadable *s = ws_stream_of(stream); + return s && s->disturbed; +} + +static const ScrWebStreamBridge ws_fetch_bridge = {ws_reader_read_into, ws_readable_unusable, ws_readable_disturbed}; /* ── WritableStream (lib/internal/webstreams/writablestream.js) ──── */ @@ -5184,6 +5279,8 @@ static ScrDynHandleTag ws_class_tag(double kind) { return tags[index]; } +void scr_webstream_install(void) { ws_install(); } + /* The web stream class `kind` names as a constructor value (+1): the * compiler's webstream.class. */ ScrDyn *scr_webstream_class(double kind) { diff --git a/tests/corpus/fetch-body-web-stream-helpers.mjs b/tests/corpus/fetch-body-web-stream-helpers.mjs new file mode 100644 index 000000000..b9f12a86b --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-helpers.mjs @@ -0,0 +1,15 @@ +// Fetch bodies reached only through untyped helpers, in a program that +// names none of the web stream classes: tee and for await in functions +// whose parameters have no type. +const decoder = new TextDecoder(); +const split = (body) => body.tee(); +const drain = async (stream) => { + let out = ""; + for await (const chunk of stream) out += decoder.decode(chunk); + return out; +}; +const response = new Response("through helpers"); +const [left, right] = split(response.body); +console.log("helpers", await drain(left), await drain(right), response.bodyUsed); +const request = new Request("http://example.com/", { method: "POST", body: "request helper", duplex: "half" }); +console.log("request", await drain(request.body), request.bodyUsed); diff --git a/tests/corpus/fetch-body-web-stream-linked.mjs b/tests/corpus/fetch-body-web-stream-linked.mjs new file mode 100644 index 000000000..75f3c0952 --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-linked.mjs @@ -0,0 +1,116 @@ +// Fetch Response and Request bodies as web ReadableStreams in a program +// that also names the classes: tee, pipeTo, pipeThrough, values and +// for await on a body, the body's lock and bodyUsed across them, and the +// body's other uses (text, clone, a new Response) after one. +const decoder = new TextDecoder(); +const text = (chunks) => chunks.map((c) => decoder.decode(c)).join(""); +const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e.name}:${e.message}`); +const caught = (fn) => { + try { + return `ok:${fn()}`; + } catch (e) { + return `${e.name}:${e.message}`; + } +}; +// Response.clone runs on checked-dynamic handles; the clone goes through +// an untyped parameter. +const cloneOf = (value) => value.clone(); +const readAll = async (stream) => { + const reader = stream.getReader(); + const chunks = []; + for (;;) { + const { done, value } = await reader.read(); + if (done) return text(chunks); + chunks.push(value); + } +}; + +{ + const response = new Response("tee body"); + const body = response.body; + console.log("members", typeof body.tee, typeof body.pipeTo, typeof body.pipeThrough, typeof body.values, "tee" in body, body instanceof ReadableStream); + console.log("before", body.locked, response.bodyUsed); + const [a, b] = body.tee(); + console.log("teed", body.locked, response.bodyUsed, a instanceof ReadableStream, response.body === body); + console.log("branches", await readAll(a), await readAll(b), response.bodyUsed); + console.log("used", await settled(response.text()), caught(() => body.tee())); +} + +{ + const chunks = []; + const response = new Response("piped"); + await response.body.pipeTo(new WritableStream({ write(c) { chunks.push(c); } })); + console.log("pipeTo", text(chunks), response.bodyUsed, response.body.locked); +} + +{ + const upper = new TransformStream({ transform(chunk, c) { c.enqueue(decoder.decode(chunk).toUpperCase()); } }); + const out = new Response("through").body.pipeThrough(upper); + const reader = out.getReader(); + console.log("pipeThrough", (await reader.read()).value, (await reader.read()).done); +} + +{ + const request = new Request("http://example.com/", { method: "POST", body: "iterate me", duplex: "half" }); + let joined = ""; + for await (const chunk of request.body) joined += decoder.decode(chunk); + console.log("for await", joined, request.bodyUsed, request.body.locked); + const values = []; + for await (const chunk of new Response("values").body.values()) values.push(decoder.decode(chunk)); + console.log("values", values.join("|")); + const early = new Response("early"); + for await (const chunk of early.body) { + console.log("break", chunk.length); + break; + } + console.log("after break", early.bodyUsed, early.body.locked, await settled(early.text())); +} + +{ + const response = new Response("read first"); + const reader = response.body.getReader(); + console.log("reader first", caught(() => { response.body.tee(); return "teed"; }), response.body.locked); + reader.releaseLock(); + const [a] = response.body.tee(); + console.log("after release", response.body.locked, await readAll(a)); +} + +{ + const response = new Response("member value"); + const tee = response.body.tee; + console.log("method value", tee === ReadableStream.prototype.tee, Array.isArray(tee.call(response.body)), response.body.locked); + const viewed = new Response("viewed"); + void viewed.body.values; + console.log("viewed text", viewed.body.locked, viewed.bodyUsed, await viewed.text(), viewed.bodyUsed, viewed.body.locked); + const cloned = new Response("cloned"); + void cloned.body.tee; + const copy = cloneOf(cloned); + console.log("viewed clone", await copy.text(), await cloned.text()); + const source = new Response("rewrapped"); + void source.body.pipeTo; + const rewrapped = new Response(source.body); + console.log("rewrapped", source.bodyUsed, await rewrapped.text(), source.bodyUsed); +} + +{ + // Starting a read, a cancel, a pipe or an iteration on the body's stream + // updates bodyUsed when it starts, as in Node. + const read = new Response("read now"); + void read.body.values; + const pending = read.body.getReader().read(); + console.log("bodyUsed read", read.bodyUsed); + await pending; + const canceled = new Response("cancel now"); + void canceled.body.tee; + const cancel = canceled.body.cancel("why"); + console.log("bodyUsed cancel", canceled.bodyUsed); + await cancel; + const piped = new Response("pipe now"); + const pipe = piped.body.pipeTo(new WritableStream()); + console.log("bodyUsed pipe", piped.bodyUsed, piped.body.locked); + await pipe; + const iterated = new Response("iterate now"); + const step = iterated.body.values().next(); + console.log("bodyUsed next", iterated.bodyUsed); + await step; +} diff --git a/tests/corpus/fetch-body-web-stream-network.mjs b/tests/corpus/fetch-body-web-stream-network.mjs new file mode 100644 index 000000000..d7e3b59d5 --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-network.mjs @@ -0,0 +1,26 @@ +// fetch() response bodies from a local server as web ReadableStreams: for +// await, tee, text() after a member read, and a break out of the +// iteration, with the body's bodyUsed after them. +import http from "node:http"; +const decoder = new TextDecoder(); +const server = http.createServer((req, res) => { + res.write("chunk-one;"); + setTimeout(() => { res.end("chunk-two"); }, 5); +}); +await new Promise((r) => server.listen(0, "127.0.0.1", r)); +const url = `http://127.0.0.1:${server.address().port}/`; +const a = await fetch(url); +let s = ""; +for await (const c of a.body) s += decoder.decode(c); +console.log("for await", s, a.bodyUsed); +const b = await fetch(url); +const [x, y] = b.body.tee(); +const readAll = async (st) => { let t = ""; for await (const c of st) t += decoder.decode(c); return t; }; +console.log("tee", await readAll(x), await readAll(y)); +const c = await fetch(url); +void c.body.values; +console.log("text", await c.text()); +const d = await fetch(url); +for await (const chunk of d.body) { console.log("first", decoder.decode(chunk)); break; } +console.log("after break", d.bodyUsed); +server.close(); diff --git a/tests/corpus/fetch-body-web-stream-unlinked.mjs b/tests/corpus/fetch-body-web-stream-unlinked.mjs new file mode 100644 index 000000000..d9c7f4c38 --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-unlinked.mjs @@ -0,0 +1,25 @@ +// Fetch bodies as web ReadableStreams in a program that names none of the +// web stream classes: tee, values and for await on Response and Request +// bodies, and the body's lock and bodyUsed after them. +const decoder = new TextDecoder(); +const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e.name}:${e.message}`); +const drain = async (stream) => { + let out = ""; + for await (const chunk of stream) out += decoder.decode(chunk); + return out; +}; + +const response = new Response("two ways"); +const [left, right] = response.body.tee(); +console.log("tee", await drain(left), await drain(right), response.bodyUsed, response.body.locked); +console.log("used", await settled(response.text())); + +const request = new Request("http://example.com/", { method: "PUT", body: "request body", duplex: "half" }); +console.log("for await", await drain(request.body), request.bodyUsed); + +const parts = []; +for await (const chunk of new Response("values()").body.values({ preventCancel: true })) parts.push(chunk.length); +console.log("values", parts.join(",")); + +const plain = new Response("plain"); +console.log("plain", plain.body.locked, await plain.text(), plain.bodyUsed); diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.json new file mode 100644 index 000000000..b65433b72 --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.json @@ -0,0 +1 @@ +{ "plain": 2600, "san": 48000 } diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json new file mode 100644 index 000000000..2a2bade0b --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json @@ -0,0 +1 @@ +{ "plain": 3200, "san": 52000 } diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json new file mode 100644 index 000000000..2a2bade0b --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json @@ -0,0 +1 @@ +{ "plain": 3200, "san": 52000 } diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json new file mode 100644 index 000000000..2a2bade0b --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json @@ -0,0 +1 @@ +{ "plain": 3200, "san": 52000 }