From 20c081ec5194540b8a734059633b016855b60140 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sebastian=20Markb=C3=A5ge?= <sema@fb.com>
Date: Wed, 6 Oct 2021 00:31:06 -0400
Subject: [PATCH] [Fizz/Flight] pipeToNodeWritable(...,
 writable).startWriting() -> renderToPipeableStream(...).pipe(writable)
 (#22450)

* Rename pipeToNodeWritable to renderToNodePipe

* Add startWriting API to Flight

We don't really need it in this case because there's way less reason to
delay the stream in Flight.

* Pass the destination to startWriting instead of renderToNode

* Rename startWriting to pipe

This mirrors the ReadableStream API in Node

* Error codes

* Rename to renderToPipeableStream

This mimics the renderToReadableStream API for the browser.
---
 fixtures/flight/server/handler.server.js      |   8 +-
 fixtures/ssr/server/render.js                 |  30 ++--
 fixtures/ssr2/server/render.js                |   7 +-
 packages/react-dom/npm/server.node.js         |   2 +-
 packages/react-dom/server.node.js             |   4 +-
 .../src/__tests__/ReactDOMFizzServer-test.js  | 151 +++++++-----------
 .../__tests__/ReactDOMFizzServerNode-test.js  |  60 ++++---
 .../src/server/ReactDOMFizzServerNode.js      |  14 +-
 .../src/ReactFlightDOMServerNode.js           |  27 +++-
 .../src/__tests__/ReactFlightDOM-test.js      |  41 +++--
 scripts/error-codes/codes.json                |   3 +-
 11 files changed, 169 insertions(+), 178 deletions(-)

diff --git a/fixtures/flight/server/handler.server.js b/fixtures/flight/server/handler.server.js
index cd4af7319f01f..b6745c0332791 100644
--- a/fixtures/flight/server/handler.server.js
+++ b/fixtures/flight/server/handler.server.js
@@ -1,6 +1,6 @@
 'use strict';
 
-const {pipeToNodeWritable} = require('react-server-dom-webpack/writer');
+const {renderToPipeableStream} = require('react-server-dom-webpack/writer');
 const {readFile} = require('fs');
 const {resolve} = require('path');
 const React = require('react');
@@ -20,7 +20,11 @@ module.exports = function(req, res) {
         const App = m.default.default || m.default;
         res.setHeader('Access-Control-Allow-Origin', '*');
         const moduleMap = JSON.parse(data);
-        pipeToNodeWritable(React.createElement(App), res, moduleMap);
+        const {pipe} = renderToPipeableStream(
+          React.createElement(App),
+          moduleMap
+        );
+        pipe(res);
       }
     );
   });
diff --git a/fixtures/ssr/server/render.js b/fixtures/ssr/server/render.js
index ababbc4ccd72a..e0fc50f3f9538 100644
--- a/fixtures/ssr/server/render.js
+++ b/fixtures/ssr/server/render.js
@@ -1,5 +1,5 @@
 import React from 'react';
-import {pipeToNodeWritable} from 'react-dom/server';
+import {renderToPipeableStream} from 'react-dom/server';
 
 import App from '../src/components/App';
 
@@ -20,22 +20,18 @@ export default function render(url, res) {
     console.error('Fatal', error);
   });
   let didError = false;
-  const {startWriting, abort} = pipeToNodeWritable(
-    <App assets={assets} />,
-    res,
-    {
-      onCompleteShell() {
-        // If something errored before we started streaming, we set the error code appropriately.
-        res.statusCode = didError ? 500 : 200;
-        res.setHeader('Content-type', 'text/html');
-        startWriting();
-      },
-      onError(x) {
-        didError = true;
-        console.error(x);
-      },
-    }
-  );
+  const {pipe, abort} = renderToPipeableStream(<App assets={assets} />, {
+    onCompleteShell() {
+      // If something errored before we started streaming, we set the error code appropriately.
+      res.statusCode = didError ? 500 : 200;
+      res.setHeader('Content-type', 'text/html');
+      pipe(res);
+    },
+    onError(x) {
+      didError = true;
+      console.error(x);
+    },
+  });
   // Abandon and switch to client rendering after 5 seconds.
   // Try lowering this to see the client recover.
   setTimeout(abort, 5000);
