@samitouri / QOS-React / commits / 9f540fcc51

[Flight] Support streaming of decodeReply in Edge environments (#31852)

We support streaming `multipart/form-data` in Node.js using Busboy since that's kind of the idiomatic ecosystem way for handling these stream there. There's not really anything idiomatic like that for Edge that's universal yet. This adds a version that's basically just `AsyncIterable.from(formData)`. It could also be a `ReadableStream` of those entries since those are also `AsyncIterable`. I imagine that in the future we might add one from a binary `ReadableStream` that does the parsing built-in.

Sebastian Markbåge committed Dec 19, 2024 at 12:54 UTC 9f540fcc51eae6fb6eab8d4ccba00cb0477a6b7d
13 files changed +196
packages/react-server-dom-parcel/npm/server.edge.js
+1
@@ -9,6 +9,7 @@ if (process.env.NODE_ENV === 'production') {
9
10 exports.renderToReadableStream = s.renderToReadableStream;
11 exports.decodeReply = s.decodeReply;
12 +exports.decodeReplyFromAsyncIterable = s.decodeReplyFromAsyncIterable;
13 exports.decodeAction = s.decodeAction;
14 exports.decodeFormState = s.decodeFormState;
15 exports.createClientReference = s.createClientReference;
packages/react-server-dom-parcel/server.edge.js
+1
@@ -10,6 +10,7 @@
10 export {
11 renderToReadableStream,
12 decodeReply,
13 + decodeReplyFromAsyncIterable,
14 decodeAction,
15 decodeFormState,
16 createClientReference,
packages/react-server-dom-parcel/src/server/ReactFlightDOMServerEdge.js
+49
@@ -17,6 +17,8 @@ import {
17 type ServerReferenceId,
18 } from '../client/ReactFlightClientConfigBundlerParcel';
19
20 +import {ASYNC_ITERATOR} from 'shared/ReactSymbols';
21 +
22 import {
23 createRequest,
24 createPrerenderRequest,
@@ -30,6 +32,9 @@ import {
32 createResponse,
33 close,
34 getRoot,
35 + reportGlobalError,
36 + resolveField,
37 + resolveFile,
38 } from 'react-server/src/ReactFlightReplyServer';
39
40 import {
@@ -189,6 +194,50 @@ export function decodeReply<T>(
194 return root;
195 }
196
197 +export function decodeReplyFromAsyncIterable<T>(
198 + iterable: AsyncIterable<[string, string | File]>,
199 + options?: {temporaryReferences?: TemporaryReferenceSet},
200 +): Thenable<T> {
201 + const iterator: AsyncIterator<[string, string | File]> =
202 + iterable[ASYNC_ITERATOR]();
203 +
204 + const response = createResponse(
205 + serverManifest,
206 + '',
207 + options ? options.temporaryReferences : undefined,
208 + );
209 +
210 + function progress(
211 + entry:
212 + | {done: false, +value: [string, string | File], ...}
213 + | {done: true, +value: void, ...},
214 + ) {
215 + if (entry.done) {
216 + close(response);
217 + } else {
218 + const [name, value] = entry.value;
219 + if (typeof value === 'string') {
220 + resolveField(response, name, value);
221 + } else {
222 + resolveFile(response, name, value);
223 + }
224 + iterator.next().then(progress, error);
225 + }
226 + }
227 + function error(reason: Error) {
228 + reportGlobalError(response, reason);
229 + if (typeof (iterator: any).throw === 'function') {
230 + // The iterator protocol doesn't necessarily include this but a generator do.
231 + // $FlowFixMe should be able to pass mixed
232 + iterator.throw(reason).then(error, error);
233 + }
234 + }
235 +
236 + iterator.next().then(progress, error);
237 +
238 + return getRoot(response);
239 +}
240 +
241 export function decodeAction<T>(body: FormData): Promise<() => T> | null {
242 return decodeActionImpl(body, serverManifest);
243 }
packages/react-server-dom-parcel/src/server/react-flight-dom-server.edge.js
+1
@@ -11,6 +11,7 @@ export {
11 renderToReadableStream,
12 prerender as unstable_prerender,
13 decodeReply,
14 + decodeReplyFromAsyncIterable,
15 decodeAction,
16 decodeFormState,
17 createClientReference,
packages/react-server-dom-turbopack/npm/server.edge.js
+1
@@ -9,6 +9,7 @@ if (process.env.NODE_ENV === 'production') {
9
10 exports.renderToReadableStream = s.renderToReadableStream;
11 exports.decodeReply = s.decodeReply;
12 +exports.decodeReplyFromAsyncIterable = s.decodeReplyFromAsyncIterable;
13 exports.decodeAction = s.decodeAction;
14 exports.decodeFormState = s.decodeFormState;
15 exports.registerServerReference = s.registerServerReference;
packages/react-server-dom-turbopack/server.edge.js
+1
@@ -10,6 +10,7 @@
10 export {
11 renderToReadableStream,
12 decodeReply,
13 + decodeReplyFromAsyncIterable,
14 decodeAction,
15 decodeFormState,
16 registerServerReference,
packages/react-server-dom-turbopack/src/server/ReactFlightDOMServerEdge.js
+51
@@ -12,6 +12,8 @@ import type {Thenable} from 'shared/ReactTypes';
12 import type {ClientManifest} from './ReactFlightServerConfigTurbopackBundler';
13 import type {ServerManifest} from 'react-client/src/ReactFlightClientConfig';
14
15 +import {ASYNC_ITERATOR} from 'shared/ReactSymbols';
16 +
17 import {
18 createRequest,
19 createPrerenderRequest,
@@ -25,6 +27,9 @@ import {
27 createResponse,
28 close,
29 getRoot,
30 + reportGlobalError,
31 + resolveField,
32 + resolveFile,
33 } from 'react-server/src/ReactFlightReplyServer';
34
35 import {
@@ -183,10 +188,56 @@ function decodeReply<T>(
188 return root;
189 }
190
191 +function decodeReplyFromAsyncIterable<T>(
192 + iterable: AsyncIterable<[string, string | File]>,
193 + turbopackMap: ServerManifest,
194 + options?: {temporaryReferences?: TemporaryReferenceSet},
195 +): Thenable<T> {
196 + const iterator: AsyncIterator<[string, string | File]> =
197 + iterable[ASYNC_ITERATOR]();
198 +
199 + const response = createResponse(
200 + turbopackMap,
201 + '',
202 + options ? options.temporaryReferences : undefined,
203 + );
204 +
205 + function progress(
206 + entry:
207 + | {done: false, +value: [string, string | File], ...}
208 + | {done: true, +value: void, ...},
209 + ) {
210 + if (entry.done) {
211 + close(response);
212 + } else {
213 + const [name, value] = entry.value;
214 + if (typeof value === 'string') {
215 + resolveField(response, name, value);
216 + } else {
217 + resolveFile(response, name, value);
218 + }
219 + iterator.next().then(progress, error);
220 + }
221 + }
222 + function error(reason: Error) {
223 + reportGlobalError(response, reason);
224 + if (typeof (iterator: any).throw === 'function') {
225 + // The iterator protocol doesn't necessarily include this but a generator do.
226 + // $FlowFixMe should be able to pass mixed
227 + iterator.throw(reason).then(error, error);
228 + }
229 + }
230 +
231 + iterator.next().then(progress, error);
232 +
233 + return getRoot(response);
234 +}
235 +
236 export {
237 renderToReadableStream,
238 prerender,
239 decodeReply,
240 + decodeReplyFromAsyncIterable,
241 decodeAction,
242 decodeFormState,
243 };
packages/react-server-dom-turbopack/src/server/react-flight-dom-server.edge.js
+1
@@ -11,6 +11,7 @@ export {
11 renderToReadableStream,
12 prerender as unstable_prerender,
13 decodeReply,
14 + decodeReplyFromAsyncIterable,
15 decodeAction,
16 decodeFormState,
17 registerServerReference,
packages/react-server-dom-webpack/npm/server.edge.js
+1
@@ -9,6 +9,7 @@ if (process.env.NODE_ENV === 'production') {
9
10 exports.renderToReadableStream = s.renderToReadableStream;
11 exports.decodeReply = s.decodeReply;
12 +exports.decodeReplyFromAsyncIterable = s.decodeReplyFromAsyncIterable;
13 exports.decodeAction = s.decodeAction;
14 exports.decodeFormState = s.decodeFormState;
15 exports.registerServerReference = s.registerServerReference;
packages/react-server-dom-webpack/server.edge.js
+1
@@ -10,6 +10,7 @@
10 export {
11 renderToReadableStream,
12 decodeReply,
13 + decodeReplyFromAsyncIterable,
14 decodeAction,
15 decodeFormState,
16 registerServerReference,
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMReplyEdge-test.js
+36
@@ -272,4 +272,40 @@ describe('ReactFlightDOMReplyEdge', () => {
272 expect(error).not.toBe(null);
273 expect(error.message).toBe('Connection closed.');
274 });
275 +
276 + it('can stream the decoding using an async iterable', async () => {
277 + let resolve;
278 + const promise = new Promise(r => (resolve = r));
279 +
280 + const buffer = new Uint8Array([
281 + 123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,
282 + ]);
283 +
284 + const formData = await ReactServerDOMClient.encodeReply({
285 + a: Promise.resolve('hello'),
286 + b: Promise.resolve(buffer),
287 + });
288 +
289 + const iterable = {
290 + async *[Symbol.asyncIterator]() {
291 + // eslint-disable-next-line no-for-of-loops/no-for-of-loops
292 + for (const entry of formData) {
293 + yield entry;
294 + await promise;
295 + }
296 + },
297 + };
298 +
299 + const decoded = await ReactServerDOMServer.decodeReplyFromAsyncIterable(
300 + iterable,
301 + webpackServerMap,
302 + );
303 +
304 + expect(Object.keys(decoded)).toEqual(['a', 'b']);
305 +
306 + await resolve();
307 +
308 + expect(await decoded.a).toBe('hello');
309 + expect(Array.from(await decoded.b)).toEqual(Array.from(buffer));
310 + });
311 });
packages/react-server-dom-webpack/src/server/ReactFlightDOMServerEdge.js
+51
@@ -12,6 +12,8 @@ import type {Thenable} from 'shared/ReactTypes';
12 import type {ClientManifest} from './ReactFlightServerConfigWebpackBundler';
13 import type {ServerManifest} from 'react-client/src/ReactFlightClientConfig';
14
15 +import {ASYNC_ITERATOR} from 'shared/ReactSymbols';
16 +
17 import {
18 createRequest,
19 createPrerenderRequest,
@@ -25,6 +27,9 @@ import {
27 createResponse,
28 close,
29 getRoot,
30 + reportGlobalError,
31 + resolveField,
32 + resolveFile,
33 } from 'react-server/src/ReactFlightReplyServer';
34
35 import {
@@ -183,10 +188,56 @@ function decodeReply<T>(
188 return root;
189 }
190
191 +function decodeReplyFromAsyncIterable<T>(
192 + iterable: AsyncIterable<[string, string | File]>,
193 + webpackMap: ServerManifest,
194 + options?: {temporaryReferences?: TemporaryReferenceSet},
195 +): Thenable<T> {
196 + const iterator: AsyncIterator<[string, string | File]> =
197 + iterable[ASYNC_ITERATOR]();
198 +
199 + const response = createResponse(
200 + webpackMap,
201 + '',
202 + options ? options.temporaryReferences : undefined,
203 + );
204 +
205 + function progress(
206 + entry:
207 + | {done: false, +value: [string, string | File], ...}
208 + | {done: true, +value: void, ...},
209 + ) {
210 + if (entry.done) {
211 + close(response);
212 + } else {
213 + const [name, value] = entry.value;
214 + if (typeof value === 'string') {
215 + resolveField(response, name, value);
216 + } else {
217 + resolveFile(response, name, value);
218 + }
219 + iterator.next().then(progress, error);
220 + }
221 + }
222 + function error(reason: Error) {
223 + reportGlobalError(response, reason);
224 + if (typeof (iterator: any).throw === 'function') {
225 + // The iterator protocol doesn't necessarily include this but a generator do.
226 + // $FlowFixMe should be able to pass mixed
227 + iterator.throw(reason).then(error, error);
228 + }
229 + }
230 +
231 + iterator.next().then(progress, error);
232 +
233 + return getRoot(response);
234 +}
235 +
236 export {
237 renderToReadableStream,
238 prerender,
239 decodeReply,
240 + decodeReplyFromAsyncIterable,
241 decodeAction,
242 decodeFormState,
243 };
packages/react-server-dom-webpack/src/server/react-flight-dom-server.edge.js
+1
@@ -11,6 +11,7 @@ export {
11 renderToReadableStream,
12 prerender as unstable_prerender,
13 decodeReply,
14 + decodeReplyFromAsyncIterable,
15 decodeAction,
16 decodeFormState,
17 registerServerReference,