Controller actions are processed in a separate thread for live

responses.

Processing controller actions in a separate thread allows us to work
around the rack api - we can allow the user to set status and headers,
then block until the first bytes are written.  As soon as the first
bytes are written, the main thread can return the status, headers, and
(essentially) a queue for the body.

1 files changed, 69 insertions, 3 deletions

diff --git a/actionpack/test/controller/live_stream_test.rb b/actionpack/test/controller/live_stream_test.rb
index 6ee6444065..5bb37a271d 100644
--- a/actionpack/test/controller/live_stream_test.rb
+++ b/actionpack/test/controller/live_stream_test.rb
@@ -1,16 +1,41 @@
 require 'abstract_unit'
+require 'active_support/concurrency/latch'
 
 module ActionController
-  class StreamingResponseTest < ActionController::TestCase
+  class LiveStreamTest < ActionController::TestCase
     class TestController < ActionController::Base
+      include ActionController::Live
+
+      attr_accessor :latch, :tc
+
       def self.controller_path
         'test'
       end
 
       def basic_stream
+        response.headers['Content-Type'] = 'text/event-stream'
+        %w{ hello world }.each do |word|
+          response.stream.write word
+        end
+        response.stream.close
+      end
+
+      def blocking_stream
+        response.headers['Content-Type'] = 'text/event-stream'
+        %w{ hello world }.each do |word|
+          response.stream.write word
+          latch.await
+        end
+        response.stream.close
+      end
+
+      def thread_locals
+        tc.assert_equal 'aaron', Thread.current[:setting]
+        tc.refute_equal Thread.current.object_id, Thread.current[:originating_thread]
+
+        response.headers['Content-Type'] = 'text/event-stream'
         %w{ hello world }.each do |word|
           response.stream.write word
-          response.stream.write "\n"
         end
         response.stream.close
       end
@@ -18,9 +43,50 @@ module ActionController
 
     tests TestController
 
+    class TestResponse < Live::Response
+      def recycle!
+        initialize
+      end
+    end
+
+    def build_response
+      TestResponse.new
+    end
+
     def test_write_to_stream
+      @controller = TestController.new
       get :basic_stream
-      assert_equal "hello\nworld\n", @response.body
+      assert_equal "helloworld", @response.body
+      assert_equal 'text/event-stream', @response.headers['Content-Type']
+    end
+
+    def test_async_stream
+      @controller.latch = ActiveSupport::Concurrency::Latch.new
+      parts             = ['hello', 'world']
+
+      @controller.request  = @request
+      @controller.response = @response
+
+      t = Thread.new(@response) { |resp|
+        resp.stream.each do |part|
+          assert_equal parts.shift, part
+          ol = @controller.latch
+          @controller.latch = ActiveSupport::Concurrency::Latch.new
+          ol.release
+        end
+      }
+
+      @controller.process :blocking_stream
+
+      assert t.join
+    end
+
+    def test_thread_locals_get_copied
+      @controller.tc = self
+      Thread.current[:originating_thread] = Thread.current.object_id
+      Thread.current[:setting]            = 'aaron'
+
+      get :thread_locals
     end
   end
 end