Replace manual Fibers with Async gem primitives for component streaming

react_on_rails_pro/app/helpers/react_on_rails_pro_helper.rb

@@ -239,36 +239,41 @@ def fetch_stream_react_component(component_name, raw_options, &block)
   end
 
   def handle_stream_cache_hit(component_name, raw_options, auto_load_bundle, cached_chunks)
+    require "async/variable"
+
     render_options = ReactOnRails::ReactComponent::RenderOptions.new(
       react_component_name: component_name,
       options: { auto_load_bundle: auto_load_bundle }.merge(raw_options)
     )
     load_pack_for_generated_component(component_name, render_options)
 
     initial_result, *rest_chunks = cached_chunks
-    hit_fiber = Fiber.new do
-      rest_chunks.each { |chunk| Fiber.yield(chunk) }
-      nil
+    first_chunk_var = Async::Variable.new
+
+    # Create async task to replay cached chunks
+    @async_barrier.async do
+      # Set first chunk (already have it from cache)
+      first_chunk_var.value = initial_result
+
+      # Enqueue remaining cached chunks
+      rest_chunks.each { |chunk| @main_output_queue.enqueue(chunk) }
     end
-    @rorp_rendering_fibers << hit_fiber
-    initial_result
+
+    # Wait for and return first chunk
+    first_chunk_var.wait
+    first_chunk_var.value
   end
 
   def handle_stream_cache_miss(component_name, raw_options, auto_load_bundle, view_cache_key, &block)
-    # Kick off the normal streaming helper to get the initial result and the original fiber
+    # Enable caching mode
+    @rorp_cache_buffering = { key: view_cache_key, chunks: [], options: raw_options[:cache_options] || {} }
+
+    # Call normal streaming (which will buffer chunks due to @rorp_cache_buffering flag)
     initial_result = render_stream_component_with_props(component_name, raw_options, auto_load_bundle, &block)
-    original_fiber = @rorp_rendering_fibers.pop
 
-    buffered_chunks = [initial_result]
-    wrapper_fiber = Fiber.new do
-      while (chunk = original_fiber.resume)
-        buffered_chunks << chunk
-        Fiber.yield(chunk)
-      end
-      Rails.cache.write(view_cache_key, buffered_chunks, raw_options[:cache_options] || {})
-      nil
-    end
-    @rorp_rendering_fibers << wrapper_fiber
+    # Disable caching mode
+    @rorp_cache_buffering = nil
+
     initial_result
   end
 
@@ -292,24 +297,50 @@ def check_caching_options!(raw_options, block)
   end
 
   def run_stream_inside_fiber
-    if @rorp_rendering_fibers.nil?
+    require "async/variable"
+
+    if @async_barrier.nil?
       raise ReactOnRails::Error,
             "You must call stream_view_containing_react_components to render the view containing the react component"
     end
 
-    rendering_fiber = Fiber.new do
+    # Create a variable to hold the first chunk for synchronous return
+    first_chunk_var = Async::Variable.new
+
+    # Check if we're in caching mode
+    cache_config = @rorp_cache_buffering
+
+    # Start an async task on the barrier to stream all chunks
+    @async_barrier.async do
       stream = yield
+      is_first = true
+
       stream.each_chunk do |chunk|
-        Fiber.yield chunk
+        # Buffer chunk for caching if enabled
+        cache_config[:chunks] << chunk if cache_config
+
+        if is_first
+          # Store first chunk in variable for synchronous access
+          first_chunk_var.value = chunk
+          is_first = false
+        else
+          # Enqueue remaining chunks to main output queue
+          @main_output_queue.enqueue(chunk)
+        end
       end
-    end
 
-    @rorp_rendering_fibers << rendering_fiber
+      # Handle case where stream has no chunks
+      first_chunk_var.value = nil if is_first
+
+      # Write to cache if caching is enabled
+      if cache_config
+        Rails.cache.write(cache_config[:key], cache_config[:chunks], cache_config[:options])
+      end
+    end
 
-    # return the first chunk of the fiber
-    # It contains the initial html of the component
-    # all updates will be appended to the stream sent to browser
-    rendering_fiber.resume
+    # Wait for and return the first chunk (blocking)
+    first_chunk_var.wait
+    first_chunk_var.value
   end
 
   def internal_stream_react_component(component_name, options = {})

