Add incremental rendering support with bidirectional streaming

Step 3: Implement render_code_with_incremental_updates and supporting infrastructure

This commit adds support for incremental rendering using HTTP2 bidirectional
streaming, enabling progressive resolution of async props during server-side
rendering.

## Key Changes:

**AsyncPropsEmitter** (lib/react_on_rails_pro/async_props_emitter.rb):
- Created emitter class for sending async prop updates to Node renderer
- Implements `call(prop_name, prop_value)` for emitting updates
- Owns generation of `end_stream_chunk` for onRequestClosedUpdateChunk
- Encapsulates all update chunk and end stream JavaScript generation

**StreamRequest** (lib/react_on_rails_pro/stream_request.rb):
- Added Async::Barrier support for non-blocking concurrent task management
- Wraps each_chunk execution in Sync block with barrier
- Passes barrier to request_executor for spawning async tasks
- Ensures proper cleanup and synchronization of async operations

**Request** (lib/react_on_rails_pro/request.rb):
- Refactored connection architecture with two separate connections:
  - standard_connection: Uses :stream plugin for regular streaming
  - incremental_connection: Uses :stream_bidi for bidirectional streaming
- Implemented render_code_with_incremental_updates:
  - Creates bidirectional streaming request with NDJSON format
  - Spawns async props block in background using barrier.async
  - Properly closes request stream after async block completes
  - Selects bundle timestamp based on RSC vs standard rendering
- Added build_initial_incremental_request helper using emitter

**Tests**:
- Added comprehensive tests for render_code_with_incremental_updates
- Tests verify NDJSON request format, barrier.async spawning, and emitter usage
- Tests for rsc_bundle_hash selection when is_rsc_payload is true
- Uses unverified doubles for HTTPX streaming interfaces with justification
- Updated stream_request_spec to test barrier parameter passing

## Technical Details:

- Uses Async gem with Async::Barrier for concurrent async prop resolution
- HTTP2 bidirectional streaming via HTTPX :stream_bidi plugin
- NDJSON protocol for streaming communication
- Proper separation of concerns: AsyncPropsEmitter owns chunk generation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: AbanoubGhadban

react_on_rails_pro/app/helpers/react_on_rails_pro_helper.rb

@@ -137,6 +137,11 @@ def stream_react_component(component_name, options = {})
     end
   end
 
+  def stream_react_component_with_async_props(component_name, options = {}, &props_block)
+    options[:async_props_block] = props_block
+    stream_react_component(component_name, options)
+  end
+
   # Renders the React Server Component (RSC) payload for a given component. This helper generates
   # a special format designed by React for serializing server components and transmitting them
   # to the client.

react_on_rails_pro/lib/react_on_rails_pro/async_props_emitter.rb

@@ -21,6 +21,15 @@ def call(prop_name, prop_value)
       # Continue - don't abort entire render because one prop failed
     end
 
+    # Generates the chunk that should be executed when the request stream closes
+    # This tells the asyncPropsManager to end the stream
+    def end_stream_chunk
+      {
+        bundleTimestamp: @bundle_timestamp,
+        updateChunk: generate_end_stream_js
+      }
+    end
+
     private
 
     def generate_update_chunk(prop_name, value)
@@ -38,5 +47,14 @@ def generate_set_prop_js(prop_name, value)
         })()
       JS
     end
+
+    def generate_end_stream_js
+      <<~JS.strip
+        (function(){
+          var asyncPropsManager = sharedExecutionContext.get("asyncPropsManager");
+          asyncPropsManager.endStream();
+        })()
+      JS
+    end
   end
 end

react_on_rails_pro/lib/react_on_rails_pro/request.rb

@@ -3,13 +3,16 @@
 require "uri"
 require "httpx"
 require_relative "stream_request"
+require_relative "async_props_emitter"
 
 module ReactOnRailsPro
   class Request # rubocop:disable Metrics/ClassLength
     class << self
       def reset_connection
-        @connection&.close
-        @connection = create_connection
+        @standard_connection&.close
+        @incremental_connection&.close
+        @standard_connection = nil
+        @incremental_connection = nil
       end
 
       def render_code(path, js_code, send_bundle)
@@ -27,7 +30,7 @@ def render_code_as_stream(path, js_code, is_rsc_payload:)
                 "rendering any RSC payload."
         end
 
