stream rsc payload in json objects like streamed react components

Author: AbanoubGhadban

lib/react_on_rails/helper.rb

@@ -429,7 +429,7 @@ def internal_rsc_react_component(react_component_name, options = {})
       render_options = create_render_options(react_component_name, options)
       json_stream = server_rendered_react_component(render_options)
       json_stream.transform do |chunk|
-        chunk[:html].html_safe
+        (chunk.to_json + "\n").html_safe
       end
     end
 
@@ -701,10 +701,7 @@ def server_rendered_react_component(render_options)
                                                js_code: js_code)
       end
 
-      # TODO: handle errors for rsc streams
-      return result if render_options.rsc?
-
-      if render_options.stream?
+      if render_options.stream? || render_options.rsc?
         result.transform do |chunk_json_result|
           if should_raise_streaming_prerender_error?(chunk_json_result, render_options)
             raise_prerender_error(chunk_json_result, react_component_name, props, js_code)

lib/react_on_rails/server_rendering_pool/ruby_embedded_java_script.rb

@@ -231,7 +231,6 @@ def parse_result_and_replay_console_messages(result_string, render_options)
           begin
             result = JSON.parse(result_string)
           rescue JSON::ParserError => e
-            return { html: result_string }
             raise ReactOnRails::JsonParseError.new(parse_error: e, json: result_string)
           end
 

node_package/src/RSCClientRoot.ts

@@ -1,5 +1,6 @@
 import * as React from 'react';
 import RSDWClient from 'react-server-dom-webpack/client';
