[Fizz] Expose a method to explicitly start writing to a Node stream (#21028)

* Expose an explicit point when to start writing in the Node API

* Add a previously failing test

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

@@ -196,7 +196,7 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should asynchronously load the suspense boundary', async () => {
     await act(async () => {
-      ReactDOMFizzServer.pipeToNodeWritable(
+      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
         <div>
           <Suspense fallback={<Text text="Loading..." />}>
             <AsyncText text="Hello World" />
@@ -204,6 +204,7 @@ describe('ReactDOMFizzServer', () => {
         </div>,
         writable,
       );
+      startWriting();
     });
     expect(getVisibleChildren(container)).toEqual(<div>Loading...</div>);
     await act(async () => {
@@ -229,7 +230,11 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      ReactDOMFizzServer.pipeToNodeWritable(<App />, writable);
+      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+        <App />,
+        writable,
+      );
+      startWriting();
     });
 
     // We're still showing a fallback.
@@ -281,6 +286,7 @@ describe('ReactDOMFizzServer', () => {
     let controls;
     await act(async () => {
       controls = ReactDOMFizzServer.pipeToNodeWritable(<App />, writable);
+      controls.startWriting();
     });
 
     // We're still showing a fallback.

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

@@ -59,11 +59,33 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should call pipeToNodeWritable', () => {
     const {writable, output} = getTestWritable();
-    ReactDOMFizzServer.pipeToNodeWritable(<div>hello world</div>, writable);
+    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      <div>hello world</div>,
+      writable,
+    );
+    startWriting();
     jest.runAllTimers();
     expect(output.result).toBe('<div>hello world</div>');
   });
 
+  // @gate experimental
+  it('should start writing after startWriting', () => {
+    const {writable, output} = getTestWritable();
+    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      <div>hello world</div>,
+      writable,
+    );
+    jest.runAllTimers();
+    // First we write our header.
+    output.result +=
+      '<!doctype html><html><head><title>test</title><head><body>';
+    // Then React starts writing.
+    startWriting();
+    expect(output.result).toBe(
+      '<!doctype html><html><head><title>test</title><head><body><div>hello world</div>',
+    );
+  });
+
   // @gate experimental
   it('should error the stream when an error is thrown at the root', async () => {
     const {writable, output, completed} = getTestWritable();
@@ -74,6 +96,8 @@ describe('ReactDOMFizzServer', () => {
       writable,
     );
 
+    // The stream is errored even if we haven't started writing.
+
     await completed;
 
     expect(output.error).toBe(theError);
@@ -83,7 +107,7 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should error the stream when an error is thrown inside a fallback', async () => {
     const {writable, output, completed} = getTestWritable();
-    ReactDOMFizzServer.pipeToNodeWritable(
+    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
       <div>
         <Suspense fallback={<Throw />}>
           <InfiniteSuspend />
@@ -91,6 +115,7 @@ describe('ReactDOMFizzServer', () => {
       </div>,
       writable,
     );
+    startWriting();
 
     await completed;
 
@@ -101,7 +126,7 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should not error the stream when an error is thrown inside suspense boundary', async () => {
     const {writable, output, completed} = getTestWritable();
-    ReactDOMFizzServer.pipeToNodeWritable(
+    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
       <div>
         <Suspense fallback={<div>Loading</div>}>
           <Throw />
@@ -109,6 +134,7 @@ describe('ReactDOMFizzServer', () => {
       </div>,
       writable,
     );
+    startWriting();
 
     await completed;
 
@@ -128,12 +154,13 @@ describe('ReactDOMFizzServer', () => {
     function Content() {
       return 'Hi';
     }
-    ReactDOMFizzServer.pipeToNodeWritable(
+    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
       <Suspense fallback={<Fallback />}>
         <Content />
       </Suspense>,
       writable,
     );
+    startWriting();
 
     await completed;
 
@@ -145,7 +172,7 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should be able to complete by aborting even if the promise never resolves', async () => {
     const {writable, output, completed} = getTestWritable();
-    const {abort} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {startWriting, abort} = ReactDOMFizzServer.pipeToNodeWritable(
       <div>
         <Suspense fallback={<div>Loading</div>}>
           <InfiniteSuspend />
@@ -153,6 +180,7 @@ describe('ReactDOMFizzServer', () => {
       </div>,
       writable,
     );
+    startWriting();
 
     jest.runAllTimers();
 

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

@@ -32,9 +32,17 @@ function pipeToNodeWritable(
   destination: Writable,
 ): Controls {
   const request = createRequest(children, destination);
-  destination.on('drain', createDrainHandler(destination, request));
+  let hasStartedFlowing = false;
   startWork(request);
   return {
+    startWriting() {
+      if (hasStartedFlowing) {
+        return;
+      }
+      hasStartedFlowing = true;
+      startFlowing(request);
+      destination.on('drain', createDrainHandler(destination, request));
+    },
     abort() {
       abort(request);
     },

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

@@ -222,6 +222,7 @@ function render(children: React$Element<any>): Destination {
   };
   const request = ReactNoopServer.createRequest(children, destination);
   ReactNoopServer.startWork(request);
+  ReactNoopServer.startFlowing(request);
   return destination;
 }
 

packages/react-server/src/ReactFizzServer.js

@@ -894,8 +894,6 @@ function flushCompletedQueues(request: Request): void {
 // This would put all waiting boundaries into client-only mode.
 
 export function startWork(request: Request): void {
-  // TODO: Don't automatically start flowing. Expose an explicit signal. Auto-start once everything is done.
-  request.status = FLOWING;
   scheduleWork(() => performWork(request));
 }