-        ReactOnRailsPro::StreamRequest.create do |send_bundle|
+        ReactOnRailsPro::StreamRequest.create do |send_bundle, _barrier|
           if send_bundle
             Rails.logger.info { "[ReactOnRailsPro] Sending bundle to the node renderer" }
             upload_assets
@@ -38,6 +41,45 @@ def render_code_as_stream(path, js_code, is_rsc_payload:)
         end
       end
 
+      def render_code_with_incremental_updates(path, js_code, async_props_block:, is_rsc_payload:)
+        Rails.logger.info { "[ReactOnRailsPro] Perform incremental rendering request #{path}" }
+
+        # Determine bundle timestamp based on RSC support
+        pool = ReactOnRailsPro::ServerRenderingPool::NodeRenderingPool
+        bundle_timestamp = is_rsc_payload ? pool.rsc_bundle_hash : pool.server_bundle_hash
+
+        ReactOnRailsPro::StreamRequest.create do |send_bundle, barrier|
+          if send_bundle
+            Rails.logger.info { "[ReactOnRailsPro] Sending bundle to the node renderer" }
+            upload_assets
+          end
+
+          # Build bidirectional streaming request
+          request = incremental_connection.build_request(
+            "POST",
+            path,
+            headers: { "content-type" => "application/x-ndjson" },
+            body: []
+          )
+
+          # Create emitter and use it to generate initial request data
+          emitter = ReactOnRailsPro::AsyncPropsEmitter.new(bundle_timestamp, request)
+          initial_data = build_initial_incremental_request(js_code, emitter)
+          request.write("#{initial_data.to_json}\n")
+
+          response = incremental_connection.request(request, stream: true)
+
+          # Execute async props block in background using barrier
+          barrier.async do
+            async_props_block.call(emitter)
+          ensure
+            request.close
+          end
+
+          response
+        end
+      end
+
       def upload_assets
         Rails.logger.info { "[ReactOnRailsPro] Uploading assets" }
 
@@ -87,8 +129,14 @@ def asset_exists_on_vm_renderer?(filename)
 
       private
 
+      # rubocop:disable Naming/MemoizedInstanceVariableName
       def connection
-        @connection ||= create_connection
+        @standard_connection ||= create_standard_connection
+      end
+      # rubocop:enable Naming/MemoizedInstanceVariableName
+
+      def incremental_connection
+        @incremental_connection ||= create_incremental_connection
       end
 
       def perform_request(path, **post_options) # rubocop:disable Metrics/AbcSize,Metrics/CyclomaticComplexity
@@ -222,7 +270,22 @@ def common_form_data
         ReactOnRailsPro::Utils.common_form_data
       end
 
-      def create_connection # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
+      def build_initial_incremental_request(js_code, emitter)
+        common_form_data.merge(
+          renderingRequest: js_code,
+          onRequestClosedUpdateChunk: emitter.end_stream_chunk
+        )
+      end
+
+      def create_standard_connection
+        build_connection_config.plugin(:stream)
+      end
+
+      def create_incremental_connection
+        build_connection_config.plugin(:stream_bidi)
+      end
+
+      def build_connection_config # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
         url = ReactOnRailsPro.configuration.renderer_url
         Rails.logger.info do
           "[ReactOnRailsPro] Setting up Node Renderer connection to #{url}"
@@ -266,7 +329,6 @@ def create_connection # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
                                      nil
                                    end
           )