+import transformRSCStreamAndReplayConsoleLogs from './transformRSCStreamAndReplayConsoleLogs';
 
 if (!('use' in React)) {
   throw new Error('React.use is not defined. Please ensure you are using React 18.3.0-canary-670811593-20240322 or later to use server components.');
@@ -11,9 +12,19 @@ const { use } = React as { use: Use };
 
 const renderCache: Record<string, Promise<unknown>> = {};
 
+const createFromFetch = async (fetchPromise: Promise<Response>) => {
+  const response = await fetchPromise;
+  const stream = response.body;
+  if (!stream) {
+    throw new Error('No stream found in response');
+  }
+  const transformedStream = transformRSCStreamAndReplayConsoleLogs(stream);
+  return RSDWClient.createFromReadableStream(transformedStream);
+}
+
 const fetchRSC = ({ componentName }: { componentName: string }) => {
   if (!renderCache[componentName]) {
-    renderCache[componentName] = RSDWClient.createFromFetch(fetch(`/rsc/${componentName}`));
+    renderCache[componentName] = createFromFetch(fetch(`/rsc/${componentName}`));
   }
   return renderCache[componentName];
 }

node_package/src/ReactOnRails.ts

@@ -1,5 +1,5 @@
 import type { ReactElement } from 'react';
-import type { Readable, PassThrough } from 'stream';
+import type { Readable } from 'stream';
 
 import * as ClientStartup from './clientStartup';
 import handleError from './handleError';
@@ -306,7 +306,7 @@ ctx.ReactOnRails = {
    * @param options
    */
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
-  serverRenderRSCReactComponent(options: RenderParams): PassThrough {
+  serverRenderRSCReactComponent(_): Readable {
     throw new Error('serverRenderRSCReactComponent is supported in RSC bundle only.');
   },
 

node_package/src/ReactOnRailsRSC.ts

@@ -1,13 +1,19 @@
-// @ts-expect-error will define this module types later
-import { renderToReadableStream } from 'react-server-dom-webpack/server.edge';
-import { PassThrough } from 'stream';
+import { renderToPipeableStream } from 'react-server-dom-webpack/server.node';
+import { PassThrough, Readable } from 'stream';
+import type { ReactElement } from 'react';
 import fs from 'fs';
 
 import { RenderParams } from './types';
-import ComponentRegistry from './ComponentRegistry';
-import createReactOutput from './createReactOutput';
-import { isPromise, isServerRenderHash } from './isServerRenderResult';
 import ReactOnRails from './ReactOnRails';
+import buildConsoleReplay from './buildConsoleReplay';
+import handleError from './handleError';
+import {
+  streamServerRenderedComponent,
+  type StreamRenderState,
+  transformRenderStreamChunksToResultObject,
+  convertToError,
+  createResultObject,
+} from './serverRenderReactComponent';
 
 (async () => {
   try {
@@ -25,68 +31,51 @@ const stringToStream = (str: string) => {
   return stream;
 };
 
-const getBundleConfig = () => {
-  const bundleConfig = JSON.parse(fs.readFileSync('./public/webpack/development/react-client-manifest.json', 'utf8'));
-  // remove file:// from keys
-  const newBundleConfig: { [key: string]: unknown } = {};
-  for (const [key, value] of Object.entries(bundleConfig)) {
-    newBundleConfig[key.replace('file://', '')] = value;
-  }
-  return newBundleConfig;
-}
-
-ReactOnRails.serverRenderRSCReactComponent = (options: RenderParams) => {
-  const { name, domNodeId, trace, props, railsContext, throwJsErrors } = options;
+const getBundleConfig = () => JSON.parse(fs.readFileSync('./public/webpack/development/react-client-manifest.json', 'utf8'))
 
-  let renderResult: null | PassThrough = null;
+const streamRenderRSCComponent = (reactElement: ReactElement, options: RenderParams): Readable => {
+  const { throwJsErrors } = options;
+  const renderState: StreamRenderState = {
+    result: null,
+    hasErrors: false,
+    isShellReady: true
+  };
 
+  const { pipeToTransform, readableStream, emitError } = transformRenderStreamChunksToResultObject(renderState);
   try {
-    const componentObj = ComponentRegistry.get(name);
-    if (componentObj.isRenderer) {
-      throw new Error(`\
-Detected a renderer while server rendering component '${name}'. \
-See https://github.com/shakacode/react_on_rails#renderer-functions`);
-    }
-
-    const reactRenderingResult = createReactOutput({
-      componentObj,
-      domNodeId,
-      trace,
-      props,
-      railsContext,
-    });
-
-    if (isServerRenderHash(reactRenderingResult) || isPromise(reactRenderingResult)) {
-      throw new Error('Server rendering of streams is not supported for server render hashes or promises.');
-    }
-
-    renderResult = new PassThrough();
-    let finalValue = "";
-    const streamReader = renderToReadableStream(reactRenderingResult, getBundleConfig()).getReader();
-    const decoder = new TextDecoder();
-    const processStream = async () => {
-      const { done, value } = await streamReader.read();
-      if (done) {
-        renderResult?.push(null);
-        // @ts-expect-error value is not typed
-        debugConsole.log('value', finalValue);
-        return;
+    const rscStream = renderToPipeableStream(
+      reactElement,
+      getBundleConfig(),
+      {
+        onError: (err) => {
+          const error = convertToError(err);
+          console.error("Error in RSC stream", error);
+          if (throwJsErrors) {
+            emitError(error);
+          }
+          renderState.hasErrors = true;
+          renderState.error = error;
+        }
       }
-
-      finalValue += decoder.decode(value);
-      renderResult?.push(value);
-      processStream();
-    }
-    processStream();
-  } catch (e: unknown) {
-    if (throwJsErrors) {
-      throw e;
-    }
-
-    renderResult = stringToStream(`Error: ${e}`);
+    );
+    pipeToTransform(rscStream);
+    return readableStream;
+  } catch (e) {
+    const error = convertToError(e);
+    renderState.hasErrors = true;
+    renderState.error = error;
+    const htmlResult = handleError({ e: error, name: options.name, serverSide: true });
+    const jsonResult = JSON.stringify(createResultObject(htmlResult, buildConsoleReplay(), renderState));
+    return stringToStream(jsonResult);
   }
+};
 
-  return renderResult;
+ReactOnRails.serverRenderRSCReactComponent = (options: RenderParams) => {
+  try {
+    return streamServerRenderedComponent(options, streamRenderRSCComponent);
+  } finally {
+    console.history = [];
+  }
 };
 
 export * from './types';

node_package/src/serverRenderReactComponent.ts

@@ -15,7 +15,7 @@ type RenderState = {
   error?: RenderingError;
 };
 
-type StreamRenderState = Omit<RenderState, 'result'> & {
+export type StreamRenderState = Omit<RenderState, 'result'> & {
   result: null | Readable;
   isShellReady: boolean;
 };
@@ -27,7 +27,7 @@ type RenderOptions = {
   renderingReturnsPromises: boolean;
 };
 
-function convertToError(e: unknown): Error {
+export function convertToError(e: unknown): Error {
   return e instanceof Error ? e : new Error(String(e));
 }
 
@@ -104,7 +104,7 @@ function handleRenderingError(e: unknown, options: { componentName: string, thro
   };
 }
 
-function createResultObject(html: string | null, consoleReplayScript: string, renderState: RenderState | StreamRenderState): RenderResult {
+export function createResultObject(html: string | null, consoleReplayScript: string, renderState: RenderState | StreamRenderState): RenderResult {
   return {
     html,
     consoleReplayScript,
@@ -210,7 +210,7 @@ const stringToStream = (str: string): Readable => {
   return stream;
 };
 
-const transformRenderStreamChunksToResultObject = (renderState: StreamRenderState) => {
+export const transformRenderStreamChunksToResultObject = (renderState: StreamRenderState) => {
   const consoleHistory = console.history;
   let previouslyReplayedConsoleMessages = 0;
 
@@ -298,7 +298,15 @@ const streamRenderReactComponent = (reactRenderingResult: ReactElement, options:
   return readableStream;
 }
 
-export const streamServerRenderedReactComponent = (options: RenderParams): Readable => {
+export type StreamRenderer<T> = (
+  reactElement: ReactElement,
+  options: RenderParams
+) => T;
+
+export const streamServerRenderedComponent = <T>(
+  options: RenderParams,
+  renderStrategy: StreamRenderer<T>
+): T => {
   const { name: componentName, domNodeId, trace, props, railsContext, throwJsErrors } = options;
 
   try {
@@ -317,7 +325,7 @@ export const streamServerRenderedReactComponent = (options: RenderParams): Reada
       throw new Error('Server rendering of streams is not supported for server render hashes or promises.');
     }
 
-    return streamRenderReactComponent(reactRenderingResult, options);
+    return renderStrategy(reactRenderingResult, options);
   } catch (e) {
     if (throwJsErrors) {
       throw e;
@@ -326,8 +334,11 @@ export const streamServerRenderedReactComponent = (options: RenderParams): Reada
     const error = convertToError(e);
     const htmlResult = handleError({ e: error, name: componentName, serverSide: true });
     const jsonResult = JSON.stringify(createResultObject(htmlResult, buildConsoleReplay(), { hasErrors: true, error, result: null }));
-    return stringToStream(jsonResult);
+    return stringToStream(jsonResult) as T;
   }
 };
 
+// Update the existing streamServerRenderedReactComponent to use the new shared function
+export const streamServerRenderedReactComponent = (options: RenderParams): Readable => streamServerRenderedComponent(options, streamRenderReactComponent);
+
 export default serverRenderReactComponent;

node_package/src/transformRSCStreamAndReplayConsoleLogs.ts

@@ -0,0 +1,40 @@
+export default function transformRSCStreamAndReplayConsoleLogs(stream: ReadableStream) {
+  return new ReadableStream({
+    async start(controller) {
+      const reader = stream.getReader();
+      const decoder = new TextDecoder();
+      const encoder = new TextEncoder();
+
+      let { value, done } = await reader.read();
+      while (!done) {
+        const decodedValue = decoder.decode(value);
+        const jsonChunks = decodedValue.split('\n')
+          .filter(line => line.trim() !== '')
+          .map((line) => {
+            try {
+              return JSON.parse(line);
+            } catch (error) {
+              console.error('Error parsing JSON:', line, error);
+              throw error;
+            }
+          });
+
+        for (const jsonChunk of jsonChunks) {
+          const { html, consoleReplayScript } = jsonChunk;
+          controller.enqueue(encoder.encode(html));
+
+          const replayConsoleCode = consoleReplayScript?.trim().replace(/^<script.*>/, '').replace(/<\/script>$/, '');
+          if (replayConsoleCode?.trim() !== '') {
+            const scriptElement = document.createElement('script');
+            scriptElement.textContent = replayConsoleCode;
+            document.body.appendChild(scriptElement);
+          }
+        }
+
+        // eslint-disable-next-line no-await-in-loop
+        ({ value, done } = await reader.read());
+      }
+      controller.close();
+    }
+  });
+}

node_package/src/types/index.ts

@@ -177,7 +177,7 @@ export interface ReactOnRails {
   getOrWaitForComponent(name: string): Promise<RegisteredComponent>;
   serverRenderReactComponent(options: RenderParams): null | string | Promise<RenderResult>;
   streamServerRenderedReactComponent(options: RenderParams): Readable;
-  serverRenderRSCReactComponent(options: RenderParams): PassThrough;
+  serverRenderRSCReactComponent(options: RenderParams): Readable;
   handleError(options: ErrorOptions): string | undefined;
   buildConsoleReplay(): string;
   registeredComponents(): Map<string, RegisteredComponent>;

node_package/types/react-server-dom-webpack.d.ts

@@ -1,3 +1,26 @@
+declare module 'react-server-dom-webpack/server.node' {
+  export interface Options {
+    environmentName?: string;
+    onError?: (error: unknown) => void;
+    onPostpone?: (reason: string) => void;
+    identifierPrefix?: string;
+  }
+
+  export interface PipeableStream {
+    abort(reason: unknown): void;
+    pipe<Writable extends NodeJS.WritableStream>(destination: Writable): Writable;
+  }
+
+  // Note: ReactClientValue is likely what React uses internally for RSC
+  // We're using 'unknown' here as it's the most accurate type we can use
+  // without accessing React's internal types
+  export function renderToPipeableStream(
+    model: unknown,
+    webpackMap: { [key: string]: unknown },
+    options?: Options
+  ): PipeableStream;
+}
+
 declare module 'react-server-dom-webpack/client' {
   export const createFromFetch: (promise: Promise<Response>) => Promise<unknown>;