@samitouri / QOS-React / commits / 7909d8eabb

[Flight] Encode ReadableStream and AsyncIterables (#28847)

This adds support in Flight for serializing four kinds of streams: - `ReadableStream` with objects as a model. This is a single shot iterator so you can read it only once. It can contain any value including Server Components. Chunks are encoded as is so if you send in 10 typed arrays, you get the same typed arrays out on the other side. - Binary `ReadableStream` with `type: 'bytes'` option. This supports the BYOB protocol. In this mode, the receiving side just gets `Uint8Array`s and they can be split across any single byte boundary into arbitrary chunks. - `AsyncIterable` where the `AsyncIterator` function is different than the `AsyncIterable` itself. In this case we assume that this might be a multi-shot iterable and so we buffer its value and you can iterate it multiple times on the other side. We support the `return` value as a value in the single completion slot, but you can't pass values in `next()`. If you want single-shot, return the AsyncIterator instead. - `AsyncIterator`. These gets serialized as a single-shot as it's just an iterator. `AsyncIterable`/`AsyncIterator` yield Promises that are instrumented with our `.status`/`.value` convention so that they can be synchronously looped over if available. They are also lazily parsed upon read. We can't do this with `ReadableStream` because we use the native implementation of `ReadableStream` which owns the promises. The format is a leading row that indicates which type of stream it is. Then a new row with the same ID is emitted for every chunk. Followed by either an error or close row. `AsyncIterable`s can also be returned as children of Server Components and then they're conceptually the same as fragment arrays/iterables. They can't actually be used as children in Fizz/Fiber but there's a separate plan for that. Only `AsyncIterable` not `AsyncIterator` will be valid as children - just like sync `Iterable` is already supported but single-shot `Iterator` is not. Notably, neither of these streams represent updates over time to a value. They represent multiple values in a list. When the server stream is aborted we also close the underlying stream. However, closing a stream on the client, doesn't close the underlying stream. A couple of possible follow ups I'm not planning on doing right now: - [ ] Free memory by releasing the buffer if an Iterator has been exhausted. Single shots could be optimized further to release individual items as you go. - [ ] We could clean up the underlying stream if the only pending data that's still flowing is from streams and all the streams have cleaned up. It's not very reliable though. It's better to do cancellation for the whole stream - e.g. at the framework level. - [ ] Implement smarter Binary Stream chunk handling. Currently we wait until we've received a whole row for binary chunks and copy them into consecutive memory. We need this to preserve semantics when passing typed arrays. However, for binary streams we don't need that. We can just send whatever pieces we have so far.

Sebastian Markbåge committed Apr 16, 2024 at 12:20 UTC 7909d8eabb7a702618f51e16a351df41aa8da88e

18 files changed +1937 -110

.eslintrc.js

@@ -499,7 +499,12 @@ module.exports = {

499 DOMHighResTimeStamp: 'readonly',

500 EventListener: 'readonly',

501 Iterable: 'readonly',

502 + AsyncIterable: 'readonly',

503 + $AsyncIterable: 'readonly',

504 + $AsyncIterator: 'readonly',

505 Iterator: 'readonly',

506 + AsyncIterator: 'readonly',

507 + IteratorResult: 'readonly',

508 JSONValue: 'readonly',

509 JSResourceReference: 'readonly',

510 MouseEventHandler: 'readonly',

@@ -520,6 +525,7 @@ module.exports = {

525 React$Portal: 'readonly',

526 React$Ref: 'readonly',

527 ReadableStreamController: 'readonly',

528 + ReadableStreamReader: 'readonly',

529 RequestInfo: 'readonly',

530 RequestOptions: 'readonly',

531 StoreAsGlobal: 'readonly',

packages/react-client/src/ReactFlightClient.js

+414 -16

@@ -42,6 +42,7 @@ import {

42 enableBinaryFlight,

43 enablePostpone,

44 enableRefAsProp,

45 + enableFlightReadableStream,

46 } from 'shared/ReactFeatureFlags';

48 import {

@@ -68,6 +69,13 @@ import {

70 export type {CallServerCallback, EncodeFormActionCallback};

72 +interface FlightStreamController {

73 + enqueueValue(value: any): void;

74 + enqueueModel(json: UninitializedModel): void;

75 + close(json: UninitializedModel): void;

76 + error(error: Error): void;

77 +}

78 +

79 type UninitializedModel = string;

81 export type JSONValue =

@@ -100,7 +108,7 @@ type PendingChunk<T> = {

108 reason: null | Array<(mixed) => mixed>,

109 _response: Response,

110 _debugInfo?: null | ReactDebugInfo,

103 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,

111 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,

112 };

113 type BlockedChunk<T> = {

114 status: 'blocked',

@@ -108,7 +116,7 @@ type BlockedChunk<T> = {

116 reason: null | Array<(mixed) => mixed>,

117 _response: Response,

118 _debugInfo?: null | ReactDebugInfo,

111 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,

119 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,

120 };

121 type CyclicChunk<T> = {

122 status: 'cyclic',

@@ -116,7 +124,7 @@ type CyclicChunk<T> = {

124 reason: null | Array<(mixed) => mixed>,

125 _response: Response,

126 _debugInfo?: null | ReactDebugInfo,

119 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,

127 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,

128 };

129 type ResolvedModelChunk<T> = {

130 status: 'resolved_model',

@@ -124,7 +132,7 @@ type ResolvedModelChunk<T> = {

132 reason: null,

133 _response: Response,

134 _debugInfo?: null | ReactDebugInfo,

127 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,

135 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,

136 };

137 type ResolvedModuleChunk<T> = {

138 status: 'resolved_module',

@@ -132,15 +140,24 @@ type ResolvedModuleChunk<T> = {

140 reason: null,

141 _response: Response,

142 _debugInfo?: null | ReactDebugInfo,

135 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,

143 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,

144 };

145 type InitializedChunk<T> = {

146 status: 'fulfilled',

147 value: T,

140 - reason: null,

148 + reason: null | FlightStreamController,

149 _response: Response,

150 _debugInfo?: null | ReactDebugInfo,

143 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,

151 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,

152 +};

153 +type InitializedStreamChunk<

154 + T: ReadableStream | $AsyncIterable<any, any, void>,

155 +> = {

156 + status: 'fulfilled',

157 + value: T,

158 + reason: FlightStreamController,

159 + _response: Response,

160 + then(resolve: (ReadableStream) => mixed, reject?: (mixed) => mixed): void,

161 };

162 type ErroredChunk<T> = {

163 status: 'rejected',

@@ -148,7 +165,7 @@ type ErroredChunk<T> = {

165 reason: mixed,

166 _response: Response,

167 _debugInfo?: null | ReactDebugInfo,

151 - then(resolve: (T) => mixed, reject: (mixed) => mixed): void,

168 + then(resolve: (T) => mixed, reject?: (mixed) => mixed): void,

169 };

170 type SomeChunk<T> =

171 | PendingChunk<T>

@@ -175,7 +192,7 @@ Chunk.prototype = (Object.create(Promise.prototype): any);

192 Chunk.prototype.then = function <T>(

193 this: SomeChunk<T>,

194 resolve: (value: T) => mixed,

178 - reject: (reason: mixed) => mixed,

195 + reject?: (reason: mixed) => mixed,

196 ) {

197 const chunk: SomeChunk<T> = this;

198 // If we have resolved content, we try to initialize it first which

@@ -210,7 +227,9 @@ Chunk.prototype.then = function <T>(

227 }

228 break;

229 default:

213 - reject(chunk.reason);

230 + if (reject) {

231 + reject(chunk.reason);

232 + }

233 break;

234 }

235 };

@@ -312,7 +331,14 @@ function wakeChunkIfInitialized<T>(

331

332 function triggerErrorOnChunk<T>(chunk: SomeChunk<T>, error: mixed): void {

333 if (chunk.status !== PENDING && chunk.status !== BLOCKED) {

315 - // We already resolved. We didn't expect to see this.

334 + if (enableFlightReadableStream) {

335 + // If we get more data to an already resolved ID, we assume that it's

336 + // a stream chunk since any other row shouldn't have more than one entry.

337 + const streamChunk: InitializedStreamChunk<any> = (chunk: any);

338 + const controller = streamChunk.reason;

339 + // $FlowFixMe[incompatible-call]: The error method should accept mixed.

340 + controller.error(error);

341 + }

342 return;

343 }

344 const listeners = chunk.reason;

@@ -356,12 +382,63 @@ function createInitializedBufferChunk(

382 return new Chunk(INITIALIZED, value, null, response);

383 }

384

385 +function createInitializedIteratorResultChunk<T>(

386 + response: Response,

387 + value: T,

388 + done: boolean,

389 +): InitializedChunk<IteratorResult<T, T>> {

390 + // $FlowFixMe[invalid-constructor] Flow doesn't support functions as constructors

391 + return new Chunk(INITIALIZED, {done: done, value: value}, null, response);

392 +}

393 +

394 +function createInitializedStreamChunk<

395 + T: ReadableStream | $AsyncIterable<any, any, void>,

396 +>(

397 + response: Response,

398 + value: T,

399 + controller: FlightStreamController,

400 +): InitializedChunk<T> {

401 + // We use the reason field to stash the controller since we already have that

402 + // field. It's a bit of a hack but efficient.

403 + // $FlowFixMe[invalid-constructor] Flow doesn't support functions as constructors

404 + return new Chunk(INITIALIZED, value, controller, response);

405 +}

406 +

407 +function createResolvedIteratorResultChunk<T>(

408 + response: Response,

409 + value: UninitializedModel,

410 + done: boolean,

411 +): ResolvedModelChunk<IteratorResult<T, T>> {

412 + // To reuse code as much code as possible we add the wrapper element as part of the JSON.

413 + const iteratorResultJSON =

414 + (done ? '{"done":true,"value":' : '{"done":false,"value":') + value + '}';

415 + // $FlowFixMe[invalid-constructor] Flow doesn't support functions as constructors

416 + return new Chunk(RESOLVED_MODEL, iteratorResultJSON, null, response);

417 +}

418 +

419 +function resolveIteratorResultChunk<T>(

420 + chunk: SomeChunk<IteratorResult<T, T>>,

421 + value: UninitializedModel,

422 + done: boolean,

423 +): void {

424 + // To reuse code as much code as possible we add the wrapper element as part of the JSON.

425 + const iteratorResultJSON =

426 + (done ? '{"done":true,"value":' : '{"done":false,"value":') + value + '}';

427 + resolveModelChunk(chunk, iteratorResultJSON);

428 +}

429 +

430 function resolveModelChunk<T>(

431 chunk: SomeChunk<T>,

432 value: UninitializedModel,

433 ): void {

434 if (chunk.status !== PENDING) {

364 - // We already resolved. We didn't expect to see this.

435 + if (enableFlightReadableStream) {

436 + // If we get more data to an already resolved ID, we assume that it's

437 + // a stream chunk since any other row shouldn't have more than one entry.

438 + const streamChunk: InitializedStreamChunk<any> = (chunk: any);

439 + const controller = streamChunk.reason;

440 + controller.enqueueModel(value);

441 + }

442 return;

443 }

444 const resolveListeners = chunk.value;

@@ -685,6 +762,7 @@ function getOutlinedModel<T>(

762 typeof chunkValue === 'object' &&

763 chunkValue !== null &&

764 (Array.isArray(chunkValue) ||

765 + typeof chunkValue[ASYNC_ITERATOR] === 'function' ||

766 chunkValue.$$typeof === REACT_ELEMENT_TYPE) &&

767 !chunkValue._debugInfo

768 ) {

@@ -966,8 +1044,17 @@ function resolveModel(

1044

1045 function resolveText(response: Response, id: number, text: string): void {

1046 const chunks = response._chunks;

969 - // We assume that we always reference large strings after they've been

970 - // emitted.

1047 + if (enableFlightReadableStream) {

1048 + const chunk = chunks.get(id);

1049 + if (chunk && chunk.status !== PENDING) {

1050 + // If we get more data to an already resolved ID, we assume that it's

1051 + // a stream chunk since any other row shouldn't have more than one entry.

1052 + const streamChunk: InitializedStreamChunk<any> = (chunk: any);

1053 + const controller = streamChunk.reason;

1054 + controller.enqueueValue(text);

1055 + return;

1056 + }

1057 + }

1058 chunks.set(id, createInitializedTextChunk(response, text));

1059 }

1060

@@ -977,7 +1064,17 @@ function resolveBuffer(

1064 buffer: $ArrayBufferView | ArrayBuffer,

1065 ): void {

1066 const chunks = response._chunks;

980 - // We assume that we always reference buffers after they've been emitted.

1067 + if (enableFlightReadableStream) {

1068 + const chunk = chunks.get(id);

1069 + if (chunk && chunk.status !== PENDING) {

1070 + // If we get more data to an already resolved ID, we assume that it's

1071 + // a stream chunk since any other row shouldn't have more than one entry.

1072 + const streamChunk: InitializedStreamChunk<any> = (chunk: any);

1073 + const controller = streamChunk.reason;

1074 + controller.enqueueValue(buffer);

1075 + return;

1076 + }

1077 + }

1078 chunks.set(id, createInitializedBufferChunk(response, buffer));

1079 }

1080

@@ -1035,6 +1132,268 @@ function resolveModule(

1132 }

1133 }

1134

1135 +function resolveStream<T: ReadableStream | $AsyncIterable<any, any, void>>(

1136 + response: Response,

1137 + id: number,

1138 + stream: T,

1139 + controller: FlightStreamController,

1140 +): void {

1141 + const chunks = response._chunks;

1142 + const chunk = chunks.get(id);

1143 + if (!chunk) {

1144 + chunks.set(id, createInitializedStreamChunk(response, stream, controller));

1145 + return;

1146 + }

1147 + if (chunk.status !== PENDING) {

1148 + // We already resolved. We didn't expect to see this.

1149 + return;

1150 + }

1151 + const resolveListeners = chunk.value;

1152 + const resolvedChunk: InitializedStreamChunk<T> = (chunk: any);

1153 + resolvedChunk.status = INITIALIZED;

1154 + resolvedChunk.value = stream;

1155 + resolvedChunk.reason = controller;

1156 + if (resolveListeners !== null) {

1157 + wakeChunk(resolveListeners, chunk.value);

1158 + }

1159 +}

1160 +

1161 +function startReadableStream<T>(

1162 + response: Response,

1163 + id: number,

1164 + type: void | 'bytes',

1165 +): void {

1166 + let controller: ReadableStreamController = (null: any);

1167 + const stream = new ReadableStream({

1168 + type: type,

1169 + start(c) {

1170 + controller = c;

1171 + },

1172 + });

1173 + let previousBlockedChunk: SomeChunk<T> | null = null;

1174 + const flightController = {

1175 + enqueueValue(value: T): void {

1176 + if (previousBlockedChunk === null) {

1177 + controller.enqueue(value);

1178 + } else {

1179 + // We're still waiting on a previous chunk so we can't enqueue quite yet.

1180 + previousBlockedChunk.then(function () {

1181 + controller.enqueue(value);

1182 + });

1183 + }

1184 + },

1185 + enqueueModel(json: UninitializedModel): void {

1186 + if (previousBlockedChunk === null) {

1187 + // If we're not blocked on any other chunks, we can try to eagerly initialize

1188 + // this as a fast-path to avoid awaiting them.

1189 + const chunk: ResolvedModelChunk<T> = createResolvedModelChunk(

1190 + response,

1191 + json,

1192 + );

1193 + initializeModelChunk(chunk);

1194 + const initializedChunk: SomeChunk<T> = chunk;

1195 + if (initializedChunk.status === INITIALIZED) {

1196 + controller.enqueue(initializedChunk.value);

1197 + } else {

1198 + chunk.then(

1199 + v => controller.enqueue(v),

1200 + e => controller.error((e: any)),

1201 + );

1202 + previousBlockedChunk = chunk;

1203 + }

1204 + } else {

1205 + // We're still waiting on a previous chunk so we can't enqueue quite yet.

1206 + const blockedChunk = previousBlockedChunk;

1207 + const chunk: SomeChunk<T> = createPendingChunk(response);

1208 + chunk.then(

1209 + v => controller.enqueue(v),

1210 + e => controller.error((e: any)),

1211 + );

1212 + previousBlockedChunk = chunk;

1213 + blockedChunk.then(function () {

1214 + if (previousBlockedChunk === chunk) {

1215 + // We were still the last chunk so we can now clear the queue and return

1216 + // to synchronous emitting.

1217 + previousBlockedChunk = null;

1218 + }

1219 + resolveModelChunk(chunk, json);

1220 + });

1221 + }

1222 + },

1223 + close(json: UninitializedModel): void {

1224 + if (previousBlockedChunk === null) {

1225 + controller.close();

1226 + } else {

1227 + const blockedChunk = previousBlockedChunk;

1228 + // We shouldn't get any more enqueues after this so we can set it back to null.

1229 + previousBlockedChunk = null;

1230 + blockedChunk.then(() => controller.close());

1231 + }

1232 + },

1233 + error(error: mixed): void {

1234 + if (previousBlockedChunk === null) {

1235 + // $FlowFixMe[incompatible-call]

1236 + controller.error(error);

1237 + } else {

1238 + const blockedChunk = previousBlockedChunk;

1239 + // We shouldn't get any more enqueues after this so we can set it back to null.

1240 + previousBlockedChunk = null;

1241 + blockedChunk.then(() => controller.error((error: any)));

1242 + }

1243 + },

1244 + };

1245 + resolveStream(response, id, stream, flightController);

1246 +}

1247 +

1248 +const ASYNC_ITERATOR = Symbol.asyncIterator;

1249 +

1250 +function asyncIterator(this: $AsyncIterator<any, any, void>) {

1251 + // Self referencing iterator.

1252 + return this;

1253 +}

1254 +

1255 +function createIterator<T>(

1256 + next: (arg: void) => SomeChunk<IteratorResult<T, T>>,

1257 +): $AsyncIterator<T, T, void> {

1258 + const iterator: any = {

1259 + next: next,

1260 + // TODO: Add return/throw as options for aborting.

1261 + };

1262 + // TODO: The iterator could inherit the AsyncIterator prototype which is not exposed as

1263 + // a global but exists as a prototype of an AsyncGenerator. However, it's not needed

1264 + // to satisfy the iterable protocol.

1265 + (iterator: any)[ASYNC_ITERATOR] = asyncIterator;

1266 + return iterator;

1267 +}

1268 +

1269 +function startAsyncIterable<T>(

1270 + response: Response,

1271 + id: number,

1272 + iterator: boolean,

1273 +): void {

1274 + const buffer: Array<SomeChunk<IteratorResult<T, T>>> = [];

1275 + let closed = false;

1276 + let nextWriteIndex = 0;

1277 + const flightController = {

1278 + enqueueValue(value: T): void {

1279 + if (nextWriteIndex === buffer.length) {

1280 + buffer[nextWriteIndex] = createInitializedIteratorResultChunk(

1281 + response,

1282 + value,

1283 + false,

1284 + );

1285 + } else {

1286 + const chunk: PendingChunk<IteratorResult<T, T>> = (buffer[

1287 + nextWriteIndex

1288 + ]: any);

1289 + const resolveListeners = chunk.value;

1290 + const rejectListeners = chunk.reason;

1291 + const initializedChunk: InitializedChunk<IteratorResult<T, T>> =

1292 + (chunk: any);

1293 + initializedChunk.status = INITIALIZED;

1294 + initializedChunk.value = {done: false, value: value};

1295 + if (resolveListeners !== null) {

1296 + wakeChunkIfInitialized(chunk, resolveListeners, rejectListeners);

1297 + }

1298 + }

1299 + nextWriteIndex++;

1300 + },

1301 + enqueueModel(value: UninitializedModel): void {

1302 + if (nextWriteIndex === buffer.length) {

1303 + buffer[nextWriteIndex] = createResolvedIteratorResultChunk(

1304 + response,

1305 + value,

1306 + false,

1307 + );

1308 + } else {

1309 + resolveIteratorResultChunk(buffer[nextWriteIndex], value, false);

1310 + }

1311 + nextWriteIndex++;

1312 + },

1313 + close(value: UninitializedModel): void {

1314 + closed = true;

1315 + if (nextWriteIndex === buffer.length) {

1316 + buffer[nextWriteIndex] = createResolvedIteratorResultChunk(

1317 + response,

1318 + value,

1319 + true,

1320 + );

1321 + } else {

1322 + resolveIteratorResultChunk(buffer[nextWriteIndex], value, true);

1323 + }

1324 + nextWriteIndex++;

1325 + while (nextWriteIndex < buffer.length) {

1326 + // In generators, any extra reads from the iterator have the value undefined.

1327 + resolveIteratorResultChunk(

1328 + buffer[nextWriteIndex++],

1329 + '"$undefined"',

1330 + true,

1331 + );

1332 + }

1333 + },

1334 + error(error: Error): void {

1335 + closed = true;

1336 + if (nextWriteIndex === buffer.length) {

1337 + buffer[nextWriteIndex] =

1338 + createPendingChunk<IteratorResult<T, T>>(response);

1339 + }

1340 + while (nextWriteIndex < buffer.length) {

1341 + triggerErrorOnChunk(buffer[nextWriteIndex++], error);

1342 + }

1343 + },

1344 + };

1345 + const iterable: $AsyncIterable<T, T, void> = {

1346 + [ASYNC_ITERATOR](): $AsyncIterator<T, T, void> {

1347 + let nextReadIndex = 0;

1348 + return createIterator(arg => {

1349 + if (arg !== undefined) {

1350 + throw new Error(

1351 + 'Values cannot be passed to next() of AsyncIterables passed to Client Components.',

1352 + );

1353 + }

1354 + if (nextReadIndex === buffer.length) {

1355 + if (closed) {

1356 + // $FlowFixMe[invalid-constructor] Flow doesn't support functions as constructors

1357 + return new Chunk(

1358 + INITIALIZED,

1359 + {done: true, value: undefined},

1360 + null,

1361 + response,

1362 + );

1363 + }

1364 + buffer[nextReadIndex] =

1365 + createPendingChunk<IteratorResult<T, T>>(response);

1366 + }

1367 + return buffer[nextReadIndex++];

1368 + });

1369 + },

1370 + };

1371 + // TODO: If it's a single shot iterator we can optimize memory by cleaning up the buffer after

1372 + // reading through the end, but currently we favor code size over this optimization.

1373 + resolveStream(

1374 + response,

1375 + id,

1376 + iterator ? iterable[ASYNC_ITERATOR]() : iterable,

1377 + flightController,

1378 + );

1379 +}

1380 +

1381 +function stopStream(

1382 + response: Response,

1383 + id: number,

1384 + row: UninitializedModel,

1385 +): void {

1386 + const chunks = response._chunks;

1387 + const chunk = chunks.get(id);

1388 + if (!chunk || chunk.status !== INITIALIZED) {

1389 + // We didn't expect not to have an existing stream;

1390 + return;

1391 + }

1392 + const streamChunk: InitializedStreamChunk<any> = (chunk: any);

1393 + const controller = streamChunk.reason;

1394 + controller.close(row === '' ? '"$undefined"' : row);

1395 +}

1396 +

1397 type ErrorWithDigest = Error & {digest?: string};

1398 function resolveErrorProd(

1399 response: Response,

@@ -1362,6 +1721,41 @@ function processFullRow(

1721 'matching versions on the server and the client.',

1722 );

1723 }

1724 + case 82 /* "R" */: {

1725 + if (enableFlightReadableStream) {

1726 + startReadableStream(response, id, undefined);

1727 + return;

1728 + }

1729 + }

1730 + // Fallthrough

1731 + case 114 /* "r" */: {

1732 + if (enableFlightReadableStream) {

1733 + startReadableStream(response, id, 'bytes');

1734 + return;

1735 + }

1736 + }

1737 + // Fallthrough

1738 + case 88 /* "X" */: {

1739 + if (enableFlightReadableStream) {

1740 + startAsyncIterable(response, id, false);

1741 + return;

1742 + }

1743 + }

1744 + // Fallthrough

1745 + case 120 /* "x" */: {

1746 + if (enableFlightReadableStream) {

1747 + startAsyncIterable(response, id, true);

1748 + return;

1749 + }

1750 + }

1751 + // Fallthrough

1752 + case 67 /* "C" */: {

1753 + if (enableFlightReadableStream) {

1754 + stopStream(response, id, row);

1755 + return;

1756 + }

1757 + }

1758 + // Fallthrough

1759 case 80 /* "P" */: {

1760 if (enablePostpone) {

1761 if (__DEV__) {

@@ -1433,7 +1827,11 @@ export function processBinaryChunk(

1827 rowTag = resolvedRowTag;

1828 rowState = ROW_LENGTH;

1829 i++;

1436 - } else if (resolvedRowTag > 64 && resolvedRowTag < 91 /* "A"-"Z" */) {

1830 + } else if (

1831 + (resolvedRowTag > 64 && resolvedRowTag < 91) /* "A"-"Z" */ ||

1832 + resolvedRowTag === 114 /* "r" */ ||

1833 + resolvedRowTag === 120 /* "x" */

1834 + ) {

1835 rowTag = resolvedRowTag;

1836 rowState = ROW_CHUNK_BY_NEWLINE;

1837 i++;

packages/react-client/src/__tests__/ReactFlight-test.js

+260 -1

@@ -2068,6 +2068,149 @@ describe('ReactFlight', () => {

2068 expect(ReactNoop).toMatchRenderedOutput(<div>Ba</div>);

2069 });

2070

2071 + it('shares state when moving keyed Server Components that render fragments', async () => {

2072 + function StatefulClient({name, initial}) {

2073 + const [state] = React.useState(initial);

2074 + return <span>{state}</span>;

2075 + }

2076 + const Stateful = clientReference(StatefulClient);

2077 +

2078 + function ServerComponent({item, initial}) {

2079 + return [

2080 + <Stateful key="a" initial={'a' + initial} />,

2081 + <Stateful key="b" initial={'b' + initial} />,

2082 + ];

2083 + }

2084 +

2085 + const transport = ReactNoopFlightServer.render(

2086 + <div>

2087 + <ServerComponent key="A" initial={1} />

2088 + <ServerComponent key="B" initial={2} />

2089 + </div>,

2090 + );

2091 +

2092 + await act(async () => {

2093 + ReactNoop.render(await ReactNoopFlightClient.read(transport));

2094 + });

2095 +

2096 + expect(ReactNoop).toMatchRenderedOutput(

2097 + <div>

2098 + <span>a1</span>

2099 + <span>b1</span>

2100 + <span>a2</span>

2101 + <span>b2</span>

2102 + </div>,

2103 + );

2104 +

2105 + // We swap the Server Components and the state of each child inside each fragment should move.

2106 + // Really the Fragment itself moves.

2107 + const transport2 = ReactNoopFlightServer.render(

2108 + <div>

2109 + <ServerComponent key="B" initial={4} />

2110 + <ServerComponent key="A" initial={3} />

2111 + </div>,

2112 + );

2113 +

2114 + await act(async () => {

2115 + ReactNoop.render(await ReactNoopFlightClient.read(transport2));

2116 + });

2117 +

2118 + expect(ReactNoop).toMatchRenderedOutput(

2119 + <div>

2120 + <span>a2</span>

2121 + <span>b2</span>

2122 + <span>a1</span>

2123 + <span>b1</span>

2124 + </div>,

2125 + );

2126 + });

2127 +

2128 + // @gate enableFlightReadableStream

2129 + it('shares state when moving keyed Server Components that render async iterables', async () => {

2130 + function StatefulClient({name, initial}) {

2131 + const [state] = React.useState(initial);

2132 + return <span>{state}</span>;

2133 + }

2134 + const Stateful = clientReference(StatefulClient);

2135 +

2136 + function ServerComponent({item, initial}) {

2137 + // While the ServerComponent itself could be an async generator, single-shot iterables

2138 + // are not supported as React children since React might need to re-map them based on

2139 + // state updates. So we create an AsyncIterable instead.

2140 + return {

2141 + async *[Symbol.asyncIterator]() {

2142 + yield <Stateful key="a" initial={'a' + initial} />;

2143 + yield <Stateful key="b" initial={'b' + initial} />;

2144 + },

2145 + };

2146 + }

2147 +

2148 + function ListClient({children}) {

2149 + // TODO: Unwrap AsyncIterables natively in React. For now we do it in this wrapper.

2150 + const resolvedChildren = [];

2151 + // eslint-disable-next-line no-for-of-loops/no-for-of-loops

2152 + for (const fragment of children) {

2153 + // We should've wrapped each child in a keyed Fragment.

2154 + expect(fragment.type).toBe(React.Fragment);

2155 + const fragmentChildren = [];

2156 + const iterator = fragment.props.children[Symbol.asyncIterator]();

2157 + for (let entry; !(entry = React.use(iterator.next())).done; ) {

2158 + fragmentChildren.push(entry.value);

2159 + }

2160 + resolvedChildren.push(

2161 + <React.Fragment key={fragment.key}>

2162 + {fragmentChildren}

2163 + </React.Fragment>,

2164 + );

2165 + }

2166 + return <div>{resolvedChildren}</div>;

2167 + }

2168 +

2169 + const List = clientReference(ListClient);

2170 +

2171 + const transport = ReactNoopFlightServer.render(

2172 + <List>

2173 + <ServerComponent key="A" initial={1} />

2174 + <ServerComponent key="B" initial={2} />

2175 + </List>,

2176 + );

2177 +

2178 + await act(async () => {

2179 + ReactNoop.render(await ReactNoopFlightClient.read(transport));

2180 + });

2181 +

2182 + expect(ReactNoop).toMatchRenderedOutput(

2183 + <div>

2184 + <span>a1</span>

2185 + <span>b1</span>

2186 + <span>a2</span>

2187 + <span>b2</span>

2188 + </div>,

2189 + );

2190 +

2191 + // We swap the Server Components and the state of each child inside each fragment should move.

2192 + // Really the Fragment itself moves.

2193 + const transport2 = ReactNoopFlightServer.render(

2194 + <List>

2195 + <ServerComponent key="B" initial={4} />

2196 + <ServerComponent key="A" initial={3} />

2197 + </List>,

2198 + );

2199 +

2200 + await act(async () => {

2201 + ReactNoop.render(await ReactNoopFlightClient.read(transport2));

2202 + });

2203 +

2204 + expect(ReactNoop).toMatchRenderedOutput(

2205 + <div>

2206 + <span>a2</span>

2207 + <span>b2</span>

2208 + <span>a1</span>

2209 + <span>b1</span>

2210 + </div>,

2211 + );

2212 + });

2213 +

2214 it('preserves debug info for server-to-server pass through', async () => {

2215 function ThirdPartyLazyComponent() {

2216 return <span>!</span>;

@@ -2081,6 +2224,10 @@ describe('ReactFlight', () => {

2224 return <span>stranger</span>;

2225 }

2226

2227 + function ThirdPartyFragmentComponent() {

2228 + return [<span>Who</span>, ' ', <span>dis?</span>];

2229 + }

2230 +

2231 function ServerComponent({transport}) {

2232 // This is a Server Component that receives other Server Components from a third party.

2233 const children = ReactNoopFlightClient.read(transport);

@@ -2090,7 +2237,7 @@ describe('ReactFlight', () => {

2237 const promiseComponent = Promise.resolve(<ThirdPartyComponent />);

2238

2239 const thirdPartyTransport = ReactNoopFlightServer.render(

2093 - [promiseComponent, lazy],

2240 + [promiseComponent, lazy, <ThirdPartyFragmentComponent />],

2241 {

2242 environmentName: 'third-party',

2243 },

@@ -2123,6 +2270,17 @@ describe('ReactFlight', () => {

2270 ? [{name: 'ThirdPartyLazyComponent', env: 'third-party', owner: null}]

2271 : undefined,

2272 );

2273 + expect(thirdPartyChildren[2]._debugInfo).toEqual(

2274 + __DEV__

2275 + ? [

2276 + {

2277 + name: 'ThirdPartyFragmentComponent',

2278 + env: 'third-party',

2279 + owner: null,

2280 + },

2281 + ]

2282 + : undefined,

2283 + );

2284 ReactNoop.render(result);

2285 });

2286

@@ -2130,6 +2288,107 @@ describe('ReactFlight', () => {

2288 <div>

2289 Hello, <span>stranger</span>

2290 <span>!</span>

2291 + <span>Who</span> <span>dis?</span>

2292 + </div>,

2293 + );

2294 + });

2295 +

2296 + // @gate enableFlightReadableStream

2297 + it('preserves debug info for server-to-server pass through of async iterables', async () => {

2298 + let resolve;

2299 + const iteratorPromise = new Promise(r => (resolve = r));

2300 +

2301 + function ThirdPartyAsyncIterableComponent({item, initial}) {

2302 + // While the ServerComponent itself could be an async generator, single-shot iterables

2303 + // are not supported as React children since React might need to re-map them based on

2304 + // state updates. So we create an AsyncIterable instead.

2305 + return {

2306 + async *[Symbol.asyncIterator]() {

2307 + yield <span>Who</span>;

2308 + yield <span>dis?</span>;

2309 + resolve();

2310 + },

2311 + };

2312 + }

2313 +

2314 + function ListClient({children: fragment}) {

2315 + // TODO: Unwrap AsyncIterables natively in React. For now we do it in this wrapper.

2316 + const resolvedChildren = [];

2317 + const iterator = fragment.props.children[Symbol.asyncIterator]();

2318 + for (let entry; !(entry = React.use(iterator.next())).done; ) {

2319 + resolvedChildren.push(entry.value);

2320 + }

2321 + return <div>{resolvedChildren}</div>;

2322 + }

2323 +

2324 + const List = clientReference(ListClient);

2325 +

2326 + function Keyed({children}) {

2327 + // Keying this should generate a fragment.

2328 + return children;

2329 + }

2330 +

2331 + function ServerComponent({transport}) {

2332 + // This is a Server Component that receives other Server Components from a third party.

2333 + const children = ReactServer.use(

2334 + ReactNoopFlightClient.read(transport),

2335 + ).root;

2336 + return (

2337 + <List>

2338 + <Keyed key="keyed">{children}</Keyed>

2339 + </List>

2340 + );

2341 + }

2342 +

2343 + const thirdPartyTransport = ReactNoopFlightServer.render(

2344 + {root: <ThirdPartyAsyncIterableComponent />},

2345 + {

2346 + environmentName: 'third-party',

2347 + },

2348 + );

2349 +

2350 + if (gate(flag => flag.enableFlightReadableStream)) {

2351 + // Wait for the iterator to finish

2352 + await iteratorPromise;

2353 + }

2354 + await 0; // One more tick for the return value / closing.

2355 +

2356 + const transport = ReactNoopFlightServer.render(

2357 + <ServerComponent transport={thirdPartyTransport} />,

2358 + );

2359 +

2360 + await act(async () => {

2361 + const promise = ReactNoopFlightClient.read(transport);

2362 + expect(promise._debugInfo).toEqual(

2363 + __DEV__

2364 + ? [{name: 'ServerComponent', env: 'Server', owner: null}]

2365 + : undefined,

2366 + );

2367 + const result = await promise;

2368 + const thirdPartyFragment = await result.props.children;

2369 + expect(thirdPartyFragment._debugInfo).toEqual(

2370 + __DEV__ ? [{name: 'Keyed', env: 'Server', owner: null}] : undefined,

2371 + );

2372 + // We expect the debug info to be transferred from the inner stream to the outer.

2373 + expect(thirdPartyFragment.props.children._debugInfo).toEqual(

2374 + __DEV__

2375 + ? [

2376 + {

2377 + name: 'ThirdPartyAsyncIterableComponent',

2378 + env: 'third-party',

2379 + owner: null,

2380 + },

2381 + ]

2382 + : undefined,

2383 + );

2384 +

2385 + ReactNoop.render(result);

2386 + });

2387 +

2388 + expect(ReactNoop).toMatchRenderedOutput(

2389 + <div>

2390 + <span>Who</span>

2391 + <span>dis?</span>

2392 </div>,

2393 );

2394 });

packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMBrowser-test.js

+381

@@ -1406,4 +1406,385 @@ describe('ReactFlightDOMBrowser', () => {

1406 expect(postponed).toBe('testing postpone');

1407 expect(error).toBe(null);

1408 });

1409 +

1410 + function passThrough(stream) {

1411 + // Simulate more realistic network by splitting up and rejoining some chunks.

1412 + // This lets us test that we don't accidentally rely on particular bounds of the chunks.

1413 + return new ReadableStream({

1414 + async start(controller) {

1415 + const reader = stream.getReader();

1416 + function push() {

1417 + reader.read().then(({done, value}) => {

1418 + if (done) {

1419 + controller.close();

1420 + return;

1421 + }

1422 + controller.enqueue(value);

1423 + push();

1424 + return;

1425 + });

1426 + }

1427 + push();

1428 + },

1429 + });

1430 + }

1431 +

1432 + // @gate enableFlightReadableStream

1433 + it('should supports streaming ReadableStream with objects', async () => {

1434 + const errors = [];

1435 + let controller1;

1436 + let controller2;

1437 + const s1 = new ReadableStream({

1438 + start(c) {

1439 + controller1 = c;

1440 + },

1441 + });

1442 + const s2 = new ReadableStream({

1443 + start(c) {

1444 + controller2 = c;

1445 + },

1446 + });

1447 + const rscStream = ReactServerDOMServer.renderToReadableStream(

1448 + {

1449 + s1,

1450 + s2,

1451 + },

1452 + {},

1453 + {

1454 + onError(x) {

1455 + errors.push(x);

1456 + return x;

1457 + },

1458 + },

1459 + );

1460 + const result = await ReactServerDOMClient.createFromReadableStream(

1461 + passThrough(rscStream),

1462 + );

1463 + const reader1 = result.s1.getReader();

1464 + const reader2 = result.s2.getReader();

1465 +

1466 + controller1.enqueue({hello: 'world'});

1467 + controller2.enqueue({hi: 'there'});

1468 + expect(await reader1.read()).toEqual({

1469 + value: {hello: 'world'},

1470 + done: false,

1471 + });

1472 + expect(await reader2.read()).toEqual({

1473 + value: {hi: 'there'},

1474 + done: false,

1475 + });

1476 +

1477 + controller1.enqueue('text1');

1478 + controller2.enqueue('text2');

1479 + controller1.close();

1480 + controller2.error('rejected');

1481 +

1482 + expect(await reader1.read()).toEqual({

1483 + value: 'text1',

1484 + done: false,

1485 + });

1486 + expect(await reader1.read()).toEqual({

1487 + value: undefined,

1488 + done: true,

1489 + });

1490 + expect(await reader2.read()).toEqual({

1491 + value: 'text2',

1492 + done: false,

1493 + });

1494 + let error = null;

1495 + try {

1496 + await reader2.read();

1497 + } catch (x) {

1498 + error = x;

1499 + }

1500 + expect(error.digest).toBe('rejected');

1501 + expect(errors).toEqual(['rejected']);

1502 + });

1503 +

1504 + // @gate enableFlightReadableStream

1505 + it('should cancels the underlying ReadableStream when we are cancelled', async () => {

1506 + let controller;

1507 + let cancelReason;

1508 + const s = new ReadableStream({

1509 + start(c) {

1510 + controller = c;

1511 + },

1512 + cancel(r) {

1513 + cancelReason = r;

1514 + },

1515 + });

1516 + let loggedReason;

1517 + const rscStream = ReactServerDOMServer.renderToReadableStream(

1518 + s,

1519 + {},

1520 + {

1521 + onError(reason) {

1522 + loggedReason = reason;

1523 + },

1524 + },

1525 + );

1526 + const reader = rscStream.getReader();

1527 + controller.enqueue('hi');

1528 + const reason = new Error('aborted');

1529 + reader.cancel(reason);

1530 + await reader.read();

1531 + expect(cancelReason).toBe(reason);

1532 + expect(loggedReason).toBe(reason);

1533 + });

1534 +

1535 + // @gate enableFlightReadableStream

1536 + it('should cancels the underlying ReadableStream when we abort', async () => {

1537 + const errors = [];

1538 + let controller;

1539 + let cancelReason;

1540 + const abortController = new AbortController();

1541 + const s = new ReadableStream({

1542 + start(c) {

1543 + controller = c;

1544 + },

1545 + cancel(r) {

1546 + cancelReason = r;

1547 + },

1548 + });

1549 + const rscStream = ReactServerDOMServer.renderToReadableStream(

1550 + s,

1551 + {},

1552 + {

1553 + signal: abortController.signal,

1554 + onError(x) {

1555 + errors.push(x);

1556 + return x.message;

1557 + },

1558 + },

1559 + );

1560 + const result = await ReactServerDOMClient.createFromReadableStream(

1561 + passThrough(rscStream),

1562 + );

1563 + const reader = result.getReader();

1564 + controller.enqueue('hi');

1565 +

1566 + await 0;

1567 +

1568 + const reason = new Error('aborted');

1569 + abortController.abort(reason);

1570 +

1571 + // We should be able to read the part we already emitted before the abort

1572 + expect(await reader.read()).toEqual({

1573 + value: 'hi',

1574 + done: false,

1575 + });

1576 +

1577 + expect(cancelReason).toBe(reason);

1578 +

1579 + let error = null;

1580 + try {

1581 + await reader.read();

1582 + } catch (x) {

1583 + error = x;

1584 + }

1585 + expect(error.digest).toBe('aborted');

1586 + expect(errors).toEqual([reason]);

1587 + });

1588 +

1589 + // @gate enableFlightReadableStream

1590 + it('should supports streaming AsyncIterables with objects', async () => {

1591 + let resolve;

1592 + const wait = new Promise(r => (resolve = r));

1593 + const errors = [];

1594 + const multiShotIterable = {

1595 + async *[Symbol.asyncIterator]() {

1596 + const next = yield {hello: 'A'};

1597 + expect(next).toBe(undefined);

1598 + await wait;

1599 + yield {hi: 'B'};

1600 + return 'C';

1601 + },

1602 + };

1603 + const singleShotIterator = (async function* () {

1604 + const next = yield {hello: 'D'};

1605 + expect(next).toBe(undefined);

1606 + await wait;

1607 + yield {hi: 'E'};

1608 + // eslint-disable-next-line no-throw-literal

1609 + throw 'F';

1610 + })();

1611 +

1612 + const rscStream = ReactServerDOMServer.renderToReadableStream(

1613 + {

1614 + multiShotIterable,

1615 + singleShotIterator,

1616 + },

1617 + {},

1618 + {

1619 + onError(x) {

1620 + errors.push(x);

1621 + return x;

1622 + },

1623 + },

1624 + );

1625 + const result = await ReactServerDOMClient.createFromReadableStream(

1626 + passThrough(rscStream),

1627 + );

1628 +

1629 + const iterator1 = result.multiShotIterable[Symbol.asyncIterator]();

1630 + const iterator2 = result.singleShotIterator[Symbol.asyncIterator]();

1631 +

1632 + expect(iterator1).not.toBe(result.multiShotIterable);

1633 + expect(iterator2).toBe(result.singleShotIterator);

1634 +

1635 + expect(await iterator1.next()).toEqual({

1636 + value: {hello: 'A'},

1637 + done: false,

1638 + });

1639 + expect(await iterator2.next()).toEqual({

1640 + value: {hello: 'D'},

1641 + done: false,

1642 + });

1643 +

1644 + await resolve();

1645 +

1646 + expect(await iterator1.next()).toEqual({

1647 + value: {hi: 'B'},

1648 + done: false,

1649 + });

1650 + expect(await iterator2.next()).toEqual({

1651 + value: {hi: 'E'},

1652 + done: false,

1653 + });

1654 + expect(await iterator1.next()).toEqual({

1655 + value: 'C', // Return value

1656 + done: true,

1657 + });

1658 + expect(await iterator1.next()).toEqual({

1659 + value: undefined,

1660 + done: true,

1661 + });

1662 +

1663 + let error = null;

1664 + try {

1665 + await iterator2.next();

1666 + } catch (x) {

1667 + error = x;

1668 + }

1669 + expect(error.digest).toBe('F');

1670 + expect(errors).toEqual(['F']);

1671 +

1672 + // Multi-shot iterables should be able to do the same thing again

1673 + const iterator3 = result.multiShotIterable[Symbol.asyncIterator]();

1674 +

1675 + expect(iterator3).not.toBe(iterator1);

1676 +

1677 + // We should be able to iterate over the iterable again and it should be

1678 + // synchronously available using instrumented promises so that React can

1679 + // rerender it synchronously.

1680 + expect(iterator3.next().value).toEqual({

1681 + value: {hello: 'A'},

1682 + done: false,

1683 + });

1684 + expect(iterator3.next().value).toEqual({

1685 + value: {hi: 'B'},

1686 + done: false,

1687 + });

1688 + expect(iterator3.next().value).toEqual({

1689 + value: 'C', // Return value

1690 + done: true,

1691 + });

1692 + expect(iterator3.next().value).toEqual({

1693 + value: undefined,

1694 + done: true,

1695 + });

1696 +

1697 + expect(() => iterator3.next('this is not allowed')).toThrow(

1698 + 'Values cannot be passed to next() of AsyncIterables passed to Client Components.',

1699 + );

1700 + });

1701 +

1702 + // @gate enableFlightReadableStream

1703 + it('should cancels the underlying AsyncIterable when we are cancelled', async () => {

1704 + let resolve;

1705 + const wait = new Promise(r => (resolve = r));

1706 + let thrownReason;

1707 + const iterator = (async function* () {

1708 + try {

1709 + await wait;

1710 + yield 'a';

1711 + yield 'b';

1712 + } catch (x) {

1713 + thrownReason = x;

1714 + }

1715 + yield 'c';

1716 + })();

1717 + let loggedReason;

1718 + const rscStream = ReactServerDOMServer.renderToReadableStream(

1719 + iterator,

1720 + {},

1721 + {

1722 + onError(reason) {

1723 + loggedReason = reason;

1724 + },

1725 + },

1726 + );

1727 + const reader = rscStream.getReader();

1728 + const reason = new Error('aborted');

1729 + reader.cancel(reason);

1730 + await resolve();

1731 + await reader.read();

1732 + expect(thrownReason).toBe(reason);

1733 + expect(loggedReason).toBe(reason);

1734 + });

1735 +

1736 + // @gate enableFlightReadableStream

1737 + it('should cancels the underlying AsyncIterable when we abort', async () => {

1738 + const errors = [];

1739 + const abortController = new AbortController();

1740 + let resolve;

1741 + const wait = new Promise(r => (resolve = r));

1742 + let thrownReason;

1743 + const iterator = (async function* () {

1744 + try {

1745 + yield 'a';

1746 + await wait;

1747 + yield 'b';

1748 + } catch (x) {

1749 + thrownReason = x;

1750 + }

1751 + yield 'c';

1752 + })();

1753 + const rscStream = ReactServerDOMServer.renderToReadableStream(

1754 + iterator,

1755 + {},

1756 + {

1757 + signal: abortController.signal,

1758 + onError(x) {

1759 + errors.push(x);

1760 + return x.message;

1761 + },

1762 + },

1763 + );

1764 + const result = await ReactServerDOMClient.createFromReadableStream(

1765 + passThrough(rscStream),

1766 + );

1767 +

1768 + const reason = new Error('aborted');

1769 + abortController.abort(reason);

1770 +

1771 + await resolve();

1772 +

1773 + // We should be able to read the part we already emitted before the abort

1774 + expect(await result.next()).toEqual({

1775 + value: 'a',

1776 + done: false,

1777 + });

1778 +

1779 + expect(thrownReason).toBe(reason);

1780 +

1781 + let error = null;

1782 + try {

1783 + await result.next();

1784 + } catch (x) {

1785 + error = x;

1786 + }

1787 + expect(error.digest).toBe('aborted');

1788 + expect(errors).toEqual([reason]);

1789 + });

1790 });

packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMEdge-test.js

+236

@@ -80,6 +80,7 @@ describe('ReactFlightDOMEdge', () => {

80 reader.read().then(({done, value}) => {

81 if (done) {

82 controller.enqueue(prevChunk);

83 + prevChunk = new Uint8Array(0);

84 controller.close();

85 return;

86 }

@@ -90,7 +91,18 @@ describe('ReactFlightDOMEdge', () => {

91 controller.enqueue(chunk.subarray(0, chunk.length - 50));

92 prevChunk = chunk.subarray(chunk.length - 50);

93 } else {

94 + // Wait to see if we get some more bytes to join in.

95 prevChunk = chunk;

96 + // Flush if we don't get any more.

97 + (async function flushAfterAFewTasks() {

98 + for (let i = 0; i < 10; i++) {

99 + await i;

100 + }

101 + if (prevChunk.byteLength > 0) {

102 + controller.enqueue(prevChunk);

103 + }

104 + prevChunk = new Uint8Array(0);

105 + })();

106 }

107 push();

108 });

@@ -112,6 +124,18 @@ describe('ReactFlightDOMEdge', () => {

124 }

125 }

126

127 + async function readByteLength(stream) {

128 + const reader = stream.getReader();

129 + let length = 0;

130 + while (true) {

131 + const {done, value} = await reader.read();

132 + if (done) {

133 + return length;

134 + }

135 + length += value.byteLength;

136 + }

137 + }

138 +

139 it('should allow an alternative module mapping to be used for SSR', async () => {

140 function ClientComponent() {

141 return <span>Client Component</span>;

@@ -430,6 +454,97 @@ describe('ReactFlightDOMEdge', () => {

454 expect(result.get('value')).toBe('hello');

455 });

456

457 + // @gate enableFlightReadableStream

458 + it('can pass an async import to a ReadableStream while enqueuing in order', async () => {

459 + let resolve;

460 + const promise = new Promise(r => (resolve = r));

461 +

462 + const asyncClient = clientExports(promise);

463 +

464 + // We await the value on the servers so it's an async value that the client should wait for

465 + const awaitedValue = await asyncClient;

466 +

467 + const s = new ReadableStream({

468 + start(c) {

469 + c.enqueue('hello');

470 + c.enqueue(awaitedValue);

471 + c.enqueue('!');

472 + c.close();

473 + },

474 + });

475 +

476 + const stream = passThrough(

477 + ReactServerDOMServer.renderToReadableStream(s, webpackMap),

478 + );

479 +

480 + const result = await ReactServerDOMClient.createFromReadableStream(stream, {

481 + ssrManifest: {

482 + moduleMap: null,

483 + moduleLoading: null,

484 + },

485 + });

486 +

487 + const reader = result.getReader();

488 +

489 + expect(await reader.read()).toEqual({value: 'hello', done: false});

490 +

491 + const readPromise = reader.read();

492 + // We resolve this after we've already received the '!' row.

493 + await resolve('world');

494 +

495 + expect(await readPromise).toEqual({value: 'world', done: false});

496 + expect(await reader.read()).toEqual({value: '!', done: false});

497 + expect(await reader.read()).toEqual({value: undefined, done: true});

498 + });

499 +

500 + // @gate enableFlightReadableStream

501 + it('can pass an async import a AsyncIterable while allowing peaking at future values', async () => {

502 + let resolve;

503 + const promise = new Promise(r => (resolve = r));

504 +

505 + const asyncClient = clientExports(promise);

506 +

507 + const multiShotIterable = {

508 + async *[Symbol.asyncIterator]() {

509 + yield 'hello';

510 + // We await the value on the servers so it's an async value that the client should wait for

511 + yield await asyncClient;

512 + yield '!';

513 + },

514 + };

515 +

516 + const stream = passThrough(

517 + ReactServerDOMServer.renderToReadableStream(

518 + multiShotIterable,

519 + webpackMap,

520 + ),

521 + );

522 +

523 + // Parsing the root blocks because the module hasn't loaded yet

524 + const result = await ReactServerDOMClient.createFromReadableStream(stream, {

525 + ssrManifest: {

526 + moduleMap: null,

527 + moduleLoading: null,

528 + },

529 + });

530 +

531 + const iterator = result[Symbol.asyncIterator]();

532 +

533 + expect(await iterator.next()).toEqual({value: 'hello', done: false});

534 +

535 + const readPromise = iterator.next();

536 +

537 + // While the previous promise didn't resolve yet, we should be able to peak at the next value

538 + // by iterating past it.

539 + expect(await iterator.next()).toEqual({value: '!', done: false});

540 +

541 + // We resolve the previous row after we've already received the '!' row.

542 + await resolve('world');

543 + expect(await readPromise).toEqual({value: 'world', done: false});

544 +

545 + expect(await iterator.next()).toEqual({value: undefined, done: true});

546 + });

547 +

548 it('warns if passing a this argument to bind() of a server reference', async () => {

549 const ServerModule = serverExports({

550 greet: function () {},

@@ -456,4 +571,125 @@ describe('ReactFlightDOMEdge', () => {

571 {withoutStack: true},

572 );

573 });

574 +

575 + // @gate enableFlightReadableStream && enableBinaryFlight

576 + it('should supports ReadableStreams with typed arrays', async () => {

577 + const buffer = new Uint8Array([

578 + 123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,

579 + ]).buffer;

580 + const buffers = [

581 + buffer,

582 + new Int8Array(buffer, 1),

583 + new Uint8Array(buffer, 2),

584 + new Uint8ClampedArray(buffer, 2),

585 + new Int16Array(buffer, 2),

586 + new Uint16Array(buffer, 2),

587 + new Int32Array(buffer, 4),

588 + new Uint32Array(buffer, 4),

589 + new Float32Array(buffer, 4),

590 + new Float64Array(buffer, 0),

591 + new BigInt64Array(buffer, 0),

592 + new BigUint64Array(buffer, 0),

593 + new DataView(buffer, 3),

594 + ];

595 +

596 + // This is not a binary stream, it's a stream that contain binary chunks.

597 + const s = new ReadableStream({

598 + start(c) {

599 + for (let i = 0; i < buffers.length; i++) {

600 + c.enqueue(buffers[i]);

601 + }

602 + c.close();

603 + },

604 + });

605 +

606 + const stream = ReactServerDOMServer.renderToReadableStream(s, {});

607 +

608 + const [stream1, stream2] = passThrough(stream).tee();

609 +

610 + const result = await ReactServerDOMClient.createFromReadableStream(

611 + stream1,

612 + {

613 + ssrManifest: {

614 + moduleMap: null,

615 + moduleLoading: null,

616 + },

617 + },

618 + );

619 +

620 + expect(await readByteLength(stream2)).toBeLessThan(300);

621 +

622 + const streamedBuffers = [];

623 + const reader = result.getReader();

624 + let entry;

625 + while (!(entry = await reader.read()).done) {

626 + streamedBuffers.push(entry.value);

627 + }

628 +

629 + expect(streamedBuffers).toEqual(buffers);

630 + });

631 +

632 + // @gate enableFlightReadableStream && enableBinaryFlight

633 + it('should support BYOB binary ReadableStreams', async () => {

634 + const buffer = new Uint8Array([

635 + 123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,

636 + ]).buffer;

637 + const buffers = [

638 + new Int8Array(buffer, 1),

639 + new Uint8Array(buffer, 2),

640 + new Uint8ClampedArray(buffer, 2),

641 + new Int16Array(buffer, 2),

642 + new Uint16Array(buffer, 2),

643 + new Int32Array(buffer, 4),

644 + new Uint32Array(buffer, 4),

645 + new Float32Array(buffer, 4),

646 + new Float64Array(buffer, 0),

647 + new BigInt64Array(buffer, 0),

648 + new BigUint64Array(buffer, 0),

649 + new DataView(buffer, 3),

650 + ];

651 +

652 + // This a binary stream where each chunk ends up as Uint8Array.

653 + const s = new ReadableStream({

654 + type: 'bytes',

655 + start(c) {

656 + for (let i = 0; i < buffers.length; i++) {

657 + c.enqueue(buffers[i]);

658 + }

659 + c.close();

660 + },

661 + });

662 +

663 + const stream = ReactServerDOMServer.renderToReadableStream(s, {});

664 +

665 + const [stream1, stream2] = passThrough(stream).tee();

666 +

667 + const result = await ReactServerDOMClient.createFromReadableStream(

668 + stream1,

669 + {

670 + ssrManifest: {

671 + moduleMap: null,

672 + moduleLoading: null,

673 + },

674 + },

675 + );

676 +

677 + expect(await readByteLength(stream2)).toBeLessThan(300);

678 +

679 + const streamedBuffers = [];

680 + const reader = result.getReader({mode: 'byob'});

681 + let entry;

682 + while (!(entry = await reader.read(new Uint8Array(10))).done) {

683 + expect(entry.value instanceof Uint8Array).toBe(true);

684 + streamedBuffers.push(entry.value);

685 + }

686 +

687 + // The streamed buffers might be in different chunks and in Uint8Array form but

688 + // the concatenated bytes should be the same.

689 + expect(streamedBuffers.flatMap(t => Array.from(t))).toEqual(

690 + buffers.flatMap(c =>

691 + Array.from(new Uint8Array(c.buffer, c.byteOffset, c.byteLength)),

692 + ),

693 + );

694 + });

695 });

packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMNode-test.js

+94

@@ -9,6 +9,9 @@

10 'use strict';

12 +global.ReadableStream =

13 + require('web-streams-polyfill/ponyfill/es6').ReadableStream;

14 +

15 // Don't wait before processing work on the server.

16 // TODO: we can replace this with FlightServer.act().

17 global.setImmediate = cb => cb();

@@ -258,4 +261,95 @@ describe('ReactFlightDOMNode', () => {

261 '<script src="/path/to/chunk.js" async="" nonce="r4nd0m"></script><span>Client Component</span>',

262 );

263 });

264 +

265 + // @gate enableFlightReadableStream

266 + it('should cancels the underlying ReadableStream when we are cancelled', async () => {

267 + let controller;

268 + let cancelReason;

269 + const s = new ReadableStream({

270 + start(c) {

271 + controller = c;

272 + },

273 + cancel(r) {

274 + cancelReason = r;

275 + },

276 + });

277 +

278 + const rscStream = ReactServerDOMServer.renderToPipeableStream(

279 + s,

280 + {},

281 + {

282 + onError(error) {

283 + return error.message;

284 + },

285 + },

286 + );

287 +

288 + const writable = new Stream.PassThrough();

289 + rscStream.pipe(writable);

290 +

291 + controller.enqueue('hi');

292 +

293 + const reason = new Error('aborted');

294 + writable.destroy(reason);

295 +

296 + await new Promise(resolve => {

297 + writable.on('error', () => {

298 + resolve();

299 + });

300 + });

301 +

302 + expect(cancelReason.message).toBe(

303 + 'The destination stream errored while writing data.',

304 + );

305 + });

306 +

307 + // @gate enableFlightReadableStream

308 + it('should cancels the underlying ReadableStream when we abort', async () => {

309 + const errors = [];

310 + let controller;

311 + let cancelReason;

312 + const s = new ReadableStream({

313 + start(c) {

314 + controller = c;

315 + },

316 + cancel(r) {

317 + cancelReason = r;

318 + },

319 + });

320 + const rscStream = ReactServerDOMServer.renderToPipeableStream(

321 + s,

322 + {},

323 + {

324 + onError(x) {

325 + errors.push(x);

326 + return x.message;

327 + },

328 + },

329 + );

330 +

331 + const readable = new Stream.PassThrough();

332 + rscStream.pipe(readable);

333 +

334 + const result = await ReactServerDOMClient.createFromNodeStream(readable, {

335 + moduleMap: {},

336 + moduleLoading: webpackModuleLoading,

337 + });

338 + const reader = result.getReader();

339 + controller.enqueue('hi');

340 +

341 + const reason = new Error('aborted');

342 + rscStream.abort(reason);

343 +

344 + expect(cancelReason).toBe(reason);

345 +

346 + let error = null;

347 + try {

348 + await reader.read();

349 + } catch (x) {

350 + error = x;

351 + }

352 + expect(error.digest).toBe('aborted');

353 + expect(errors).toEqual([reason]);

354 + });

355 });

packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMReplyEdge-test.js

+30 -28

@@ -109,33 +109,35 @@ describe('ReactFlightDOMReplyEdge', () => {

109 expect(await result.arrayBuffer()).toEqual(await blob.arrayBuffer());

110 });

111

112 - it('can transport FormData (blobs)', async () => {

113 - const bytes = new Uint8Array([

114 - 123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,

115 - ]);

116 - const blob = new Blob([bytes, bytes], {

117 - type: 'application/x-test',

112 + if (typeof FormData !== 'undefined' && typeof File !== 'undefined') {

113 + it('can transport FormData (blobs)', async () => {

114 + const bytes = new Uint8Array([

115 + 123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,

116 + ]);

117 + const blob = new Blob([bytes, bytes], {

118 + type: 'application/x-test',

119 + });

120 +

121 + const formData = new FormData();

122 + formData.append('hi', 'world');

123 + formData.append('file', blob, 'filename.test');

124 +

125 + expect(formData.get('file') instanceof File).toBe(true);

126 + expect(formData.get('file').name).toBe('filename.test');

127 +

128 + const body = await ReactServerDOMClient.encodeReply(formData);

129 + const result = await ReactServerDOMServer.decodeReply(

130 + body,

131 + webpackServerMap,

132 + );

133 +

134 + expect(result instanceof FormData).toBe(true);

135 + expect(result.get('hi')).toBe('world');

136 + const resultBlob = result.get('file');

137 + expect(resultBlob instanceof Blob).toBe(true);

138 + expect(resultBlob.name).toBe('filename.test'); // In this direction we allow file name to pass through but not other direction.

139 + expect(resultBlob.size).toBe(bytes.length * 2);

140 + expect(await resultBlob.arrayBuffer()).toEqual(await blob.arrayBuffer());

141 });

119 -

120 - const formData = new FormData();

121 - formData.append('hi', 'world');

122 - formData.append('file', blob, 'filename.test');

123 -

124 - expect(formData.get('file') instanceof File).toBe(true);

125 - expect(formData.get('file').name).toBe('filename.test');

126 -

127 - const body = await ReactServerDOMClient.encodeReply(formData);

128 - const result = await ReactServerDOMServer.decodeReply(

129 - body,

130 - webpackServerMap,

131 - );

132 -

133 - expect(result instanceof FormData).toBe(true);

134 - expect(result.get('hi')).toBe('world');

135 - const resultBlob = result.get('file');

136 - expect(resultBlob instanceof Blob).toBe(true);

137 - expect(resultBlob.name).toBe('filename.test'); // In this direction we allow file name to pass through but not other direction.

138 - expect(resultBlob.size).toBe(bytes.length * 2);

139 - expect(await resultBlob.arrayBuffer()).toEqual(await blob.arrayBuffer());

140 - });

142 + }

143 });

packages/react-server/src/ReactFizzThenable.js

+13 -10

@@ -82,6 +82,9 @@ export function trackUsedThenable<T>(

82 // Only instrument the thenable if the status if not defined. If

83 // it's defined, but an unknown value, assume it's been instrumented by

84 // some custom userspace implementation. We treat it as "pending".

85 + // Attach a dummy listener, to ensure that any lazy initialization can

86 + // happen. Flight lazily parses JSON when the value is actually awaited.

87 + thenable.then(noop, noop);

88 } else {

89 const pendingThenable: PendingThenable<T> = (thenable: any);

90 pendingThenable.status = 'pending';

@@ -101,17 +104,17 @@ export function trackUsedThenable<T>(

104 }

105 },

106 );

107 + }

108

105 - // Check one more time in case the thenable resolved synchronously

106 - switch (thenable.status) {

107 - case 'fulfilled': {

108 - const fulfilledThenable: FulfilledThenable<T> = (thenable: any);

109 - return fulfilledThenable.value;

110 - }

111 - case 'rejected': {

112 - const rejectedThenable: RejectedThenable<T> = (thenable: any);

113 - throw rejectedThenable.reason;

114 - }

109 + // Check one more time in case the thenable resolved synchronously

110 + switch (thenable.status) {

111 + case 'fulfilled': {

112 + const fulfilledThenable: FulfilledThenable<T> = (thenable: any);

113 + return fulfilledThenable.value;

114 + }

115 + case 'rejected': {

116 + const rejectedThenable: RejectedThenable<T> = (thenable: any);

117 + throw rejectedThenable.reason;

118 }

119 }

120

packages/react-server/src/ReactFlightServer.js

+479 -44

@@ -20,6 +20,8 @@ import {

20 enableServerComponentLogs,

21 } from 'shared/ReactFeatureFlags';

23 +import {enableFlightReadableStream} from 'shared/ReactFeatureFlags';

24 +

25 import {

26 scheduleWork,

27 flushBuffered,

@@ -199,6 +201,8 @@ if (

201

202 const ObjectPrototype = Object.prototype;

203

204 +const ASYNC_ITERATOR = Symbol.asyncIterator;

205 +

206 type JSONValue =

207 | string

208 | boolean

@@ -236,6 +240,8 @@ export type ReactClientValue =

240 | null

241 | void

242 | bigint

243 + | ReadableStream

244 + | $AsyncIterable<ReactClientValue, ReactClientValue, void>

245 | Iterable<ReactClientValue>

246 | Array<ReactClientValue>

247 | Map<ReactClientValue, ReactClientValue>

@@ -282,6 +288,7 @@ export type Request = {

288 nextChunkId: number,

289 pendingChunks: number,

290 hints: Hints,

291 + abortListeners: Set<(reason: mixed) => void>,

292 abortableTasks: Set<Task>,

293 pingedTasks: Array<Task>,

294 completedImportChunks: Array<Chunk>,

@@ -378,6 +385,7 @@ export function createRequest(

385 nextChunkId: 0,

386 pendingChunks: 0,

387 hints,

388 + abortListeners: new Set(),

389 abortableTasks: abortSet,

390 pingedTasks: pingedTasks,

391 completedImportChunks: ([]: Array<Chunk>),

@@ -509,15 +517,220 @@ function serializeThenable(

517 emitErrorChunk(request, newTask.id, digest, reason);

518 }

519 request.abortableTasks.delete(newTask);

512 - if (request.destination !== null) {

513 - flushCompletedChunks(request, request.destination);

514 - }

520 + enqueueFlush(request);

521 },

522 );

523

524 return newTask.id;

525 }

526

527 +function serializeReadableStream(

528 + request: Request,

529 + task: Task,

530 + stream: ReadableStream,

531 +): string {

532 + // Detect if this is a BYOB stream. BYOB streams should be able to be read as bytes on the

533 + // receiving side. It also implies that different chunks can be split up or merged as opposed

534 + // to a readable stream that happens to have Uint8Array as the type which might expect it to be

535 + // received in the same slices.

536 + // $FlowFixMe: This is a Node.js extension.

537 + let supportsBYOB: void | boolean = stream.supportsBYOB;

538 + if (supportsBYOB === undefined) {

539 + try {

540 + // $FlowFixMe[extra-arg]: This argument is accepted.

541 + stream.getReader({mode: 'byob'}).releaseLock();

542 + supportsBYOB = true;

543 + } catch (x) {

544 + supportsBYOB = false;

545 + }

546 + }

547 +

548 + const reader = stream.getReader();

549 +

550 + // This task won't actually be retried. We just use it to attempt synchronous renders.

551 + const streamTask = createTask(

552 + request,

553 + task.model,

554 + task.keyPath,

555 + task.implicitSlot,

556 + request.abortableTasks,

557 + );

558 + request.abortableTasks.delete(streamTask);

559 +

560 + request.pendingChunks++; // The task represents the Start row. This adds a Stop row.

561 +

562 + const startStreamRow =

563 + streamTask.id.toString(16) + ':' + (supportsBYOB ? 'r' : 'R') + '\n';

564 + request.completedRegularChunks.push(stringToChunk(startStreamRow));

565 +

566 + // There's a race condition between when the stream is aborted and when the promise

567 + // resolves so we track whether we already aborted it to avoid writing twice.

568 + let aborted = false;

569 + function progress(entry: {done: boolean, value: ReactClientValue, ...}) {

570 + if (aborted) {

571 + return;

572 + }

573 +

574 + if (entry.done) {

575 + request.abortListeners.delete(error);

576 + const endStreamRow = streamTask.id.toString(16) + ':C\n';

577 + request.completedRegularChunks.push(stringToChunk(endStreamRow));

578 + enqueueFlush(request);

579 + aborted = true;

580 + } else {

581 + try {

582 + streamTask.model = entry.value;

583 + request.pendingChunks++;

584 + tryStreamTask(request, streamTask);

585 + enqueueFlush(request);

586 + reader.read().then(progress, error);

587 + } catch (x) {

588 + error(x);

589 + }

590 + }

591 + }

592 + function error(reason: mixed) {

593 + if (aborted) {

594 + return;

595 + }

596 + aborted = true;

597 + request.abortListeners.delete(error);

598 + if (

599 + enablePostpone &&

600 + typeof reason === 'object' &&

601 + reason !== null &&

602 + (reason: any).$$typeof === REACT_POSTPONE_TYPE

603 + ) {

604 + const postponeInstance: Postpone = (reason: any);

605 + logPostpone(request, postponeInstance.message);

606 + emitPostponeChunk(request, streamTask.id, postponeInstance);

607 + } else {

608 + const digest = logRecoverableError(request, reason);

609 + emitErrorChunk(request, streamTask.id, digest, reason);

610 + }

611 + enqueueFlush(request);

612 + // $FlowFixMe should be able to pass mixed

613 + reader.cancel(reason).then(error, error);

614 + }

615 + request.abortListeners.add(error);

616 + reader.read().then(progress, error);

617 + return serializeByValueID(streamTask.id);

618 +}

619 +

620 +function serializeAsyncIterable(

621 + request: Request,

622 + task: Task,

623 + iterable: $AsyncIterable<ReactClientValue, ReactClientValue, void>,

624 + iterator: $AsyncIterator<ReactClientValue, ReactClientValue, void>,

625 +): string {

626 + // Generators/Iterators are Iterables but they're also their own iterator

627 + // functions. If that's the case, we treat them as single-shot. Otherwise,

628 + // we assume that this iterable might be a multi-shot and allow it to be

629 + // iterated more than once on the client.

630 + const isIterator = iterable === iterator;

631 +

632 + // This task won't actually be retried. We just use it to attempt synchronous renders.

633 + const streamTask = createTask(

634 + request,

635 + task.model,

636 + task.keyPath,

637 + task.implicitSlot,

638 + request.abortableTasks,

639 + );

640 + request.abortableTasks.delete(streamTask);

641 +

642 + request.pendingChunks++; // The task represents the Start row. This adds a Stop row.

643 +

644 + const startStreamRow =

645 + streamTask.id.toString(16) + ':' + (isIterator ? 'x' : 'X') + '\n';

646 + request.completedRegularChunks.push(stringToChunk(startStreamRow));

647 +

648 + if (__DEV__) {

649 + const debugInfo: ?ReactDebugInfo = (iterable: any)._debugInfo;

650 + if (debugInfo) {

651 + forwardDebugInfo(request, streamTask.id, debugInfo);

652 + }

653 + }

654 +

655 + // There's a race condition between when the stream is aborted and when the promise

656 + // resolves so we track whether we already aborted it to avoid writing twice.

657 + let aborted = false;

658 + function progress(

659 + entry:

660 + | {done: false, +value: ReactClientValue, ...}

661 + | {done: true, +value: ReactClientValue, ...},

662 + ) {

663 + if (aborted) {

664 + return;

665 + }

666 +

667 + if (entry.done) {

668 + request.abortListeners.delete(error);

669 + let endStreamRow;

670 + if (entry.value === undefined) {

671 + endStreamRow = streamTask.id.toString(16) + ':C\n';

672 + } else {

673 + // Unlike streams, the last value may not be undefined. If it's not

674 + // we outline it and encode a reference to it in the closing instruction.

675 + try {

676 + const chunkId = outlineModel(request, entry.value);

677 + endStreamRow =

678 + streamTask.id.toString(16) +

679 + ':C' +

680 + stringify(serializeByValueID(chunkId)) +

681 + '\n';

682 + } catch (x) {

683 + error(x);

684 + return;

685 + }

686 + }

687 + request.completedRegularChunks.push(stringToChunk(endStreamRow));

688 + enqueueFlush(request);

689 + aborted = true;

690 + } else {

691 + try {

692 + streamTask.model = entry.value;

693 + request.pendingChunks++;

694 + tryStreamTask(request, streamTask);

695 + enqueueFlush(request);

696 + iterator.next().then(progress, error);

697 + } catch (x) {

698 + error(x);

699 + return;

700 + }

701 + }

702 + }

703 + function error(reason: mixed) {

704 + if (aborted) {

705 + return;

706 + }

707 + aborted = true;

708 + request.abortListeners.delete(error);

709 + if (

710 + enablePostpone &&

711 + typeof reason === 'object' &&

712 + reason !== null &&

713 + (reason: any).$$typeof === REACT_POSTPONE_TYPE

714 + ) {

715 + const postponeInstance: Postpone = (reason: any);

716 + logPostpone(request, postponeInstance.message);

717 + emitPostponeChunk(request, streamTask.id, postponeInstance);

718 + } else {

719 + const digest = logRecoverableError(request, reason);

720 + emitErrorChunk(request, streamTask.id, digest, reason);

721 + }

722 + enqueueFlush(request);

723 + if (typeof (iterator: any).throw === 'function') {

724 + // The iterator protocol doesn't necessarily include this but a generator do.

725 + // $FlowFixMe should be able to pass mixed

726 + iterator.throw(reason).then(error, error);

727 + }

728 + }

729 + request.abortListeners.add(error);

730 + iterator.next().then(progress, error);

731 + return serializeByValueID(streamTask.id);

732 +}

733 +

734 export function emitHint<Code: HintCode>(

735 request: Request,

736 code: Code,

@@ -691,6 +904,37 @@ function renderFragment(

904 task: Task,

905 children: $ReadOnlyArray<ReactClientValue>,

906 ): ReactJSONValue {

907 + if (enableServerComponentKeys && task.keyPath !== null) {

908 + // We have a Server Component that specifies a key but we're now splitting

909 + // the tree using a fragment.

910 + const fragment = [

911 + REACT_ELEMENT_TYPE,

912 + REACT_FRAGMENT_TYPE,

913 + task.keyPath,

914 + {children},

915 + ];

916 + if (!task.implicitSlot) {

917 + // If this was keyed inside a set. I.e. the outer Server Component was keyed

918 + // then we need to handle reorders of the whole set. To do this we need to wrap

919 + // this array in a keyed Fragment.

920 + return fragment;

921 + }

922 + // If the outer Server Component was implicit but then an inner one had a key

923 + // we don't actually need to be able to move the whole set around. It'll always be

924 + // in an implicit slot. The key only exists to be able to reset the state of the

925 + // children. We could achieve the same effect by passing on the keyPath to the next

926 + // set of components inside the fragment. This would also allow a keyless fragment

927 + // reconcile against a single child.

928 + // Unfortunately because of JSON.stringify, we can't call the recursive loop for

929 + // each child within this context because we can't return a set with already resolved

930 + // values. E.g. a string would get double encoded. Returning would pop the context.

931 + // So instead, we wrap it with an unkeyed fragment and inner keyed fragment.

932 + return [fragment];

933 + }

934 + // Since we're yielding here, that implicitly resets the keyPath context on the

935 + // way up. Which is what we want since we've consumed it. If this changes to

936 + // be recursive serialization, we need to reset the keyPath and implicitSlot,

937 + // before recursing here.

938 if (__DEV__) {

939 const debugInfo: ?ReactDebugInfo = (children: any)._debugInfo;

940 if (debugInfo) {

@@ -705,12 +949,21 @@ function renderFragment(

949 // from the server by the time we emit it.

950 forwardDebugInfo(request, debugID, debugInfo);

951 }

952 + // Since we're rendering this array again, create a copy that doesn't

953 + // have the debug info so we avoid outlining or emitting debug info again.

954 + children = Array.from(children);

955 }

956 }

710 - if (!enableServerComponentKeys) {

711 - return children;

712 - }

713 - if (task.keyPath !== null) {

957 + return children;

958 +}

959 +

960 +function renderAsyncFragment(

961 + request: Request,

962 + task: Task,

963 + children: $AsyncIterable<ReactClientValue, ReactClientValue, void>,

964 + getAsyncIterator: () => $AsyncIterator<any, any, any>,

965 +): ReactJSONValue {

966 + if (enableServerComponentKeys && task.keyPath !== null) {

967 // We have a Server Component that specifies a key but we're now splitting

968 // the tree using a fragment.

969 const fragment = [

@@ -737,11 +990,13 @@ function renderFragment(

990 // So instead, we wrap it with an unkeyed fragment and inner keyed fragment.

991 return [fragment];

992 }

993 +

994 // Since we're yielding here, that implicitly resets the keyPath context on the

995 // way up. Which is what we want since we've consumed it. If this changes to

996 // be recursive serialization, we need to reset the keyPath and implicitSlot,

997 // before recursing here.

744 - return children;

998 + const asyncIterator = getAsyncIterator.call(children);

999 + return serializeAsyncIterable(request, task, children, asyncIterator);

1000 }

1001

1002 function renderClientElement(

@@ -1156,13 +1411,9 @@ function serializeTemporaryReference(

1411 }

1412

1413 function serializeLargeTextString(request: Request, text: string): string {

1159 - request.pendingChunks += 2;

1414 + request.pendingChunks++;

1415 const textId = request.nextChunkId++;

1161 - const textChunk = stringToChunk(text);

1162 - const binaryLength = byteLengthOfChunk(textChunk);

1163 - const row = textId.toString(16) + ':T' + binaryLength.toString(16) + ',';

1164 - const headerChunk = stringToChunk(row);

1165 - request.completedRegularChunks.push(headerChunk, textChunk);

1416 + emitTextChunk(request, textId, text);

1417 return serializeByValueID(textId);

1418 }

1419

@@ -1212,27 +1463,9 @@ function serializeTypedArray(

1463 tag: string,

1464 typedArray: $ArrayBufferView,

1465 ): string {

1215 - if (enableTaint) {

1216 - if (TaintRegistryByteLengths.has(typedArray.byteLength)) {

1217 - // If we have had any tainted values of this length, we check

1218 - // to see if these bytes matches any entries in the registry.

1219 - const tainted = TaintRegistryValues.get(

1220 - binaryToComparableString(typedArray),

1221 - );

1222 - if (tainted !== undefined) {

1223 - throwTaintViolation(tainted.message);

1224 - }

1225 - }

1226 - }

1227 - request.pendingChunks += 2;

1466 + request.pendingChunks++;

1467 const bufferId = request.nextChunkId++;

1229 - // TODO: Convert to little endian if that's not the server default.

1230 - const binaryChunk = typedArrayToBinaryChunk(typedArray);

1231 - const binaryLength = byteLengthOfBinaryChunk(binaryChunk);

1232 - const row =

1233 - bufferId.toString(16) + ':' + tag + binaryLength.toString(16) + ',';

1234 - const headerChunk = stringToChunk(row);

1235 - request.completedRegularChunks.push(headerChunk, binaryChunk);

1468 + emitTypedArrayChunk(request, bufferId, tag, typedArray);

1469 return serializeByValueID(bufferId);

1470 }

1471

@@ -1248,10 +1481,16 @@ function serializeBlob(request: Request, blob: Blob): string {

1481

1482 const reader = blob.stream().getReader();

1483

1484 + let aborted = false;

1485 function progress(

1486 entry: {done: false, value: Uint8Array} | {done: true, value: void},

1487 ): Promise<void> | void {

1488 + if (aborted) {

1489 + return;

1490 + }

1491 if (entry.done) {

1492 + request.abortListeners.delete(error);

1493 + aborted = true;

1494 pingTask(request, newTask);

1495 return;

1496 }

@@ -1262,13 +1501,21 @@ function serializeBlob(request: Request, blob: Blob): string {

1501 }

1502

1503 function error(reason: mixed) {

1504 + if (aborted) {

1505 + return;

1506 + }

1507 + aborted = true;

1508 + request.abortListeners.delete(error);

1509 const digest = logRecoverableError(request, reason);

1510 emitErrorChunk(request, newTask.id, digest, reason);

1511 request.abortableTasks.delete(newTask);

1268 - if (request.destination !== null) {

1269 - flushCompletedChunks(request, request.destination);

1270 - }

1512 + enqueueFlush(request);

1513 + // $FlowFixMe should be able to pass mixed

1514 + reader.cancel(reason).then(error, error);

1515 }

1516 +

1517 + request.abortListeners.add(error);

1518 +

1519 // $FlowFixMe[incompatible-call]

1520 reader.read().then(progress).catch(error);

1521

@@ -1667,6 +1914,27 @@ function renderModelDestructive(

1914 return renderFragment(request, task, Array.from((value: any)));

1915 }

1916

1917 + if (enableFlightReadableStream) {

1918 + // TODO: Blob is not available in old Node. Remove the typeof check later.

1919 + if (

1920 + typeof ReadableStream === 'function' &&

1921 + value instanceof ReadableStream

1922 + ) {

1923 + return serializeReadableStream(request, task, value);

1924 + }

1925 + const getAsyncIterator: void | (() => $AsyncIterator<any, any, any>) =

1926 + (value: any)[ASYNC_ITERATOR];

1927 + if (typeof getAsyncIterator === 'function') {

1928 + // We treat AsyncIterables as a Fragment and as such we might need to key them.

1929 + return renderAsyncFragment(

1930 + request,

1931 + task,

1932 + (value: any),

1933 + getAsyncIterator,

1934 + );

1935 + }

1936 + }

1937 +

1938 // Verify that this is a simple plain object.

1939 const proto = getPrototypeOf(value);

1940 if (

@@ -2031,6 +2299,42 @@ function emitDebugChunk(

2299 request.completedRegularChunks.push(processedChunk);

2300 }

2301

2302 +function emitTypedArrayChunk(

2303 + request: Request,

2304 + id: number,

2305 + tag: string,

2306 + typedArray: $ArrayBufferView,

2307 +): void {

2308 + if (enableTaint) {

2309 + if (TaintRegistryByteLengths.has(typedArray.byteLength)) {

2310 + // If we have had any tainted values of this length, we check

2311 + // to see if these bytes matches any entries in the registry.

2312 + const tainted = TaintRegistryValues.get(

2313 + binaryToComparableString(typedArray),

2314 + );

2315 + if (tainted !== undefined) {

2316 + throwTaintViolation(tainted.message);

2317 + }

2318 + }

2319 + }

2320 + request.pendingChunks++; // Extra chunk for the header.

2321 + // TODO: Convert to little endian if that's not the server default.

2322 + const binaryChunk = typedArrayToBinaryChunk(typedArray);

2323 + const binaryLength = byteLengthOfBinaryChunk(binaryChunk);

2324 + const row = id.toString(16) + ':' + tag + binaryLength.toString(16) + ',';

2325 + const headerChunk = stringToChunk(row);

2326 + request.completedRegularChunks.push(headerChunk, binaryChunk);

2327 +}

2328 +

2329 +function emitTextChunk(request: Request, id: number, text: string): void {

2330 + request.pendingChunks++; // Extra chunk for the header.

2331 + const textChunk = stringToChunk(text);

2332 + const binaryLength = byteLengthOfChunk(textChunk);

2333 + const row = id.toString(16) + ':T' + binaryLength.toString(16) + ',';

2334 + const headerChunk = stringToChunk(row);

2335 + request.completedRegularChunks.push(headerChunk, textChunk);

2336 +}

2337 +

2338 function serializeEval(source: string): string {

2339 if (!__DEV__) {

2340 // These errors should never make it into a build so we don't need to encode them in codes.json

@@ -2391,6 +2695,96 @@ function forwardDebugInfo(

2695 }

2696 }

2697

2698 +function emitChunk(

2699 + request: Request,

2700 + task: Task,

2701 + value: ReactClientValue,

2702 +): void {

2703 + const id = task.id;

2704 + // For certain types we have special types, we typically outlined them but

2705 + // we can emit them directly for this row instead of through an indirection.

2706 + if (typeof value === 'string') {

2707 + if (enableTaint) {

2708 + const tainted = TaintRegistryValues.get(value);

2709 + if (tainted !== undefined) {

2710 + throwTaintViolation(tainted.message);

2711 + }

2712 + }

2713 + emitTextChunk(request, id, value);

2714 + return;

2715 + }

2716 + if (enableBinaryFlight) {

2717 + if (value instanceof ArrayBuffer) {

2718 + emitTypedArrayChunk(request, id, 'A', new Uint8Array(value));

2719 + return;

2720 + }

2721 + if (value instanceof Int8Array) {

2722 + // char

2723 + emitTypedArrayChunk(request, id, 'O', value);

2724 + return;

2725 + }

2726 + if (value instanceof Uint8Array) {

2727 + // unsigned char

2728 + emitTypedArrayChunk(request, id, 'o', value);

2729 + return;

2730 + }

2731 + if (value instanceof Uint8ClampedArray) {

2732 + // unsigned clamped char

2733 + emitTypedArrayChunk(request, id, 'U', value);

2734 + return;

2735 + }

2736 + if (value instanceof Int16Array) {

2737 + // sort

2738 + emitTypedArrayChunk(request, id, 'S', value);

2739 + return;

2740 + }

2741 + if (value instanceof Uint16Array) {

2742 + // unsigned short

2743 + emitTypedArrayChunk(request, id, 's', value);

2744 + return;

2745 + }

2746 + if (value instanceof Int32Array) {

2747 + // long

2748 + emitTypedArrayChunk(request, id, 'L', value);

2749 + return;

2750 + }

2751 + if (value instanceof Uint32Array) {

2752 + // unsigned long

2753 + emitTypedArrayChunk(request, id, 'l', value);

2754 + return;

2755 + }

2756 + if (value instanceof Float32Array) {

2757 + // float

2758 + emitTypedArrayChunk(request, id, 'G', value);

2759 + return;

2760 + }

2761 + if (value instanceof Float64Array) {

2762 + // double

2763 + emitTypedArrayChunk(request, id, 'g', value);

2764 + return;

2765 + }

2766 + if (value instanceof BigInt64Array) {

2767 + // number

2768 + emitTypedArrayChunk(request, id, 'M', value);

2769 + return;

2770 + }

2771 + if (value instanceof BigUint64Array) {

2772 + // unsigned number

2773 + // We use "m" instead of "n" since JSON can start with "null"

2774 + emitTypedArrayChunk(request, id, 'm', value);

2775 + return;

2776 + }

2777 + if (value instanceof DataView) {

2778 + emitTypedArrayChunk(request, id, 'V', value);

2779 + return;

2780 + }

2781 + }

2782 + // For anything else we need to try to serialize it using JSON.

2783 + // $FlowFixMe[incompatible-type] stringify can return null for undefined but we never do

2784 + const json: string = stringify(value, task.toJSON);

2785 + emitModelChunk(request, task.id, json);

2786 +}

2787 +

2788 const emptyRoot = {};

2789

2790 function retryTask(request: Request, task: Task): void {

@@ -2435,19 +2829,17 @@ function retryTask(request: Request, task: Task): void {

2829 task.keyPath = null;

2830 task.implicitSlot = false;

2831

2438 - let json: string;

2832 if (typeof resolvedModel === 'object' && resolvedModel !== null) {

2833 // Object might contain unresolved values like additional elements.

2834 // This is simulating what the JSON loop would do if this was part of it.

2442 - // $FlowFixMe[incompatible-type] stringify can return null for undefined but we never do

2443 - json = stringify(resolvedModel, task.toJSON);

2835 + emitChunk(request, task, resolvedModel);

2836 } else {

2837 // If the value is a string, it means it's a terminal value and we already escaped it

2838 // We don't need to escape it again so it's not passed the toJSON replacer.

2839 // $FlowFixMe[incompatible-type] stringify can return null for undefined but we never do

2448 - json = stringify(resolvedModel);

2840 + const json: string = stringify(resolvedModel);

2841 + emitModelChunk(request, task.id, json);

2842 }

2450 - emitModelChunk(request, task.id, json);

2843

2844 request.abortableTasks.delete(task);

2845 task.status = COMPLETED;

@@ -2489,6 +2881,24 @@ function retryTask(request: Request, task: Task): void {

2881 }

2882 }

2883

2884 +function tryStreamTask(request: Request, task: Task): void {

2885 + // This is used to try to emit something synchronously but if it suspends,

2886 + // we emit a reference to a new outlined task immediately instead.

2887 + const prevDebugID = debugID;

2888 + if (__DEV__) {

2889 + // We don't use the id of the stream task for debugID. Instead we leave it null

2890 + // so that we instead outline the row to get a new debugID if needed.

2891 + debugID = null;

2892 + }

2893 + try {

2894 + emitChunk(request, task, task.model);

2895 + } finally {

2896 + if (__DEV__) {

2897 + debugID = prevDebugID;

2898 + }

2899 + }

2900 +}

2901 +

2902 function performWork(request: Request): void {

2903 const prevDispatcher = ReactSharedInternals.H;

2904 ReactSharedInternals.H = HooksDispatcher;

@@ -2603,6 +3013,7 @@ function flushCompletedChunks(

3013 cleanupTaintQueue(request);

3014 }

3015 close(destination);

3016 + request.destination = null;

3017 }

3018 }

3019

@@ -2660,9 +3071,9 @@ export function stopFlowing(request: Request): void {

3071 export function abort(request: Request, reason: mixed): void {

3072 try {

3073 const abortableTasks = request.abortableTasks;

3074 + // We have tasks to abort. We'll emit one error row and then emit a reference

3075 + // to that row from every row that's still remaining.

3076 if (abortableTasks.size > 0) {

2664 - // We have tasks to abort. We'll emit one error row and then emit a reference

2665 - // to that row from every row that's still remaining.

3077 request.pendingChunks++;

3078 const errorId = request.nextChunkId++;

3079 if (

@@ -2687,6 +3098,30 @@ export function abort(request: Request, reason: mixed): void {

3098 abortableTasks.forEach(task => abortTask(task, request, errorId));

3099 abortableTasks.clear();

3100 }

3101 + const abortListeners = request.abortListeners;

3102 + if (abortListeners.size > 0) {

3103 + let error;

3104 + if (

3105 + enablePostpone &&

3106 + typeof reason === 'object' &&

3107 + reason !== null &&

3108 + (reason: any).$$typeof === REACT_POSTPONE_TYPE

3109 + ) {

3110 + // We aborted with a Postpone but since we're passing this to an

3111 + // external handler, passing this object would leak it outside React.

3112 + // We create an alternative reason for it instead.

3113 + error = new Error('The render was aborted due to being postponed.');

3114 + } else {

3115 + error =

3116 + reason === undefined

3117 + ? new Error(

3118 + 'The render was aborted by the server without a reason.',

3119 + )

3120 + : reason;

3121 + }

3122 + abortListeners.forEach(callback => callback(error));

3123 + abortListeners.clear();

3124 + }

3125 if (request.destination !== null) {

3126 flushCompletedChunks(request, request.destination);

3127 }

packages/react-server/src/ReactFlightThenable.js

+13 -10

@@ -82,6 +82,9 @@ export function trackUsedThenable<T>(

82 // Only instrument the thenable if the status if not defined. If

83 // it's defined, but an unknown value, assume it's been instrumented by

84 // some custom userspace implementation. We treat it as "pending".

85 + // Attach a dummy listener, to ensure that any lazy initialization can

86 + // happen. Flight lazily parses JSON when the value is actually awaited.

87 + thenable.then(noop, noop);

88 } else {

89 const pendingThenable: PendingThenable<T> = (thenable: any);

90 pendingThenable.status = 'pending';

@@ -101,17 +104,17 @@ export function trackUsedThenable<T>(

104 }

105 },

106 );

107 + }

108

105 - // Check one more time in case the thenable resolved synchronously

106 - switch (thenable.status) {

107 - case 'fulfilled': {

108 - const fulfilledThenable: FulfilledThenable<T> = (thenable: any);

109 - return fulfilledThenable.value;

110 - }

111 - case 'rejected': {

112 - const rejectedThenable: RejectedThenable<T> = (thenable: any);

113 - throw rejectedThenable.reason;

114 - }

109 + // Check one more time in case the thenable resolved synchronously

110 + switch (thenable.status) {

111 + case 'fulfilled': {

112 + const fulfilledThenable: FulfilledThenable<T> = (thenable: any);

113 + return fulfilledThenable.value;

114 + }

115 + case 'rejected': {

116 + const rejectedThenable: RejectedThenable<T> = (thenable: any);

117 + throw rejectedThenable.reason;

118 }

119 }

120

packages/shared/ReactFeatureFlags.js

@@ -81,6 +81,7 @@ export const enableLegacyCache = __EXPERIMENTAL__;

81 export const enableFetchInstrumentation = true;

83 export const enableBinaryFlight = __EXPERIMENTAL__;

84 +export const enableFlightReadableStream = __EXPERIMENTAL__;

86 export const enableTaint = __EXPERIMENTAL__;

packages/shared/forks/ReactFeatureFlags.native-fb.js

@@ -44,6 +44,7 @@ export const enableCache = true;

44 export const enableLegacyCache = false;

45 export const enableFetchInstrumentation = false;

46 export const enableBinaryFlight = true;

47 +export const enableFlightReadableStream = true;

48 export const enableTaint = true;

49 export const enablePostpone = false;

50 export const debugRenderPhaseSideEffectsForStrictMode = __DEV__;

packages/shared/forks/ReactFeatureFlags.native-oss.js

@@ -52,6 +52,7 @@ export const enableTaint = __NEXT_RN_MAJOR__;

52 export const enableUnifiedSyncLane = __NEXT_RN_MAJOR__;

53 export const enableFizzExternalRuntime = __NEXT_RN_MAJOR__; // DOM-only

54 export const enableBinaryFlight = __NEXT_RN_MAJOR__; // DOM-only

55 +export const enableFlightReadableStream = __NEXT_RN_MAJOR__; // DOM-only

56 export const enableServerComponentKeys = __NEXT_RN_MAJOR__;

57 export const enableServerComponentLogs = __NEXT_RN_MAJOR__;

packages/shared/forks/ReactFeatureFlags.test-renderer.js

@@ -22,6 +22,7 @@ export const enableCache = true;

22 export const enableLegacyCache = __EXPERIMENTAL__;

23 export const enableFetchInstrumentation = true;

24 export const enableBinaryFlight = true;

25 +export const enableFlightReadableStream = true;

26 export const enableTaint = true;

27 export const enablePostpone = false;

28 export const disableCommentsAsDOMContainers = true;

packages/shared/forks/ReactFeatureFlags.test-renderer.native-fb.js

@@ -22,6 +22,7 @@ export const enableCache = true;

22 export const enableLegacyCache = false;

23 export const enableFetchInstrumentation = false;

24 export const enableBinaryFlight = true;

25 +export const enableFlightReadableStream = true;

26 export const enableTaint = true;

27 export const enablePostpone = false;

28 export const disableCommentsAsDOMContainers = true;

packages/shared/forks/ReactFeatureFlags.test-renderer.www.js

@@ -22,6 +22,7 @@ export const enableCache = true;

22 export const enableLegacyCache = true;

23 export const enableFetchInstrumentation = false;

24 export const enableBinaryFlight = true;

25 +export const enableFlightReadableStream = true;

26 export const enableTaint = true;

27 export const enablePostpone = false;

28 export const disableCommentsAsDOMContainers = true;

packages/shared/forks/ReactFeatureFlags.www.js

@@ -69,6 +69,8 @@ export const enableLegacyCache = true;

69 export const enableFetchInstrumentation = false;

71 export const enableBinaryFlight = false;

72 +export const enableFlightReadableStream = false;

73 +

74 export const enableTaint = false;

76 export const enablePostpone = false;

scripts/error-codes/codes.json

+3 -1

@@ -507,5 +507,7 @@

507 "519": "Hydration Mismatch Exception: This is not a real error, and should not leak into userspace. If you're seeing this, it's likely a bug in React.",

508 "520": "There was an error during concurrent rendering but React was able to recover by instead synchronously rendering the entire root.",

509 "521": "flushSyncWork should not be called from builds that support legacy mode. This is a bug in React.",

510 - "522": "Invalid form element. requestFormReset must be passed a form that was rendered by React."

510 + "522": "Invalid form element. requestFormReset must be passed a form that was rendered by React.",

511 + "523": "The render was aborted due to being postponed.",

512 + "524": "Values cannot be passed to next() of AsyncIterables passed to Client Components."

513 }