-          .plugin(:stream_bidi)
           # See https://www.rubydoc.info/gems/httpx/1.3.3/HTTPX%2FOptions:initialize for the available options
           .with(
             origin: url,

react_on_rails_pro/lib/react_on_rails_pro/stream_request.rb

@@ -1,5 +1,8 @@
 # frozen_string_literal: true
 
+require "async"
+require "async/barrier"
+
 module ReactOnRailsPro
   class StreamDecorator
     def initialize(component)
@@ -92,22 +95,28 @@ def initialize(&request_block)
     def each_chunk(&block)
       return enum_for(:each_chunk) unless block
 
-      send_bundle = false
-      error_body = +""
-      loop do
-        stream_response = @request_executor.call(send_bundle)
-
-        # Chunks can be merged during streaming, so we separate them by newlines
-        # Also, we check the status code inside the loop block because calling `status` outside the loop block
-        # is blocking, it will wait for the response to be fully received
-        # Look at the spec of `status` in `spec/react_on_rails_pro/stream_spec.rb` for more details
-        process_response_chunks(stream_response, error_body, &block)
-        break
-      rescue HTTPX::HTTPError => e
-        send_bundle = handle_http_error(e, error_body, send_bundle)
-      rescue HTTPX::ReadTimeoutError => e
-        raise ReactOnRailsPro::Error, "Time out error while server side render streaming a component.\n" \
-                                      "Original error:\n#{e}\n#{e.backtrace}"
+      Sync do
+        barrier = Async::Barrier.new
+
+        send_bundle = false
+        error_body = +""
+        loop do
+          stream_response = @request_executor.call(send_bundle, barrier)
+
+          # Chunks can be merged during streaming, so we separate them by newlines
+          # Also, we check the status code inside the loop block because calling `status` outside the loop block
+          # is blocking, it will wait for the response to be fully received
+          # Look at the spec of `status` in `spec/react_on_rails_pro/stream_spec.rb` for more details
+          process_response_chunks(stream_response, error_body, &block)
+          break
+        rescue HTTPX::HTTPError => e
+          send_bundle = handle_http_error(e, error_body, send_bundle)
+        rescue HTTPX::ReadTimeoutError => e
+          raise ReactOnRailsPro::Error, "Time out error while server side render streaming a component.\n" \
+                                        "Original error:\n#{e}\n#{e.backtrace}"
+        end
+
+        barrier.wait
       end
     end
 

react_on_rails_pro/spec/dummy/Procfile.dev

@@ -1,6 +1,6 @@
 # Procfile for development with hot reloading of JavaScript and CSS
 
-rails: rails s -p 3000
+# rails: rails s -p 3000
 
 # Run the hot reload server for client development
 webpack-dev-server: HMR=true bin/shakapacker-dev-server

react_on_rails_pro/spec/react_on_rails_pro/request_spec.rb

@@ -2,6 +2,9 @@
 
 require_relative "spec_helper"
 require "fakefs/safe"
+require "httpx"
+
+HTTPX::Plugins.load_plugin(:stream)
 
 describe ReactOnRailsPro::Request do
   let(:logger_mock) { instance_double(ActiveSupport::Logger).as_null_object }
@@ -218,4 +221,102 @@
       end
     end
   end
+
+  # Unverified doubles are required for HTTPX bidirectional streaming because:
+  # 1. HTTPX::StreamResponse doesn't define status in its interface (causes verified double failures)
+  # 2. The :stream_bidi plugin adds methods (#write, #close, #build_request) not in standard interfaces
+  # 3. Using double(ClassName) documents the class while allowing interface flexibility
+  # rubocop:disable RSpec/VerifiedDoubles, RSpec/MultipleMemoizedHelpers
+  describe "render_code_with_incremental_updates" do
+    let(:js_code) { "console.log('incremental rendering');" }
+    let(:async_props_block) { proc { |_emitter| } }
+    let(:mock_request) { double(HTTPX::Request) }
+    let(:mock_response) { double(HTTPX::StreamResponse, status: 200) }
+    let(:mock_connection) { double(HTTPX::Session) }
+
+    before do
+      allow(ReactOnRailsPro::ServerRenderingPool::NodeRenderingPool).to receive_messages(
+        server_bundle_hash: "server_bundle.js",
+        rsc_bundle_hash: "rsc_bundle.js"
+      )
+
+      allow(mock_connection).to receive_messages(build_request: mock_request, request: mock_response)
+      allow(mock_request).to receive(:close)
+      allow(mock_request).to receive(:write)
+      allow(mock_response).to receive(:is_a?).with(HTTPX::ErrorResponse).and_return(false)
+      allow(mock_response).to receive(:each).and_yield("chunk\n")
+      allow(described_class).to receive(:incremental_connection).and_return(mock_connection)
+
+      # Stub AsyncPropsEmitter to return a mock with end_stream_chunk
+      allow(ReactOnRailsPro::AsyncPropsEmitter).to receive(:new) do |_bundle_timestamp, _request|
+        double(
+          ReactOnRailsPro::AsyncPropsEmitter,
+          end_stream_chunk: { bundleTimestamp: "mocked", updateChunk: "mocked_js" }
+        )
+      end
+    end
+
+    it "creates NDJSON request with correct initial data" do
+      stream = described_class.render_code_with_incremental_updates(
+        "/render-incremental",
+        js_code,
+        async_props_block: async_props_block,
+        is_rsc_payload: false
+      )
+
+      stream.each_chunk(&:itself)
+
+      expect(mock_connection).to have_received(:build_request).with(
+        "POST",
+        "/render-incremental",
+        headers: { "content-type" => "application/x-ndjson" },
+        body: []
+      )
+      expect(mock_request).to have_received(:write).at_least(:once)
+    end
+
+    it "spawns barrier.async task and passes emitter to async_props_block" do
+      emitter_received = nil
+      test_async_props_block = proc { |emitter| emitter_received = emitter }
+
+      # Allow real emitter to be created for this test
+      allow(ReactOnRailsPro::AsyncPropsEmitter).to receive(:new).and_call_original
+
+      stream = described_class.render_code_with_incremental_updates(
+        "/render-incremental",
+        js_code,
+        async_props_block: test_async_props_block,
+        is_rsc_payload: false
+      )
+
+      stream.each_chunk(&:itself)
+
+      expect(emitter_received).to be_a(ReactOnRailsPro::AsyncPropsEmitter)
+    end
+
+    it "uses rsc_bundle_hash when is_rsc_payload is true" do
+      allow(ReactOnRailsPro.configuration).to receive(:enable_rsc_support).and_return(true)
+
+      emitter_captured = nil
+      allow(ReactOnRailsPro::AsyncPropsEmitter).to receive(:new) do |bundle_timestamp, request_stream|
+        emitter_captured = { bundle_timestamp: bundle_timestamp, request_stream: request_stream }
+        double(
+          ReactOnRailsPro::AsyncPropsEmitter,
+          end_stream_chunk: { bundleTimestamp: bundle_timestamp, updateChunk: "mocked_js" }
+        )
+      end
+
+      stream = described_class.render_code_with_incremental_updates(
+        "/render-incremental",
+        js_code,
+        async_props_block: async_props_block,
+        is_rsc_payload: true
+      )
+
+      stream.each_chunk(&:itself)
+
+      expect(emitter_captured[:bundle_timestamp]).to eq("rsc_bundle.js")
+    end
+  end
+  # rubocop:enable RSpec/VerifiedDoubles, RSpec/MultipleMemoizedHelpers
 end

react_on_rails_pro/spec/react_on_rails_pro/stream_request_spec.rb

@@ -2,6 +2,10 @@
 
 require_relative "spec_helper"
 require "react_on_rails_pro/stream_request"
+require "async/barrier"
+require "httpx"
+
+HTTPX::Plugins.load_plugin(:stream)
 
 RSpec.describe ReactOnRailsPro::StreamRequest do
   describe ".create" do
@@ -10,4 +14,44 @@
       expect(result).to be_a(ReactOnRailsPro::StreamDecorator)
     end
   end
+
+  # Unverified doubles are required for streaming responses because:
+  # 1. HTTP streaming responses don't have a dedicated class type in HTTPX
+  # 2. The #each method for streaming is added dynamically at runtime
+  # 3. The interface varies based on the streaming mode (HTTP/2, chunked, etc.)
+  # rubocop:disable RSpec/VerifiedDoubles
+  describe "#each_chunk with barrier" do
+    it "passes barrier to request_executor block" do
+      barrier_received = nil
+      mock_response = double(HTTPX::StreamResponse, status: 200)
+      allow(mock_response).to receive(:is_a?).with(HTTPX::ErrorResponse).and_return(false)
+      allow(mock_response).to receive(:each).and_yield("chunk\n")
+
+      stream = described_class.create do |_send_bundle, barrier|
+        barrier_received = barrier
+        mock_response
+      end
+
+      stream.each_chunk(&:itself)
+
+      expect(barrier_received).to be_a(Async::Barrier)
+    end
+
+    it "calls barrier.wait after yielding chunks" do
+      barrier = Async::Barrier.new
+      allow(Async::Barrier).to receive(:new).and_return(barrier)
+      expect(barrier).to receive(:wait)
+
+      mock_response = double(HTTPX::StreamResponse, status: 200)
+      allow(mock_response).to receive(:is_a?).with(HTTPX::ErrorResponse).and_return(false)
+      allow(mock_response).to receive(:each).and_yield("chunk\n")
+
+      stream = described_class.create do |_send_bundle, _barrier|
+        mock_response
+      end
+
+      stream.each_chunk(&:itself)
+    end
+  end
+  # rubocop:enable RSpec/VerifiedDoubles
 end