@samitouri / QOS-React / commits / ec9400dc41

[Flight Reply] Encode ReadableStream and AsyncIterables (#28893)

Same as #28847 but in the other direction. Like other promises, this doesn't actually stream in the outgoing direction. It buffers until the stream is done. This is mainly due to our protocol remains compatible with Safari's lack of outgoing streams until recently. However, the stream chunks are encoded as separate fields and so does support the busboy streaming on the receiving side.

Sebastian Markbåge committed May 3, 2024 at 23:23 UTC ec9400dc41715bb6ff0392d6320c33627fa7e2ba
4 files changed +722 -14
packages/react-client/src/ReactFlightReplyClient.js
+141 -5
@@ -20,6 +20,7 @@ import type {TemporaryReferenceSet} from './ReactFlightTemporaryReferences';
20 import {
21 enableRenderableContext,
22 enableBinaryFlight,
23 + enableFlightReadableStream,
24 } from 'shared/ReactFeatureFlags';
25
26 import {
@@ -28,6 +29,7 @@ import {
29 REACT_CONTEXT_TYPE,
30 REACT_PROVIDER_TYPE,
31 getIteratorFn,
32 + ASYNC_ITERATOR,
33 } from 'shared/ReactSymbols';
34
35 import {
@@ -206,6 +208,123 @@ export function processReply(
208 return '$' + tag + blobId.toString(16);
209 }
210
211 + function serializeReadableStream(stream: ReadableStream): string {
212 + if (formData === null) {
213 + // Upgrade to use FormData to allow us to stream this value.
214 + formData = new FormData();
215 + }
216 + const data = formData;
217 +
218 + pendingParts++;
219 + const streamId = nextPartId++;
220 +
221 + // Detect if this is a BYOB stream. BYOB streams should be able to be read as bytes on the
222 + // receiving side. It also implies that different chunks can be split up or merged as opposed
223 + // to a readable stream that happens to have Uint8Array as the type which might expect it to be
224 + // received in the same slices.
225 + // $FlowFixMe: This is a Node.js extension.
226 + let supportsBYOB: void | boolean = stream.supportsBYOB;
227 + if (supportsBYOB === undefined) {
228 + try {
229 + // $FlowFixMe[extra-arg]: This argument is accepted.
230 + stream.getReader({mode: 'byob'}).releaseLock();
231 + supportsBYOB = true;
232 + } catch (x) {
233 + supportsBYOB = false;
234 + }
235 + }
236 +
237 + const reader = stream.getReader();
238 +
239 + function progress(entry: {done: boolean, value: ReactServerValue, ...}) {
240 + if (entry.done) {
241 + // eslint-disable-next-line react-internal/safe-string-coercion
242 + data.append(formFieldPrefix + streamId, 'C'); // Close signal
243 + pendingParts--;
244 + if (pendingParts === 0) {
245 + resolve(data);
246 + }
247 + } else {
248 + try {
249 + // $FlowFixMe[incompatible-type]: While plain JSON can return undefined we never do here.
250 + const partJSON: string = JSON.stringify(entry.value, resolveToJSON);
251 + // eslint-disable-next-line react-internal/safe-string-coercion
252 + data.append(formFieldPrefix + streamId, partJSON);
253 + reader.read().then(progress, reject);
254 + } catch (x) {
255 + reject(x);
256 + }
257 + }
258 + }
259 + reader.read().then(progress, reject);
260 +
261 + return '$' + (supportsBYOB ? 'r' : 'R') + streamId.toString(16);
262 + }
263 +
264 + function serializeAsyncIterable(
265 + iterable: $AsyncIterable<ReactServerValue, ReactServerValue, void>,
266 + iterator: $AsyncIterator<ReactServerValue, ReactServerValue, void>,
267 + ): string {
268 + if (formData === null) {
269 + // Upgrade to use FormData to allow us to stream this value.
270 + formData = new FormData();
271 + }
272 + const data = formData;
273 +
274 + pendingParts++;
275 + const streamId = nextPartId++;
276 +
277 + // Generators/Iterators are Iterables but they're also their own iterator
278 + // functions. If that's the case, we treat them as single-shot. Otherwise,
279 + // we assume that this iterable might be a multi-shot and allow it to be
280 + // iterated more than once on the receiving server.
281 + const isIterator = iterable === iterator;
282 +
283 + // There's a race condition between when the stream is aborted and when the promise
284 + // resolves so we track whether we already aborted it to avoid writing twice.
285 + function progress(
286 + entry:
287 + | {done: false, +value: ReactServerValue, ...}
288 + | {done: true, +value: ReactServerValue, ...},
289 + ) {
290 + if (entry.done) {
291 + if (entry.value === undefined) {
292 + // eslint-disable-next-line react-internal/safe-string-coercion
293 + data.append(formFieldPrefix + streamId, 'C'); // Close signal
294 + } else {
295 + // Unlike streams, the last value may not be undefined. If it's not
296 + // we outline it and encode a reference to it in the closing instruction.
297 + try {
298 + // $FlowFixMe[incompatible-type]: While plain JSON can return undefined we never do here.
299 + const partJSON: string = JSON.stringify(entry.value, resolveToJSON);
300 + data.append(formFieldPrefix + streamId, 'C' + partJSON); // Close signal
301 + } catch (x) {
302 + reject(x);
303 + return;
304 + }
305 + }
306 + pendingParts--;
307 + if (pendingParts === 0) {
308 + resolve(data);
309 + }
310 + } else {
311 + try {
312 + // $FlowFixMe[incompatible-type]: While plain JSON can return undefined we never do here.
313 + const partJSON: string = JSON.stringify(entry.value, resolveToJSON);
314 + // eslint-disable-next-line react-internal/safe-string-coercion
315 + data.append(formFieldPrefix + streamId, partJSON);
316 + iterator.next().then(progress, reject);
317 + } catch (x) {
318 + reject(x);
319 + return;
320 + }
321 + }
322 + }
323 +
324 + iterator.next().then(progress, reject);
325 + return '$' + (isIterator ? 'x' : 'X') + streamId.toString(16);
326 + }
327 +
328 function resolveToJSON(
329 this:
330 | {+[key: string | number]: ReactServerValue}
@@ -349,11 +468,9 @@ export function processReply(
468 reject(reason);
469 }
470 },
352 - reason => {
353 - // In the future we could consider serializing this as an error
354 - // that throws on the server instead.
355 - reject(reason);
356 - },
471 + // In the future we could consider serializing this as an error
472 + // that throws on the server instead.
473 + reject,
474 );
475 return serializePromiseID(promiseId);
476 }
@@ -486,6 +603,25 @@ export function processReply(
603 return Array.from((iterator: any));
604 }
605
606 + if (enableFlightReadableStream) {
607 + // TODO: ReadableStream is not available in old Node. Remove the typeof check later.
608 + if (
609 + typeof ReadableStream === 'function' &&
610 + value instanceof ReadableStream
611 + ) {
612 + return serializeReadableStream(value);
613 + }
614 + const getAsyncIterator: void | (() => $AsyncIterator<any, any, any>) =
615 + (value: any)[ASYNC_ITERATOR];
616 + if (typeof getAsyncIterator === 'function') {
617 + // We treat AsyncIterables as a Fragment and as such we might need to key them.
618 + return serializeAsyncIterable(
619 + (value: any),
620 + getAsyncIterator.call((value: any)),
621 + );
622 + }
623 + }
624 +
625 // Verify that this is a simple plain object.
626 const proto = getPrototypeOf(value);
627 if (
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMReply-test.js
+161
@@ -376,4 +376,165 @@ describe('ReactFlightDOMReply', () => {
376 // This should've been the same reference that we already saw.
377 expect(response.children).toBe(children);
378 });
379 +
380 + // @gate enableFlightReadableStream
381 + it('should supports streaming ReadableStream with objects', async () => {
382 + let controller1;
383 + let controller2;
384 + const s1 = new ReadableStream({
385 + start(c) {
386 + controller1 = c;
387 + },
388 + });
389 + const s2 = new ReadableStream({
390 + start(c) {
391 + controller2 = c;
392 + },
393 + });
394 +
395 + const promise = ReactServerDOMClient.encodeReply({s1, s2});
396 +
397 + controller1.enqueue({hello: 'world'});
398 + controller2.enqueue({hi: 'there'});
399 +
400 + controller1.enqueue('text1');
401 + controller2.enqueue('text2');
402 +
403 + controller1.close();
404 + controller2.close();
405 +
406 + const body = await promise;
407 +
408 + const result = await ReactServerDOMServer.decodeReply(
409 + body,
410 + webpackServerMap,
411 + );
412 + const reader1 = result.s1.getReader();
413 + const reader2 = result.s2.getReader();
414 +
415 + expect(await reader1.read()).toEqual({
416 + value: {hello: 'world'},
417 + done: false,
418 + });
419 + expect(await reader2.read()).toEqual({
420 + value: {hi: 'there'},
421 + done: false,
422 + });
423 +
424 + expect(await reader1.read()).toEqual({
425 + value: 'text1',
426 + done: false,
427 + });
428 + expect(await reader1.read()).toEqual({
429 + value: undefined,
430 + done: true,
431 + });
432 + expect(await reader2.read()).toEqual({
433 + value: 'text2',
434 + done: false,
435 + });
436 + expect(await reader2.read()).toEqual({
437 + value: undefined,
438 + done: true,
439 + });
440 + });
441 +
442 + // @gate enableFlightReadableStream
443 + it('should supports streaming AsyncIterables with objects', async () => {
444 + let resolve;
445 + const wait = new Promise(r => (resolve = r));
446 + const multiShotIterable = {
447 + async *[Symbol.asyncIterator]() {
448 + const next = yield {hello: 'A'};
449 + expect(next).toBe(undefined);
450 + await wait;
451 + yield {hi: 'B'};
452 + return 'C';
453 + },
454 + };
455 + const singleShotIterator = (async function* () {
456 + const next = yield {hello: 'D'};
457 + expect(next).toBe(undefined);
458 + await wait;
459 + yield {hi: 'E'};
460 + return 'F';
461 + })();
462 +
463 + await resolve();
464 +
465 + const body = await ReactServerDOMClient.encodeReply({
466 + multiShotIterable,
467 + singleShotIterator,
468 + });
469 + const result = await ReactServerDOMServer.decodeReply(
470 + body,
471 + webpackServerMap,
472 + );
473 +
474 + const iterator1 = result.multiShotIterable[Symbol.asyncIterator]();
475 + const iterator2 = result.singleShotIterator[Symbol.asyncIterator]();
476 +
477 + expect(iterator1).not.toBe(result.multiShotIterable);
478 + expect(iterator2).toBe(result.singleShotIterator);
479 +
480 + expect(await iterator1.next()).toEqual({
481 + value: {hello: 'A'},
482 + done: false,
483 + });
484 + expect(await iterator2.next()).toEqual({
485 + value: {hello: 'D'},
486 + done: false,
487 + });
488 +
489 + expect(await iterator1.next()).toEqual({
490 + value: {hi: 'B'},
491 + done: false,
492 + });
493 + expect(await iterator2.next()).toEqual({
494 + value: {hi: 'E'},
495 + done: false,
496 + });
497 + expect(await iterator1.next()).toEqual({
498 + value: 'C', // Return value
499 + done: true,
500 + });
501 + expect(await iterator1.next()).toEqual({
502 + value: undefined,
503 + done: true,
504 + });
505 +
506 + expect(await iterator2.next()).toEqual({
507 + value: 'F', // Return value
508 + done: true,
509 + });
510 +
511 + // Multi-shot iterables should be able to do the same thing again
512 + const iterator3 = result.multiShotIterable[Symbol.asyncIterator]();
513 +
514 + expect(iterator3).not.toBe(iterator1);
515 +
516 + // We should be able to iterate over the iterable again and it should be
517 + // synchronously available using instrumented promises so that React can
518 + // rerender it synchronously.
519 + expect(iterator3.next().value).toEqual({
520 + value: {hello: 'A'},
521 + done: false,
522 + });
523 + expect(iterator3.next().value).toEqual({
524 + value: {hi: 'B'},
525 + done: false,
526 + });
527 + expect(iterator3.next().value).toEqual({
528 + value: 'C', // Return value
529 + done: true,
530 + });
531 + expect(iterator3.next().value).toEqual({
532 + value: undefined,
533 + done: true,
534 + });
535 +
536 + expect(() => iterator3.next('this is not allowed')).toThrow(
537 + 'Values cannot be passed to next() of AsyncIterables passed to Client Components.',
538 + );
539 + });
540 });
packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMReplyEdge-test.js
+101
@@ -132,4 +132,105 @@ describe('ReactFlightDOMReplyEdge', () => {
132 expect(resultBlob.size).toBe(bytes.length * 2);
133 expect(await resultBlob.arrayBuffer()).toEqual(await blob.arrayBuffer());
134 });
135 +
136 + // @gate enableFlightReadableStream && enableBinaryFlight
137 + it('should supports ReadableStreams with typed arrays', async () => {
138 + const buffer = new Uint8Array([
139 + 123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,
140 + ]).buffer;
141 + const buffers = [
142 + buffer,
143 + new Int8Array(buffer, 1),
144 + new Uint8Array(buffer, 2),
145 + new Uint8ClampedArray(buffer, 2),
146 + new Int16Array(buffer, 2),
147 + new Uint16Array(buffer, 2),
148 + new Int32Array(buffer, 4),
149 + new Uint32Array(buffer, 4),
150 + new Float32Array(buffer, 4),
151 + new Float64Array(buffer, 0),
152 + new BigInt64Array(buffer, 0),
153 + new BigUint64Array(buffer, 0),
154 + new DataView(buffer, 3),
155 + ];
156 +
157 + // This is not a binary stream, it's a stream that contain binary chunks.
158 + const s = new ReadableStream({
159 + start(c) {
160 + for (let i = 0; i < buffers.length; i++) {
161 + c.enqueue(buffers[i]);
162 + }
163 + c.close();
164 + },
165 + });
166 +
167 + const body = await ReactServerDOMClient.encodeReply(s);
168 + const result = await ReactServerDOMServer.decodeReply(
169 + body,
170 + webpackServerMap,
171 + );
172 +
173 + const streamedBuffers = [];
174 + const reader = result.getReader();
175 + let entry;
176 + while (!(entry = await reader.read()).done) {
177 + streamedBuffers.push(entry.value);
178 + }
179 +
180 + expect(streamedBuffers).toEqual(buffers);
181 + });
182 +
183 + // @gate enableFlightReadableStream && enableBinaryFlight
184 + it('should support BYOB binary ReadableStreams', async () => {
185 + const buffer = new Uint8Array([
186 + 123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,
187 + ]).buffer;
188 + const buffers = [
189 + new Int8Array(buffer, 1),
190 + new Uint8Array(buffer, 2),
191 + new Uint8ClampedArray(buffer, 2),
192 + new Int16Array(buffer, 2),
193 + new Uint16Array(buffer, 2),
194 + new Int32Array(buffer, 4),
195 + new Uint32Array(buffer, 4),
196 + new Float32Array(buffer, 4),
197 + new Float64Array(buffer, 0),
198 + new BigInt64Array(buffer, 0),
199 + new BigUint64Array(buffer, 0),
200 + new DataView(buffer, 3),
201 + ];
202 +
203 + // This a binary stream where each chunk ends up as Uint8Array.
204 + const s = new ReadableStream({
205 + type: 'bytes',
206 + start(c) {
207 + for (let i = 0; i < buffers.length; i++) {
208 + c.enqueue(buffers[i]);
209 + }
210 + c.close();
211 + },
212 + });
213 +
214 + const body = await ReactServerDOMClient.encodeReply(s);
215 + const result = await ReactServerDOMServer.decodeReply(
216 + body,
217 + webpackServerMap,
218 + );
219 +
220 + const streamedBuffers = [];
221 + const reader = result.getReader({mode: 'byob'});
222 + let entry;
223 + while (!(entry = await reader.read(new Uint8Array(10))).done) {
224 + expect(entry.value instanceof Uint8Array).toBe(true);
225 + streamedBuffers.push(entry.value);
226 + }
227 +
228 + // The streamed buffers might be in different chunks and in Uint8Array form but
229 + // the concatenated bytes should be the same.
230 + expect(streamedBuffers.flatMap(t => Array.from(t))).toEqual(
231 + buffers.flatMap(c =>
232 + Array.from(new Uint8Array(c.buffer, c.byteOffset, c.byteLength)),
233 + ),
234 + );
235 + });
236 });
packages/react-server/src/ReactFlightReplyServer.js
+319 -9
@@ -25,7 +25,17 @@ import {
25 } from 'react-client/src/ReactFlightClientConfig';
26
27 import {createTemporaryReference} from './ReactFlightServerTemporaryReferences';
28 -import {enableBinaryFlight} from 'shared/ReactFeatureFlags';
28 +import {
29 + enableBinaryFlight,
30 + enableFlightReadableStream,
31 +} from 'shared/ReactFeatureFlags';
32 +import {ASYNC_ITERATOR} from 'shared/ReactSymbols';
33 +
34 +interface FlightStreamController {
35 + enqueueModel(json: string): void;
36 + close(json: string): void;
37 + error(error: Error): void;
38 +}
39
40 export type JSONValue =
41 | number
@@ -46,35 +56,44 @@ type PendingChunk<T> = {
56 value: null | Array<(T) => mixed>,
57 reason: null | Array<(mixed) => mixed>,
58 _response: Response,
49 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,
59 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,
60 };
61 type BlockedChunk<T> = {
62 status: 'blocked',
63 value: null | Array<(T) => mixed>,
64 reason: null | Array<(mixed) => mixed>,
65 _response: Response,
56 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,
66 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,
67 };
68 type ResolvedModelChunk<T> = {
69 status: 'resolved_model',
70 value: string,
71 reason: null,
72 _response: Response,
63 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,
73 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,
74 };
75 type InitializedChunk<T> = {
76 status: 'fulfilled',
77 value: T,
78 reason: null,
79 _response: Response,
70 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,
80 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,
81 +};
82 +type InitializedStreamChunk<
83 + T: ReadableStream | $AsyncIterable<any, any, void>,
84 +> = {
85 + status: 'fulfilled',
86 + value: T,
87 + reason: FlightStreamController,
88 + _response: Response,
89 + then(resolve: (ReadableStream) => mixed, reject?: (mixed) => mixed): void,
90 };
91 type ErroredChunk<T> = {
92 status: 'rejected',
93 value: null,
94 reason: mixed,
95 _response: Response,
77 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,
96 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,
97 };
98 type SomeChunk<T> =
99 | PendingChunk<T>
@@ -181,7 +200,14 @@ function wakeChunkIfInitialized<T>(
200
201 function triggerErrorOnChunk<T>(chunk: SomeChunk<T>, error: mixed): void {
202 if (chunk.status !== PENDING && chunk.status !== BLOCKED) {
184 - // We already resolved. We didn't expect to see this.
203 + if (enableFlightReadableStream) {
204 + // If we get more data to an already resolved ID, we assume that it's
205 + // a stream chunk since any other row shouldn't have more than one entry.
206 + const streamChunk: InitializedStreamChunk<any> = (chunk: any);
207 + const controller = streamChunk.reason;
208 + // $FlowFixMe[incompatible-call]: The error method should accept mixed.
209 + controller.error(error);
210 + }
211 return;
212 }
213 const listeners = chunk.reason;
@@ -203,7 +229,17 @@ function createResolvedModelChunk<T>(
229
230 function resolveModelChunk<T>(chunk: SomeChunk<T>, value: string): void {
231 if (chunk.status !== PENDING) {
206 - // We already resolved. We didn't expect to see this.
232 + if (enableFlightReadableStream) {
233 + // If we get more data to an already resolved ID, we assume that it's
234 + // a stream chunk since any other row shouldn't have more than one entry.
235 + const streamChunk: InitializedStreamChunk<any> = (chunk: any);
236 + const controller = streamChunk.reason;
237 + if (value[0] === 'C') {
238 + controller.close(value === 'C' ? '"$undefined"' : value.slice(1));
239 + } else {
240 + controller.enqueueModel(value);
241 + }
242 + }
243 return;
244 }
245 const resolveListeners = chunk.value;
@@ -221,6 +257,42 @@ function resolveModelChunk<T>(chunk: SomeChunk<T>, value: string): void {
257 }
258 }
259
260 +function createInitializedStreamChunk<
261 + T: ReadableStream | $AsyncIterable<any, any, void>,
262 +>(
263 + response: Response,
264 + value: T,
265 + controller: FlightStreamController,
266 +): InitializedChunk<T> {
267 + // We use the reason field to stash the controller since we already have that
268 + // field. It's a bit of a hack but efficient.
269 + // $FlowFixMe[invalid-constructor] Flow doesn't support functions as constructors
270 + return new Chunk(INITIALIZED, value, controller, response);
271 +}
272 +
273 +function createResolvedIteratorResultChunk<T>(
274 + response: Response,
275 + value: string,
276 + done: boolean,
277 +): ResolvedModelChunk<IteratorResult<T, T>> {
278 + // To reuse code as much code as possible we add the wrapper element as part of the JSON.
279 + const iteratorResultJSON =
280 + (done ? '{"done":true,"value":' : '{"done":false,"value":') + value + '}';
281 + // $FlowFixMe[invalid-constructor] Flow doesn't support functions as constructors
282 + return new Chunk(RESOLVED_MODEL, iteratorResultJSON, null, response);
283 +}
284 +
285 +function resolveIteratorResultChunk<T>(
286 + chunk: SomeChunk<IteratorResult<T, T>>,
287 + value: string,
288 + done: boolean,
289 +): void {
290 + // To reuse code as much code as possible we add the wrapper element as part of the JSON.
291 + const iteratorResultJSON =
292 + (done ? '{"done":true,"value":' : '{"done":false,"value":') + value + '}';
293 + resolveModelChunk(chunk, iteratorResultJSON);
294 +}
295 +
296 function bindArgs(fn: any, args: any) {
297 return fn.bind.apply(fn, [null].concat(args));
298 }
@@ -342,11 +414,18 @@ function createModelResolver<T>(
414 } else {
415 blocked = initializingChunkBlockedModel = {
416 deps: 1,
345 - value: null,
417 + value: (null: any),
418 };
419 }
420 return value => {
421 parentObject[key] = value;
422 +
423 + // If this is the root object for a model reference, where `blocked.value`
424 + // is a stale `null`, the resolved value can be used directly.
425 + if (key === '' && blocked.value === null) {
426 + blocked.value = parentObject[key];
427 + }
428 +
429 blocked.deps--;
430 if (blocked.deps === 0) {
431 if (chunk.status !== BLOCKED) {
@@ -411,6 +490,221 @@ function parseTypedArray(
490 return null;
491 }
492
493 +function resolveStream<T: ReadableStream | $AsyncIterable<any, any, void>>(
494 + response: Response,
495 + id: number,
496 + stream: T,
497 + controller: FlightStreamController,
498 +): void {
499 + const chunks = response._chunks;
500 + const chunk = createInitializedStreamChunk(response, stream, controller);
501 + chunks.set(id, chunk);
502 +
503 + const prefix = response._prefix;
504 + const key = prefix + id;
505 + const existingEntries = response._formData.getAll(key);
506 + for (let i = 0; i < existingEntries.length; i++) {
507 + // We assume that this is a string entry for now.
508 + const value: string = (existingEntries[i]: any);
509 + if (value[0] === 'C') {
510 + controller.close(value === 'C' ? '"$undefined"' : value.slice(1));
511 + } else {
512 + controller.enqueueModel(value);
513 + }
514 + }
515 +}
516 +
517 +function parseReadableStream<T>(
518 + response: Response,
519 + reference: string,
520 + type: void | 'bytes',
521 + parentObject: Object,
522 + parentKey: string,
523 +): ReadableStream {
524 + const id = parseInt(reference.slice(2), 16);
525 +
526 + let controller: ReadableStreamController = (null: any);
527 + const stream = new ReadableStream({
528 + type: type,
529 + start(c) {
530 + controller = c;
531 + },
532 + });
533 + let previousBlockedChunk: SomeChunk<T> | null = null;
534 + const flightController = {
535 + enqueueModel(json: string): void {
536 + if (previousBlockedChunk === null) {
537 + // If we're not blocked on any other chunks, we can try to eagerly initialize
538 + // this as a fast-path to avoid awaiting them.
539 + const chunk: ResolvedModelChunk<T> = createResolvedModelChunk(
540 + response,
541 + json,
542 + );
543 + initializeModelChunk(chunk);
544 + const initializedChunk: SomeChunk<T> = chunk;
545 + if (initializedChunk.status === INITIALIZED) {
546 + controller.enqueue(initializedChunk.value);
547 + } else {
548 + chunk.then(
549 + v => controller.enqueue(v),
550 + e => controller.error((e: any)),
551 + );
552 + previousBlockedChunk = chunk;
553 + }
554 + } else {
555 + // We're still waiting on a previous chunk so we can't enqueue quite yet.
556 + const blockedChunk = previousBlockedChunk;
557 + const chunk: SomeChunk<T> = createPendingChunk(response);
558 + chunk.then(
559 + v => controller.enqueue(v),
560 + e => controller.error((e: any)),
561 + );
562 + previousBlockedChunk = chunk;
563 + blockedChunk.then(function () {
564 + if (previousBlockedChunk === chunk) {
565 + // We were still the last chunk so we can now clear the queue and return
566 + // to synchronous emitting.
567 + previousBlockedChunk = null;
568 + }
569 + resolveModelChunk(chunk, json);
570 + });
571 + }
572 + },
573 + close(json: string): void {
574 + if (previousBlockedChunk === null) {
575 + controller.close();
576 + } else {
577 + const blockedChunk = previousBlockedChunk;
578 + // We shouldn't get any more enqueues after this so we can set it back to null.
579 + previousBlockedChunk = null;
580 + blockedChunk.then(() => controller.close());
581 + }
582 + },
583 + error(error: mixed): void {
584 + if (previousBlockedChunk === null) {
585 + // $FlowFixMe[incompatible-call]
586 + controller.error(error);
587 + } else {
588 + const blockedChunk = previousBlockedChunk;
589 + // We shouldn't get any more enqueues after this so we can set it back to null.
590 + previousBlockedChunk = null;
591 + blockedChunk.then(() => controller.error((error: any)));
592 + }
593 + },
594 + };
595 + resolveStream(response, id, stream, flightController);
596 + return stream;
597 +}
598 +
599 +function asyncIterator(this: $AsyncIterator<any, any, void>) {
600 + // Self referencing iterator.
601 + return this;
602 +}
603 +
604 +function createIterator<T>(
605 + next: (arg: void) => SomeChunk<IteratorResult<T, T>>,
606 +): $AsyncIterator<T, T, void> {
607 + const iterator: any = {
608 + next: next,
609 + // TODO: Add return/throw as options for aborting.
610 + };
611 + // TODO: The iterator could inherit the AsyncIterator prototype which is not exposed as
612 + // a global but exists as a prototype of an AsyncGenerator. However, it's not needed
613 + // to satisfy the iterable protocol.
614 + (iterator: any)[ASYNC_ITERATOR] = asyncIterator;
615 + return iterator;
616 +}
617 +
618 +function parseAsyncIterable<T>(
619 + response: Response,
620 + reference: string,
621 + iterator: boolean,
622 + parentObject: Object,
623 + parentKey: string,
624 +): $AsyncIterable<T, T, void> | $AsyncIterator<T, T, void> {
625 + const id = parseInt(reference.slice(2), 16);
626 +
627 + const buffer: Array<SomeChunk<IteratorResult<T, T>>> = [];
628 + let closed = false;
629 + let nextWriteIndex = 0;
630 + const flightController = {
631 + enqueueModel(value: string): void {
632 + if (nextWriteIndex === buffer.length) {
633 + buffer[nextWriteIndex] = createResolvedIteratorResultChunk(
634 + response,
635 + value,
636 + false,
637 + );
638 + } else {
639 + resolveIteratorResultChunk(buffer[nextWriteIndex], value, false);
640 + }
641 + nextWriteIndex++;
642 + },
643 + close(value: string): void {
644 + closed = true;
645 + if (nextWriteIndex === buffer.length) {
646 + buffer[nextWriteIndex] = createResolvedIteratorResultChunk(
647 + response,
648 + value,
649 + true,
650 + );
651 + } else {
652 + resolveIteratorResultChunk(buffer[nextWriteIndex], value, true);
653 + }
654 + nextWriteIndex++;
655 + while (nextWriteIndex < buffer.length) {
656 + // In generators, any extra reads from the iterator have the value undefined.
657 + resolveIteratorResultChunk(
658 + buffer[nextWriteIndex++],
659 + '"$undefined"',
660 + true,
661 + );
662 + }
663 + },
664 + error(error: Error): void {
665 + closed = true;
666 + if (nextWriteIndex === buffer.length) {
667 + buffer[nextWriteIndex] =
668 + createPendingChunk<IteratorResult<T, T>>(response);
669 + }
670 + while (nextWriteIndex < buffer.length) {
671 + triggerErrorOnChunk(buffer[nextWriteIndex++], error);
672 + }
673 + },
674 + };
675 + const iterable: $AsyncIterable<T, T, void> = {
676 + [ASYNC_ITERATOR](): $AsyncIterator<T, T, void> {
677 + let nextReadIndex = 0;
678 + return createIterator(arg => {
679 + if (arg !== undefined) {
680 + throw new Error(
681 + 'Values cannot be passed to next() of AsyncIterables passed to Client Components.',
682 + );
683 + }
684 + if (nextReadIndex === buffer.length) {
685 + if (closed) {
686 + // $FlowFixMe[invalid-constructor] Flow doesn't support functions as constructors
687 + return new Chunk(
688 + INITIALIZED,
689 + {done: true, value: undefined},
690 + null,
691 + response,
692 + );
693 + }
694 + buffer[nextReadIndex] =
695 + createPendingChunk<IteratorResult<T, T>>(response);
696 + }
697 + return buffer[nextReadIndex++];
698 + });
699 + },
700 + };
701 + // TODO: If it's a single shot iterator we can optimize memory by cleaning up the buffer after
702 + // reading through the end, but currently we favor code size over this optimization.
703 + const stream = iterator ? iterable[ASYNC_ITERATOR]() : iterable;
704 + resolveStream(response, id, stream, flightController);
705 + return stream;
706 +}
707 +
708 function parseModelString(
709 response: Response,
710 obj: Object,
@@ -560,6 +854,22 @@ function parseModelString(
854 }
855 }
856 }
857 + if (enableFlightReadableStream) {
858 + switch (value[1]) {
859 + case 'R': {
860 + return parseReadableStream(response, value, undefined, obj, key);
861 + }
862 + case 'r': {
863 + return parseReadableStream(response, value, 'bytes', obj, key);
864 + }
865 + case 'X': {
866 + return parseAsyncIterable(response, value, false, obj, key);
867 + }
868 + case 'x': {
869 + return parseAsyncIterable(response, value, true, obj, key);
870 + }
871 + }
872 + }
873
874 // We assume that anything else is a reference ID.
875 const id = parseInt(value.slice(1), 16);