diff --git a/fixtures/ssr2/server/render.js b/fixtures/ssr2/server/render.js
index b5d919e82ca6d..5ebbed0313c69 100644
--- a/fixtures/ssr2/server/render.js
+++ b/fixtures/ssr2/server/render.js
@@ -8,7 +8,7 @@
 
 import * as React from 'react';
 // import {renderToString} from 'react-dom/server';
-import {pipeToNodeWritable} from 'react-dom/server';
+import {renderToPipeableStream} from 'react-dom/server';
 import App from '../src/App';
 import {DataProvider} from '../src/data';
 import {API_DELAY, ABORT_DELAY} from './delays';
@@ -37,17 +37,16 @@ module.exports = function render(url, res) {
   });
   let didError = false;
   const data = createServerData();
-  const {startWriting, abort} = pipeToNodeWritable(
+  const {pipe, abort} = renderToPipeableStream(
     <DataProvider data={data}>
       <App assets={assets} />
     </DataProvider>,
-    res,
     {
       onCompleteShell() {
         // If something errored before we started streaming, we set the error code appropriately.
         res.statusCode = didError ? 500 : 200;
         res.setHeader('Content-type', 'text/html');
-        startWriting();
+        pipe(res);
       },
       onError(x) {
         didError = true;
diff --git a/packages/react-dom/npm/server.node.js b/packages/react-dom/npm/server.node.js
index e7d3251c4d3dd..2eb97e80354c8 100644
--- a/packages/react-dom/npm/server.node.js
+++ b/packages/react-dom/npm/server.node.js
@@ -14,4 +14,4 @@ exports.renderToString = l.renderToString;
 exports.renderToStaticMarkup = l.renderToStaticMarkup;
 exports.renderToNodeStream = l.renderToNodeStream;
 exports.renderToStaticNodeStream = l.renderToStaticNodeStream;
-exports.pipeToNodeWritable = s.pipeToNodeWritable;
+exports.renderToPipeableStream = s.renderToPipeableStream;
diff --git a/packages/react-dom/server.node.js b/packages/react-dom/server.node.js
index 7fb8cfaeb5a82..1fcab6c8c9ad4 100644
--- a/packages/react-dom/server.node.js
+++ b/packages/react-dom/server.node.js
@@ -36,8 +36,8 @@ export function renderToStaticNodeStream() {
   );
 }
 
-export function pipeToNodeWritable() {
-  return require('./src/server/ReactDOMFizzServerNode').pipeToNodeWritable.apply(
+export function renderToPipeableStream() {
+  return require('./src/server/ReactDOMFizzServerNode').renderToPipeableStream.apply(
     this,
     arguments,
   );
diff --git a/packages/react-dom/src/__tests__/ReactDOMFizzServer-test.js b/packages/react-dom/src/__tests__/ReactDOMFizzServer-test.js
index d786062043dee..cd0c87a1df3b2 100644
--- a/packages/react-dom/src/__tests__/ReactDOMFizzServer-test.js
+++ b/packages/react-dom/src/__tests__/ReactDOMFizzServer-test.js
@@ -239,7 +239,7 @@ describe('ReactDOMFizzServer', () => {
     };
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <div>
           <div>
             <Suspense fallback={<Text text="Loading..." />}>
@@ -252,9 +252,8 @@ describe('ReactDOMFizzServer', () => {
             </Suspense>
           </div>
         </div>,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
     expect(getVisibleChildren(container)).toEqual(
       <div>
@@ -304,16 +303,16 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <App isClient={false} />,
-        writable,
+
         {
           onError(x) {
             loggedErrors.push(x);
           },
         },
       );
-      startWriting();
+      pipe(writable);
     });
     expect(loggedErrors).toEqual([]);
 
@@ -356,15 +355,14 @@ describe('ReactDOMFizzServer', () => {
     });
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <div>
           <Suspense fallback={<Text text="Loading..." />}>
             {lazyElement}
           </Suspense>
         </div>,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
     expect(getVisibleChildren(container)).toEqual(<div>Loading...</div>);
     await act(async () => {
@@ -396,16 +394,16 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <App isClient={false} />,
-        writable,
+
         {
           onError(x) {
             loggedErrors.push(x);
           },
         },
       );
-      startWriting();
+      pipe(writable);
     });
     expect(loggedErrors).toEqual([]);
 
@@ -441,15 +439,14 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should asynchronously load the suspense boundary', async () => {
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <div>
           <Suspense fallback={<Text text="Loading..." />}>
             <AsyncText text="Hello World" />
           </Suspense>
         </div>,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
     expect(getVisibleChildren(container)).toEqual(<div>Loading...</div>);
     await act(async () => {
@@ -475,11 +472,8 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
-        <App />,
-        writable,
-      );
-      startWriting();
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(<App />);
+      pipe(writable);
     });
 
     // We're still showing a fallback.
@@ -550,16 +544,16 @@ describe('ReactDOMFizzServer', () => {
 
     // We originally suspend the boundary and start streaming the loading state.
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <App />,
-        writable,
+
         {
           onError(x) {
             loggedErrors.push(x);
           },
         },
       );
-      startWriting();
+      pipe(writable);
     });
 
     // We're still showing a fallback.
@@ -633,11 +627,10 @@ describe('ReactDOMFizzServer', () => {
 
     // We originally suspend the boundary and start streaming the loading state.
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <App showMore={false} />,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
 
     const root = ReactDOM.createRoot(container, {hydrate: true});
@@ -701,8 +694,8 @@ describe('ReactDOMFizzServer', () => {
 
     let controls;
     await act(async () => {
-      controls = ReactDOMFizzServer.pipeToNodeWritable(<App />, writable);
-      controls.startWriting();
+      controls = ReactDOMFizzServer.renderToPipeableStream(<App />);
+      controls.pipe(writable);
     });
 
     // We're still showing a fallback.
@@ -753,7 +746,7 @@ describe('ReactDOMFizzServer', () => {
     };
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         // We use two nested boundaries to flush out coverage of an old reentrancy bug.
         <Suspense fallback="Loading...">
           <Suspense fallback={<Text text="Loading A..." />}>
@@ -765,12 +758,11 @@ describe('ReactDOMFizzServer', () => {
             </>
           </Suspense>
         </Suspense>,
-        writableA,
         {
           identifierPrefix: 'A_',
           onCompleteShell() {
             writableA.write('<div id="container-A">');
-            startWriting();
+            pipe(writableA);
             writableA.write('</div>');
           },
         },
@@ -778,19 +770,18 @@ describe('ReactDOMFizzServer', () => {
     });
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <Suspense fallback={<Text text="Loading B..." />}>
           <Text text="This will show B: " />
           <div>
             <AsyncText text="B" />
           </div>
         </Suspense>,
-        writableB,
         {
           identifierPrefix: 'B_',
           onCompleteShell() {
             writableB.write('<div id="container-B">');
-            startWriting();
+            pipe(writableB);
             writableB.write('</div>');
           },
         },
@@ -876,11 +867,8 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
-        <App />,
-        writable,
-      );
-      startWriting();
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(<App />);
+      pipe(writable);
     });
 
     expect(getVisibleChildren(container)).toEqual(
@@ -967,11 +955,8 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
-        <App />,
-        writable,
-      );
-      startWriting();
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(<App />);
+      pipe(writable);
     });
 
     expect(getVisibleChildren(container)).toEqual(
@@ -1024,14 +1009,14 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <App />,
-        writable,
+
         {
           namespaceURI: 'http://www.w3.org/2000/svg',
           onCompleteShell() {
             writable.write('<svg>');
-            startWriting();
+            pipe(writable);
             writable.write('</svg>');
           },
         },
@@ -1111,11 +1096,8 @@ describe('ReactDOMFizzServer', () => {
 
     try {
       await act(async () => {
-        const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
-          <A />,
-          writable,
-        );
-        startWriting();
+        const {pipe} = ReactDOMFizzServer.renderToPipeableStream(<A />);
+        pipe(writable);
       });
 
       expect(getVisibleChildren(container)).toEqual(
@@ -1213,7 +1195,7 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <TestProvider ctx="A">
           <div>
             <Suspense fallback={[<Text text="Loading: " />, <TestConsumer />]}>
@@ -1224,9 +1206,8 @@ describe('ReactDOMFizzServer', () => {
             </Suspense>
           </div>
         </TestProvider>,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
     expect(getVisibleChildren(container)).toEqual(
       <div>
@@ -1272,7 +1253,7 @@ describe('ReactDOMFizzServer', () => {
     }
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <div>
           <PrintA />
           <div>
@@ -1287,9 +1268,8 @@ describe('ReactDOMFizzServer', () => {
           </div>
           <PrintA />
         </div>,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
     expect(getVisibleChildren(container)).toEqual(
       <div>
@@ -1335,7 +1315,7 @@ describe('ReactDOMFizzServer', () => {
 
     const loggedErrors = [];
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <div>
           <PrintA />
           <div>
@@ -1355,14 +1335,14 @@ describe('ReactDOMFizzServer', () => {
           </div>
           <PrintA />
         </div>,
-        writable,
+
         {
           onError(x) {
             loggedErrors.push(x);
           },
         },
       );
-      startWriting();
+      pipe(writable);
     });
     expect(loggedErrors.length).toBe(1);
     expect(loggedErrors[0].message).toEqual('A0.1.1');
@@ -1396,16 +1376,16 @@ describe('ReactDOMFizzServer', () => {
     const loggedErrors = [];
     let controls;
     await act(async () => {
-      controls = ReactDOMFizzServer.pipeToNodeWritable(
+      controls = ReactDOMFizzServer.renderToPipeableStream(
         <App isClient={false} />,
-        writable,
+
         {
           onError(x) {
             loggedErrors.push(x);
           },
         },
       );
-      controls.startWriting();
+      controls.pipe(writable);
     });
 
     // We're still showing a fallback.
@@ -1456,7 +1436,7 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should be able to abort the fallback if the main content finishes first', async () => {
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <Suspense fallback={<Text text="Loading Outer" />}>
           <div>
             <Suspense
@@ -1470,9 +1450,8 @@ describe('ReactDOMFizzServer', () => {
             </Suspense>
           </div>
         </Suspense>,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
     expect(getVisibleChildren(container)).toEqual('Loading Outer');
     // We should have received a partial segment containing the a partial of the fallback.
@@ -1554,11 +1533,10 @@ describe('ReactDOMFizzServer', () => {
     await jest.runAllTimers();
 
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <App isClient={false} />,
-        writable,
       );
-      startWriting();
+      pipe(writable);
     });
 
     // Nothing is output since root has a suspense with avoidedThisFallback that hasn't resolved
@@ -1669,18 +1647,18 @@ describe('ReactDOMFizzServer', () => {
 
     const loggedErrors = [];
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <Suspense fallback="Loading...">
           <App />
         </Suspense>,
-        writable,
+
         {
           onError(x) {
             loggedErrors.push(x);
           },
         },
       );
-      startWriting();
+      pipe(writable);
     });
     expect(Scheduler).toHaveYielded(['server']);
 
@@ -1752,18 +1730,18 @@ describe('ReactDOMFizzServer', () => {
 
     const loggedErrors = [];
     await act(async () => {
-      const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+      const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
         <Suspense fallback="Loading...">
           <App />
         </Suspense>,
-        writable,
+
         {
           onError(x) {
             loggedErrors.push(x);
           },
         },
       );
-      startWriting();
+      pipe(writable);
     });
     expect(Scheduler).toHaveYielded(['server']);
 
@@ -1838,11 +1816,8 @@ describe('ReactDOMFizzServer', () => {
       }
 
       await act(async () => {
-        const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
-          <App />,
-          writable,
-        );
-        startWriting();
+        const {pipe} = ReactDOMFizzServer.renderToPipeableStream(<App />);
+        pipe(writable);
       });
       expect(Scheduler).toHaveYielded(['Yay!']);
 
@@ -1922,11 +1897,8 @@ describe('ReactDOMFizzServer', () => {
       }
 
       await act(async () => {
-        const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
-          <App />,
-          writable,
-        );
-        startWriting();
+        const {pipe} = ReactDOMFizzServer.renderToPipeableStream(<App />);
+        pipe(writable);
       });
       expect(Scheduler).toHaveYielded(['Yay!']);
 
@@ -1997,11 +1969,8 @@ describe('ReactDOMFizzServer', () => {
       }
 
       await act(async () => {
-        const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
-          <App />,
-          writable,
-        );
-        startWriting();
+        const {pipe} = ReactDOMFizzServer.renderToPipeableStream(<App />);
+        pipe(writable);
       });
       expect(Scheduler).toHaveYielded(['Yay!']);
 
diff --git a/packages/react-dom/src/__tests__/ReactDOMFizzServerNode-test.js b/packages/react-dom/src/__tests__/ReactDOMFizzServerNode-test.js
index 949ddcb04fa2d..7fa1c208dffc9 100644
--- a/packages/react-dom/src/__tests__/ReactDOMFizzServerNode-test.js
+++ b/packages/react-dom/src/__tests__/ReactDOMFizzServerNode-test.js
@@ -57,13 +57,12 @@ describe('ReactDOMFizzServer', () => {
   }
 
   // @gate experimental
-  it('should call pipeToNodeWritable', () => {
+  it('should call renderToPipeableStream', () => {
     const {writable, output} = getTestWritable();
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <div>hello world</div>,
-      writable,
     );
-    startWriting();
+    pipe(writable);
     jest.runAllTimers();
     expect(output.result).toMatchInlineSnapshot(`"<div>hello world</div>"`);
   });
@@ -71,13 +70,12 @@ describe('ReactDOMFizzServer', () => {
   // @gate experimental
   it('should emit DOCTYPE at the root of the document', () => {
     const {writable, output} = getTestWritable();
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <html>
         <body>hello world</body>
       </html>,
-      writable,
     );
-    startWriting();
+    pipe(writable);
     jest.runAllTimers();
     expect(output.result).toMatchInlineSnapshot(
       `"<!DOCTYPE html><html><body>hello world</body></html>"`,
@@ -85,18 +83,17 @@ describe('ReactDOMFizzServer', () => {
   });
 
   // @gate experimental
-  it('should start writing after startWriting', () => {
+  it('should start writing after pipe', () => {
     const {writable, output} = getTestWritable();
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <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();
+    pipe(writable);
     expect(output.result).toMatchInlineSnapshot(
       `"<!doctype html><html><head><title>test</title><head><body><div>hello world</div>"`,
     );
@@ -115,13 +112,13 @@ describe('ReactDOMFizzServer', () => {
     }
     let isCompleteCalls = 0;
     const {writable, output} = getTestWritable();
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <div>
         <Suspense fallback="Loading">
           <Wait />
         </Suspense>
       </div>,
-      writable,
+
       {
         onCompleteAll() {
           isCompleteCalls++;
@@ -144,7 +141,7 @@ describe('ReactDOMFizzServer', () => {
     output.result +=
       '<!doctype html><html><head><title>test</title><head><body>';
     // Then React starts writing.
-    startWriting();
+    pipe(writable);
     expect(output.result).toMatchInlineSnapshot(
       `"<!doctype html><html><head><title>test</title><head><body><div><!--$-->Done<!-- --><!--/$--></div>"`,
     );
@@ -154,11 +151,11 @@ describe('ReactDOMFizzServer', () => {
   it('should error the stream when an error is thrown at the root', async () => {
     const reportedErrors = [];
     const {writable, output, completed} = getTestWritable();
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <div>
         <Throw />
       </div>,
-      writable,
+
       {
         onError(x) {
           reportedErrors.push(x);
@@ -167,7 +164,7 @@ describe('ReactDOMFizzServer', () => {
     );
 
     // The stream is errored once we start writing.
-    startWriting();
+    pipe(writable);
 
     await completed;
 
@@ -181,20 +178,20 @@ describe('ReactDOMFizzServer', () => {
   it('should error the stream when an error is thrown inside a fallback', async () => {
     const reportedErrors = [];
     const {writable, output, completed} = getTestWritable();
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <div>
         <Suspense fallback={<Throw />}>
           <InfiniteSuspend />
         </Suspense>
       </div>,
-      writable,
+
       {
         onError(x) {
           reportedErrors.push(x);
         },
       },
     );
-    startWriting();
+    pipe(writable);
 
     await completed;
 
@@ -207,20 +204,20 @@ describe('ReactDOMFizzServer', () => {
   it('should not error the stream when an error is thrown inside suspense boundary', async () => {
     const reportedErrors = [];
     const {writable, output, completed} = getTestWritable();
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <div>
         <Suspense fallback={<div>Loading</div>}>
           <Throw />
         </Suspense>
       </div>,
-      writable,
+
       {
         onError(x) {
           reportedErrors.push(x);
         },
       },
     );
-    startWriting();
+    pipe(writable);
 
     await completed;
 
@@ -242,13 +239,12 @@ describe('ReactDOMFizzServer', () => {
     function Content() {
       return 'Hi';
     }
-    const {startWriting} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe} = ReactDOMFizzServer.renderToPipeableStream(
       <Suspense fallback={<Fallback />}>
         <Content />
       </Suspense>,
-      writable,
     );
-    startWriting();
+    pipe(writable);
 
     await completed;
 
@@ -261,20 +257,20 @@ describe('ReactDOMFizzServer', () => {
   it('should be able to complete by aborting even if the promise never resolves', async () => {
     let isCompleteCalls = 0;
     const {writable, output, completed} = getTestWritable();
-    const {startWriting, abort} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe, abort} = ReactDOMFizzServer.renderToPipeableStream(
       <div>
         <Suspense fallback={<div>Loading</div>}>
           <InfiniteSuspend />
         </Suspense>
       </div>,
-      writable,
+
       {
         onCompleteAll() {
           isCompleteCalls++;
         },
       },
     );
-    startWriting();
+    pipe(writable);
 
     jest.runAllTimers();
 
@@ -294,7 +290,7 @@ describe('ReactDOMFizzServer', () => {
   it('should be able to complete by abort when the fallback is also suspended', async () => {
     let isCompleteCalls = 0;
     const {writable, output, completed} = getTestWritable();
-    const {startWriting, abort} = ReactDOMFizzServer.pipeToNodeWritable(
+    const {pipe, abort} = ReactDOMFizzServer.renderToPipeableStream(
       <div>
         <Suspense fallback="Loading">
           <Suspense fallback={<InfiniteSuspend />}>
@@ -302,14 +298,14 @@ describe('ReactDOMFizzServer', () => {
           </Suspense>
         </Suspense>
       </div>,
-      writable,
+
       {
         onCompleteAll() {
           isCompleteCalls++;
         },
       },
     );
-    startWriting();
+    pipe(writable);
 
     jest.runAllTimers();
 
diff --git a/packages/react-dom/src/server/ReactDOMFizzServerNode.js b/packages/react-dom/src/server/ReactDOMFizzServerNode.js
index fb1939f0c404f..4233a88e1fb94 100644
--- a/packages/react-dom/src/server/ReactDOMFizzServerNode.js
+++ b/packages/react-dom/src/server/ReactDOMFizzServerNode.js
@@ -41,7 +41,7 @@ type Controls = {|
   // Cancel any pending I/O and put anything remaining into
   // client rendered mode.
   abort(): void,
-  startWriting(): void,
+  pipe<T: Writable>(destination: T): T,
 |};
 
 function createRequestImpl(children: ReactNodeList, options: void | Options) {
@@ -56,22 +56,24 @@ function createRequestImpl(children: ReactNodeList, options: void | Options) {
   );
 }
 
-function pipeToNodeWritable(
+function renderToPipeableStream(
   children: ReactNodeList,
-  destination: Writable,
   options?: Options,
 ): Controls {
   const request = createRequestImpl(children, options);
   let hasStartedFlowing = false;
   startWork(request);
   return {
-    startWriting() {
+    pipe<T: Writable>(destination: T): T {
       if (hasStartedFlowing) {
-        return;
+        throw new Error(
+          'React currently only supports piping to one writable stream.',
+        );
       }
       hasStartedFlowing = true;
       startFlowing(request, destination);
       destination.on('drain', createDrainHandler(destination, request));
+      return destination;
     },
     abort() {
       abort(request);
@@ -79,4 +81,4 @@ function pipeToNodeWritable(
   };
 }
 
-export {pipeToNodeWritable, ReactVersion as version};
+export {renderToPipeableStream, ReactVersion as version};
diff --git a/packages/react-server-dom-webpack/src/ReactFlightDOMServerNode.js b/packages/react-server-dom-webpack/src/ReactFlightDOMServerNode.js
index 35773279890ca..5f992d4b03e9b 100644
--- a/packages/react-server-dom-webpack/src/ReactFlightDOMServerNode.js
+++ b/packages/react-server-dom-webpack/src/ReactFlightDOMServerNode.js
@@ -25,20 +25,35 @@ type Options = {
   onError?: (error: mixed) => void,
 };
 
-function pipeToNodeWritable(
+type Controls = {|
+  pipe<T: Writable>(destination: T): T,
+|};
+
+function renderToPipeableStream(
   model: ReactModel,
-  destination: Writable,
   webpackMap: BundlerConfig,
   options?: Options,
-): void {
+): Controls {
   const request = createRequest(
     model,
     webpackMap,
     options ? options.onError : undefined,
   );
+  let hasStartedFlowing = false;
   startWork(request);
-  startFlowing(request, destination);
-  destination.on('drain', createDrainHandler(destination, request));
+  return {
+    pipe<T: Writable>(destination: T): T {
+      if (hasStartedFlowing) {
+        throw new Error(
+          'React currently only supports piping to one writable stream.',
+        );
+      }
+      hasStartedFlowing = true;
+      startFlowing(request, destination);
+      destination.on('drain', createDrainHandler(destination, request));
+      return destination;
+    },
+  };
 }
 
-export {pipeToNodeWritable};
+export {renderToPipeableStream};
diff --git a/packages/react-server-dom-webpack/src/__tests__/ReactFlightDOM-test.js b/packages/react-server-dom-webpack/src/__tests__/ReactFlightDOM-test.js
index 60fbac215f73d..bcae6c50f7c7f 100644
--- a/packages/react-server-dom-webpack/src/__tests__/ReactFlightDOM-test.js
+++ b/packages/react-server-dom-webpack/src/__tests__/ReactFlightDOM-test.js
@@ -57,8 +57,8 @@ describe('ReactFlightDOM', () => {
       },
     });
     return {
-      writable,
       readable,
+      writable,
     };
   }
 
@@ -113,7 +113,11 @@ describe('ReactFlightDOM', () => {
     }
 
     const {writable, readable} = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(<App />, writable, webpackMap);
+    const {pipe} = ReactServerDOMWriter.renderToPipeableStream(
+      <App />,
+      webpackMap,
+    );
+    pipe(writable);
     const response = ReactServerDOMReader.createFromReadableStream(readable);
     await waitForSuspense(() => {
       const model = response.readRoot();
@@ -162,11 +166,11 @@ describe('ReactFlightDOM', () => {
     }
 
     const {writable, readable} = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(
+    const {pipe} = ReactServerDOMWriter.renderToPipeableStream(
       <RootModel />,
-      writable,
       webpackMap,
     );
+    pipe(writable);
     const response = ReactServerDOMReader.createFromReadableStream(readable);
 
     const container = document.createElement('div');
@@ -200,11 +204,11 @@ describe('ReactFlightDOM', () => {
     }
 
     const {writable, readable} = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(
+    const {pipe} = ReactServerDOMWriter.renderToPipeableStream(
       <RootModel />,
-      writable,
       webpackMap,
     );
+    pipe(writable);
     const response = ReactServerDOMReader.createFromReadableStream(readable);
 
     const container = document.createElement('div');
@@ -236,11 +240,11 @@ describe('ReactFlightDOM', () => {
     }
 
     const {writable, readable} = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(
+    const {pipe} = ReactServerDOMWriter.renderToPipeableStream(
       <RootModel />,
-      writable,
       webpackMap,
     );
+    pipe(writable);
     const response = ReactServerDOMReader.createFromReadableStream(readable);
 
     const container = document.createElement('div');
@@ -371,11 +375,16 @@ describe('ReactFlightDOM', () => {
     }
 
     const {writable, readable} = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(model, writable, webpackMap, {
-      onError(x) {
-        reportedErrors.push(x);
+    const {pipe} = ReactServerDOMWriter.renderToPipeableStream(
+      model,
+      webpackMap,
+      {
+        onError(x) {
+          reportedErrors.push(x);
+        },
       },
-    });
+    );
+    pipe(writable);
     const response = ReactServerDOMReader.createFromReadableStream(readable);
 
     const container = document.createElement('div');
@@ -481,11 +490,11 @@ describe('ReactFlightDOM', () => {
     const root = ReactDOM.createRoot(container);
 
     const stream1 = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(
+    const {pipe} = ReactServerDOMWriter.renderToPipeableStream(
       <App color="red" />,
-      stream1.writable,
       webpackMap,
     );
+    pipe(stream1.writable);
     const response1 = ReactServerDOMReader.createFromReadableStream(
       stream1.readable,
     );
@@ -509,11 +518,11 @@ describe('ReactFlightDOM', () => {
     inputB.value = 'goodbye';
 
     const stream2 = getTestStream();
-    ReactServerDOMWriter.pipeToNodeWritable(
+    const {pipe: pipe2} = ReactServerDOMWriter.renderToPipeableStream(
       <App color="blue" />,
-      stream2.writable,
       webpackMap,
     );
+    pipe2(stream2.writable);
     const response2 = ReactServerDOMReader.createFromReadableStream(
       stream2.readable,
     );
diff --git a/scripts/error-codes/codes.json b/scripts/error-codes/codes.json
index d1688e84bc7a8..aab5d62476cc3 100644
--- a/scripts/error-codes/codes.json
+++ b/scripts/error-codes/codes.json
@@ -401,5 +401,6 @@
   "413": "Expected finished root and lanes to be set. This is a bug in React.",
   "414": "Did not expect this call in production. This is a bug in React. Please file an issue.",
   "415": "Error parsing the data. It's probably an error code or network corruption.",
-  "416": "This environment don't support binary chunks."
+  "416": "This environment don't support binary chunks.",
+  "417": "React currently only supports piping to one writable stream."
 }