[Fizz] Add Web Streams to Fizz Node entry point (#33475)

New take on #33441.

This uses a wrapper instead of a separate bundle.

Author: sebmarkbage

packages/react-dom/npm/server.node.js

@@ -13,6 +13,10 @@ exports.version = l.version;
 exports.renderToString = l.renderToString;
 exports.renderToStaticMarkup = l.renderToStaticMarkup;
 exports.renderToPipeableStream = s.renderToPipeableStream;
+exports.renderToReadableStream = s.renderToReadableStream;
 if (s.resumeToPipeableStream) {
   exports.resumeToPipeableStream = s.resumeToPipeableStream;
 }
+if (s.resume) {
+  exports.resume = s.resume;
+}

packages/react-dom/npm/static.node.js

@@ -9,4 +9,6 @@ if (process.env.NODE_ENV === 'production') {
 
 exports.version = s.version;
 exports.prerenderToNodeStream = s.prerenderToNodeStream;
+exports.prerender = s.prerender;
 exports.resumeAndPrerenderToNodeStream = s.resumeAndPrerenderToNodeStream;
+exports.resumeAndPrerender = s.resumeAndPrerender;

packages/react-dom/server.node.js

@@ -37,3 +37,17 @@ export function resumeToPipeableStream() {
     arguments,
   );
 }
+
+export function renderToReadableStream() {
+  return require('./src/server/react-dom-server.node').renderToReadableStream.apply(
+    this,
+    arguments,
+  );
+}
+
+export function resume() {
+  return require('./src/server/react-dom-server.node').resume.apply(
+    this,
+    arguments,
+  );
+}

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

@@ -56,6 +56,18 @@ describe('ReactDOMFizzServerNode', () => {
     throw theInfinitePromise;
   }
 
+  async function readContentWeb(stream) {
+    const reader = stream.getReader();
+    let content = '';
+    while (true) {
+      const {done, value} = await reader.read();
+      if (done) {
+        return content;
+      }
+      content += Buffer.from(value).toString('utf8');
+    }
+  }
+
   it('should call renderToPipeableStream', async () => {
     const {writable, output} = getTestWritable();
     await act(() => {
@@ -67,6 +79,14 @@ describe('ReactDOMFizzServerNode', () => {
     expect(output.result).toMatchInlineSnapshot(`"<div>hello world</div>"`);
   });
 
+  it('should support web streams', async () => {
+    const stream = await act(() =>
+      ReactDOMFizzServer.renderToReadableStream(<div>hello world</div>),
+    );
+    const result = await readContentWeb(stream);
+    expect(result).toMatchInlineSnapshot(`"<div>hello world</div>"`);
+  });
+
   it('flush fully if piping in on onShellReady', async () => {
     const {writable, output} = getTestWritable();
     await act(() => {

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

@@ -46,6 +46,18 @@ describe('ReactDOMFizzStaticNode', () => {
     });
   }
 
+  async function readContentWeb(stream) {
+    const reader = stream.getReader();
+    let content = '';
+    while (true) {
+      const {done, value} = await reader.read();
+      if (done) {
+        return content;
+      }
+      content += Buffer.from(value).toString('utf8');
+    }
+  }
+
   // @gate experimental
   it('should call prerenderToNodeStream', async () => {
     const result = await ReactDOMFizzStatic.prerenderToNodeStream(
@@ -55,6 +67,13 @@ describe('ReactDOMFizzStaticNode', () => {
     expect(prelude).toMatchInlineSnapshot(`"<div>hello world</div>"`);
   });
 
+  // @gate experimental
+  it('should suppport web streams', async () => {
+    const result = await ReactDOMFizzStatic.prerender(<div>hello world</div>);
+    const prelude = await readContentWeb(result.prelude);
+    expect(prelude).toMatchInlineSnapshot(`"<div>hello world</div>"`);
+  });
+
   // @gate experimental
   it('should emit DOCTYPE at the root of the document', async () => {
     const result = await ReactDOMFizzStatic.prerenderToNodeStream(

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

@@ -41,6 +41,8 @@ import {
   createRootFormatContext,
 } from 'react-dom-bindings/src/server/ReactFizzConfigDOM';
 
+import {textEncoder} from 'react-server/src/ReactServerStreamConfigNode';
+
 import {ensureCorrectIsomorphicReactVersion} from '../shared/ensureCorrectIsomorphicReactVersion';
 ensureCorrectIsomorphicReactVersion();
 
@@ -167,6 +169,141 @@ function renderToPipeableStream(
   };
 }
 
+function createFakeWritableFromReadableStreamController(
+  controller: ReadableStreamController,
+): Writable {
+  // The current host config expects a Writable so we create
+  // a fake writable for now to push into the Readable.
+  return ({
+    write(chunk: string | Uint8Array) {
+      if (typeof chunk === 'string') {
+        chunk = textEncoder.encode(chunk);
+      }
+      controller.enqueue(chunk);
+      // in web streams there is no backpressure so we can alwas write more
+      return true;
+    },
+    end() {
+      controller.close();
+    },
+    destroy(error) {
+      // $FlowFixMe[method-unbinding]
+      if (typeof controller.error === 'function') {
+        // $FlowFixMe[incompatible-call]: This is an Error object or the destination accepts other types.
+        controller.error(error);
+      } else {
+        controller.close();
+      }
+    },
+  }: any);
+}
+
+// TODO: Move to sub-classing ReadableStream.
+type ReactDOMServerReadableStream = ReadableStream & {
+  allReady: Promise<void>,
+};
+
+type WebStreamsOptions = Omit<
+  Options,
+  'onShellReady' | 'onShellError' | 'onAllReady' | 'onHeaders',
+> & {signal: AbortSignal, onHeaders?: (headers: Headers) => void};
+
+function renderToReadableStream(
+  children: ReactNodeList,
+  options?: WebStreamsOptions,
+): Promise<ReactDOMServerReadableStream> {
+  return new Promise((resolve, reject) => {
+    let onFatalError;
+    let onAllReady;
+    const allReady = new Promise<void>((res, rej) => {
+      onAllReady = res;
+      onFatalError = rej;
+    });
+
+    function onShellReady() {
+      let writable: Writable;
+      const stream: ReactDOMServerReadableStream = (new ReadableStream(
+        {
+          type: 'bytes',
+          start: (controller): ?Promise<void> => {
+            writable =
+              createFakeWritableFromReadableStreamController(controller);
+          },
+          pull: (controller): ?Promise<void> => {
+            startFlowing(request, writable);
+          },
+          cancel: (reason): ?Promise<void> => {
+            stopFlowing(request);
+            abort(request, reason);
+          },
+        },
+        // $FlowFixMe[prop-missing] size() methods are not allowed on byte streams.
+        {highWaterMark: 0},
+      ): any);
+      // TODO: Move to sub-classing ReadableStream.
+      stream.allReady = allReady;
+      resolve(stream);
+    }
+    function onShellError(error: mixed) {
+      // If the shell errors the caller of `renderToReadableStream` won't have access to `allReady`.
+      // However, `allReady` will be rejected by `onFatalError` as well.
+      // So we need to catch the duplicate, uncatchable fatal error in `allReady` to prevent a `UnhandledPromiseRejection`.
+      allReady.catch(() => {});
+      reject(error);
+    }
+
+    const onHeaders = options ? options.onHeaders : undefined;
+    let onHeadersImpl;
+    if (onHeaders) {
+      onHeadersImpl = (headersDescriptor: HeadersDescriptor) => {
+        onHeaders(new Headers(headersDescriptor));
+      };
+    }
+
+    const resumableState = createResumableState(
+      options ? options.identifierPrefix : undefined,
+      options ? options.unstable_externalRuntimeSrc : undefined,
+      options ? options.bootstrapScriptContent : undefined,
+      options ? options.bootstrapScripts : undefined,
+      options ? options.bootstrapModules : undefined,
+    );
+    const request = createRequest(
+      children,
+      resumableState,
+      createRenderState(
+        resumableState,
+        options ? options.nonce : undefined,
+        options ? options.unstable_externalRuntimeSrc : undefined,
+        options ? options.importMap : undefined,
+        onHeadersImpl,
+        options ? options.maxHeadersLength : undefined,
+      ),
+      createRootFormatContext(options ? options.namespaceURI : undefined),
+      options ? options.progressiveChunkSize : undefined,
+      options ? options.onError : undefined,
+      onAllReady,
+      onShellReady,
+      onShellError,
+      onFatalError,
+      options ? options.onPostpone : undefined,
+      options ? options.formState : undefined,
+    );
+    if (options && options.signal) {
+      const signal = options.signal;
+      if (signal.aborted) {
+        abort(request, (signal: any).reason);
+      } else {
+        const listener = () => {
+          abort(request, (signal: any).reason);
+          signal.removeEventListener('abort', listener);
+        };
+        signal.addEventListener('abort', listener);
+      }
+    }
+    startWork(request);
+  });
+}
+
 function resumeRequestImpl(
   children: ReactNodeList,
   postponedState: PostponedState,
@@ -225,8 +362,89 @@ function resumeToPipeableStream(
   };
 }
 
+type WebStreamsResumeOptions = Omit<
+  Options,
+  'onShellReady' | 'onShellError' | 'onAllReady',
+> & {signal: AbortSignal};
+
+function resume(
+  children: ReactNodeList,
+  postponedState: PostponedState,
+  options?: WebStreamsResumeOptions,
+): Promise<ReactDOMServerReadableStream> {
+  return new Promise((resolve, reject) => {
+    let onFatalError;
+    let onAllReady;
+    const allReady = new Promise<void>((res, rej) => {
+      onAllReady = res;
+      onFatalError = rej;
+    });
+
+    function onShellReady() {
+      let writable: Writable;
+      const stream: ReactDOMServerReadableStream = (new ReadableStream(
+        {
+          type: 'bytes',
+          start: (controller): ?Promise<void> => {
+            writable =
+              createFakeWritableFromReadableStreamController(controller);
+          },
+          pull: (controller): ?Promise<void> => {
+            startFlowing(request, writable);
+          },
+          cancel: (reason): ?Promise<void> => {
+            stopFlowing(request);
+            abort(request, reason);
+          },
+        },
+        // $FlowFixMe[prop-missing] size() methods are not allowed on byte streams.
+        {highWaterMark: 0},
+      ): any);
+      // TODO: Move to sub-classing ReadableStream.
+      stream.allReady = allReady;
+      resolve(stream);
+    }
+    function onShellError(error: mixed) {
+      // If the shell errors the caller of `renderToReadableStream` won't have access to `allReady`.
+      // However, `allReady` will be rejected by `onFatalError` as well.
+      // So we need to catch the duplicate, uncatchable fatal error in `allReady` to prevent a `UnhandledPromiseRejection`.
+      allReady.catch(() => {});
+      reject(error);
+    }
+    const request = resumeRequest(
+      children,
+      postponedState,
+      resumeRenderState(
+        postponedState.resumableState,
+        options ? options.nonce : undefined,
+      ),
+      options ? options.onError : undefined,
+      onAllReady,
+      onShellReady,
+      onShellError,
+      onFatalError,
+      options ? options.onPostpone : undefined,
+    );
+    if (options && options.signal) {
+      const signal = options.signal;
+      if (signal.aborted) {
+        abort(request, (signal: any).reason);
+      } else {
+        const listener = () => {
+          abort(request, (signal: any).reason);
+          signal.removeEventListener('abort', listener);
+        };
+        signal.addEventListener('abort', listener);
+      }
+    }
+    startWork(request);
+  });
+}
+
 export {
   renderToPipeableStream,
+  renderToReadableStream,
   resumeToPipeableStream,
+  resume,
   ReactVersion as version,
 };

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

@@ -159,9 +159,8 @@ function prerender(
 type ResumeOptions = {
   nonce?: NonceOption,
   signal?: AbortSignal,
-  onError?: (error: mixed) => ?string,
-  onPostpone?: (reason: string) => void,
-  unstable_externalRuntimeSrc?: string | BootstrapScriptDescriptor,
+  onError?: (error: mixed, errorInfo: ErrorInfo) => ?string,
+  onPostpone?: (reason: string, postponeInfo: PostponeInfo) => void,
 };
 
 function resumeAndPrerender(