[Fizz/Flight] Pass in Destination lazily to startFlowing instead of in createRequest (#22449)

* Pass in Destination lazily in startFlowing instead of createRequest

* Delay fatal errors until we have a destination to forward them to

* Flow can now be inferred by whether there's a destination set

We can drop the destination when we're not flowing since there's nothing to
write to.

Fatal errors now close once flowing starts back up again.

* Defer fatal errors in Flight too

Author: sebmarkbage

packages/react-dom/src/__tests__/ReactDOMFizzServerNode-test.js

@@ -154,7 +154,7 @@ describe('ReactDOMFizzServer', () => {
   it('should error the stream when an error is thrown at the root', async () => {
     const reportedErrors = [];
     const {writable, output, completed} = getTestWritable();
-    ReactDOMFizzServer.pipeToNodeWritable(
+    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
       <div>
         <Throw />
       </div>,
@@ -166,7 +166,8 @@ describe('ReactDOMFizzServer', () => {
       },
     );
 
-    // The stream is errored even if we haven't started writing.
+    // The stream is errored once we start writing.
+    startWriting();
 
     await completed;
 

packages/react-dom/src/server/ReactDOMFizzServerBrowser.js

@@ -37,7 +37,15 @@ function renderToReadableStream(
   children: ReactNodeList,
   options?: Options,
 ): ReadableStream {
-  let request;
+  const request = createRequest(
+    children,
+    createResponseState(options ? options.identifierPrefix : undefined),
+    createRootFormatContext(options ? options.namespaceURI : undefined),
+    options ? options.progressiveChunkSize : undefined,
+    options ? options.onError : undefined,
+    options ? options.onCompleteAll : undefined,
+    options ? options.onCompleteShell : undefined,
+  );
   if (options && options.signal) {
     const signal = options.signal;
     const listener = () => {
@@ -48,16 +56,6 @@ function renderToReadableStream(
   }
   const stream = new ReadableStream({
     start(controller) {
-      request = createRequest(
-        children,
-        controller,
-        createResponseState(options ? options.identifierPrefix : undefined),
-        createRootFormatContext(options ? options.namespaceURI : undefined),
-        options ? options.progressiveChunkSize : undefined,
-        options ? options.onError : undefined,
-        options ? options.onCompleteAll : undefined,
-        options ? options.onCompleteShell : undefined,
-      );
       startWork(request);
     },
     pull(controller) {
@@ -66,7 +64,7 @@ function renderToReadableStream(
       // is actually used by something so we can give it the best result possible
       // at that point.
       if (stream.locked) {
-        startFlowing(request);
+        startFlowing(request, controller);
       }
     },
     cancel(reason) {},

packages/react-dom/src/server/ReactDOMFizzServerNode.js

@@ -25,7 +25,7 @@ import {
 } from './ReactDOMServerFormatConfig';
 
 function createDrainHandler(destination, request) {
-  return () => startFlowing(request);
+  return () => startFlowing(request, destination);
 }
 
 type Options = {|
@@ -44,14 +44,9 @@ type Controls = {|
   startWriting(): void,
 |};
 
-function createRequestImpl(
-  children: ReactNodeList,
-  destination: Writable,
-  options: void | Options,
-) {
+function createRequestImpl(children: ReactNodeList, options: void | Options) {
   return createRequest(
     children,
-    destination,
     createResponseState(options ? options.identifierPrefix : undefined),
     createRootFormatContext(options ? options.namespaceURI : undefined),
     options ? options.progressiveChunkSize : undefined,
@@ -66,7 +61,7 @@ function pipeToNodeWritable(
   destination: Writable,
   options?: Options,
 ): Controls {
-  const request = createRequestImpl(children, destination, options);
+  const request = createRequestImpl(children, options);
   let hasStartedFlowing = false;
   startWork(request);
   return {
@@ -75,7 +70,7 @@ function pipeToNodeWritable(
         return;
       }
       hasStartedFlowing = true;
-      startFlowing(request);
+      startFlowing(request, destination);
       destination.on('drain', createDrainHandler(destination, request));
     },
     abort() {

packages/react-dom/src/server/ReactDOMLegacyServerBrowser.js

@@ -59,7 +59,6 @@ function renderToStringImpl(
   }
   const request = createRequest(
     children,
-    destination,
     createResponseState(
       generateStaticMarkup,
       options ? options.identifierPrefix : undefined,
@@ -74,7 +73,7 @@ function renderToStringImpl(
   // If anything suspended and is still pending, we'll abort it before writing.
   // That way we write only client-rendered boundaries from the start.
   abort(request);
-  startFlowing(request);
+  startFlowing(request, destination);
   if (didFatal) {
     throw fatalError;
   }

packages/react-dom/src/server/ReactDOMLegacyServerNode.js

@@ -54,7 +54,7 @@ class ReactMarkupReadableStream extends Readable {
 
   _read(size) {
     if (this.startedFlowing) {
-      startFlowing(this.request);
+      startFlowing(this.request, this);
     }
   }
 }
@@ -72,12 +72,11 @@ function renderToNodeStreamImpl(
     // We wait until everything has loaded before starting to write.
     // That way we only end up with fully resolved HTML even if we suspend.
     destination.startedFlowing = true;
-    startFlowing(request);
+    startFlowing(request, destination);
   }
   const destination = new ReactMarkupReadableStream();
   const request = createRequest(
     children,
-    destination,
     createResponseState(false, options ? options.identifierPrefix : undefined),
     createRootFormatContext(),
     Infinity,

packages/react-noop-renderer/src/ReactNoopFlightServer.js

@@ -63,12 +63,11 @@ function render(model: ReactModel, options?: Options): Destination {
   const bundlerConfig = undefined;
   const request = ReactNoopFlightServer.createRequest(
     model,
-    destination,
     bundlerConfig,
     options ? options.onError : undefined,
   );
   ReactNoopFlightServer.startWork(request);
-  ReactNoopFlightServer.startFlowing(request);
+  ReactNoopFlightServer.startFlowing(request, destination);
   return destination;
 }
 

packages/react-noop-renderer/src/ReactNoopServer.js

@@ -259,7 +259,6 @@ function render(children: React$Element<any>, options?: Options): Destination {
   };
   const request = ReactNoopServer.createRequest(
     children,
-    destination,
     null,
     null,
     options ? options.progressiveChunkSize : undefined,
@@ -268,7 +267,7 @@ function render(children: React$Element<any>, options?: Options): Destination {
     options ? options.onCompleteShell : undefined,
   );
   ReactNoopServer.startWork(request);
-  ReactNoopServer.startFlowing(request);
+  ReactNoopServer.startFlowing(request, destination);
   return destination;
 }
 

packages/react-server-dom-relay/src/ReactDOMServerFB.js

@@ -46,7 +46,6 @@ function renderToStream(children: ReactNodeList, options: Options): Stream {
   };
   const request = createRequest(
     children,
-    destination,
     createResponseState(options ? options.identifierPrefix : undefined),
     createRootFormatContext(undefined),
     options ? options.progressiveChunkSize : undefined,
@@ -71,7 +70,7 @@ function abortStream(stream: Stream): void {
 function renderNextChunk(stream: Stream): string {
   const {request, destination} = stream;
   performWork(request);
-  startFlowing(request);
+  startFlowing(request, destination);
   if (destination.fatal) {
     throw destination.error;
   }

packages/react-server-dom-relay/src/ReactFlightDOMRelayServer.js

@@ -31,12 +31,11 @@ function render(
 ): void {
   const request = createRequest(
     model,
-    destination,
     config,
     options ? options.onError : undefined,
   );
   startWork(request);
-  startFlowing(request);
+  startFlowing(request, destination);
 }
 
 export {render};

packages/react-server-dom-webpack/src/ReactFlightDOMServerBrowser.js

@@ -25,15 +25,13 @@ function renderToReadableStream(
   webpackMap: BundlerConfig,
   options?: Options,
 ): ReadableStream {
-  let request;
+  const request = createRequest(
+    model,
+    webpackMap,
+    options ? options.onError : undefined,
+  );
   const stream = new ReadableStream({
     start(controller) {
-      request = createRequest(
-        model,
-        controller,
-        webpackMap,
-        options ? options.onError : undefined,
-      );
       startWork(request);
     },
     pull(controller) {
@@ -42,7 +40,7 @@ function renderToReadableStream(
       // is actually used by something so we can give it the best result possible
       // at that point.
       if (stream.locked) {
-        startFlowing(request);
+        startFlowing(request, controller);
       }
     },
     cancel(reason) {},

packages/react-server-dom-webpack/src/ReactFlightDOMServerNode.js

@@ -18,7 +18,7 @@ import {
 } from 'react-server/src/ReactFlightServer';
 
 function createDrainHandler(destination, request) {
-  return () => startFlowing(request);
+  return () => startFlowing(request, destination);
 }
 
 type Options = {
@@ -33,12 +33,11 @@ function pipeToNodeWritable(
 ): void {
   const request = createRequest(
     model,
-    destination,
     webpackMap,
     options ? options.onError : undefined,
   );
   startWork(request);
-  startFlowing(request);
+  startFlowing(request, destination);
   destination.on('drain', createDrainHandler(destination, request));
 }
 

packages/react-server-native-relay/src/ReactFlightNativeRelayServer.js

@@ -24,9 +24,9 @@ function render(
   destination: Destination,
   config: BundlerConfig,
 ): void {
-  const request = createRequest(model, destination, config);
+  const request = createRequest(model, config);
   startWork(request);
-  startFlowing(request);
+  startFlowing(request, destination);
 }
 
 export {render};