react_on_rails_pro/lib/react_on_rails_pro/concerns/stream.rb

@@ -31,73 +31,47 @@ module Stream
     #
     # @see ReactOnRails::Helper#stream_react_component
     def stream_view_containing_react_components(template:, close_stream_at_end: true, **render_options)
-      @rorp_rendering_fibers = []
-      template_string = render_to_string(template: template, **render_options)
-      # View may contain extra newlines, chunk already contains a newline
-      # Having multiple newlines between chunks causes hydration errors
-      # So we strip extra newlines from the template string and add a single newline
-      response.stream.write(template_string)
-
-      begin
-        drain_streams_concurrently
-      ensure
-        response.stream.close if close_stream_at_end
-      end
-    end
-
-    private
-
-    def drain_streams_concurrently
       require "async"
+      require "async/barrier"
       require "async/limited_queue"
 
-      return if @rorp_rendering_fibers.empty?
-
-      Sync do |parent|
-        # To avoid memory bloat, we use a limited queue to buffer chunks in memory.
+      Sync do |parent_task|
+        # Initialize async primitives for concurrent component streaming
+        @async_barrier = Async::Barrier.new
         buffer_size = ReactOnRailsPro.configuration.concurrent_component_streaming_buffer_size
-        queue = Async::LimitedQueue.new(buffer_size)
+        @main_output_queue = Async::LimitedQueue.new(buffer_size)
 
-        writer = build_writer_task(parent: parent, queue: queue)
-        tasks = build_producer_tasks(parent: parent, queue: queue)
+        # Render template - components will start streaming immediately
+        template_string = render_to_string(template: template, **render_options)
+        # View may contain extra newlines, chunk already contains a newline
+        # Having multiple newlines between chunks causes hydration errors
+        # So we strip extra newlines from the template string and add a single newline
+        response.stream.write(template_string)
 
-        # This structure ensures that even if a producer task fails, we always
-        # signal the writer to stop and then wait for it to finish draining
-        # any remaining items from the queue before propagating the error.
         begin
-          tasks.each(&:wait)
+          drain_streams_concurrently(parent_task)
         ensure
-          # `close` signals end-of-stream; when writer tries to dequeue, it will get nil, so it will exit.
-          queue.close
-          writer.wait
+          response.stream.close if close_stream_at_end
         end
       end
     end
 
-    def build_producer_tasks(parent:, queue:)
-      @rorp_rendering_fibers.each_with_index.map do |fiber, idx|
-        parent.async do
-          loop do
-            chunk = fiber.resume
-            break unless chunk
+    private
 
-            # Will be blocked if the queue is full until a chunk is dequeued
-            queue.enqueue([idx, chunk])
-          end
+    def drain_streams_concurrently(parent_task)
+      writing_task = parent_task.async do
+        # Drain all remaining chunks from the queue to the response stream
+        while (chunk = @main_output_queue.dequeue)
+          response.stream.write(chunk)
         end
       end
-    end
 
-    def build_writer_task(parent:, queue:)
-      parent.async do
-        loop do
-          pair = queue.dequeue
-          break if pair.nil?
+      # Wait for all component streaming tasks to complete
+      @async_barrier.wait
 
-          _idx_from_queue, item = pair
-          response.stream.write(item)
-        end
-      end
+      # Close the queue to signal end of streaming
+      @main_output_queue.close
+      writing_task.wait
     end
   end
 end

react_on_rails_pro/spec/react_on_rails_pro/stream_spec.rb

@@ -2,6 +2,7 @@
 
 require "async"
 require "async/queue"
+require "async/variable"
 require_relative "spec_helper"
 
 class StreamController
@@ -15,13 +16,16 @@ def initialize(component_queues:, initial_response: "TEMPLATE")
   end
 
   def render_to_string(**_opts)
-    @rorp_rendering_fibers = @component_queues.map do |queue|
-      Fiber.new do
+    # Simulate component helpers creating async tasks
+    # In real implementation, first chunks are part of template HTML
+    # For testing, we enqueue all chunks including first ones
+    @component_queues.each do |queue|
+      @async_barrier.async do
         loop do
           chunk = queue.dequeue
           break if chunk.nil?
 
-          Fiber.yield chunk
+          @main_output_queue.enqueue(chunk)
         end
       end
     end