1 files changed, 105 insertions, 0 deletions
diff --git a/actionpack/test/controller/live_stream_test.rb b/actionpack/test/controller/live_stream_test.rb
index 947f64176b..0500b7c789 100644
--- a/actionpack/test/controller/live_stream_test.rb
+++ b/actionpack/test/controller/live_stream_test.rb
@@ -39,6 +39,13 @@ module ActionController
       ensure
         sse.close
       end
+
+      def sse_with_multiple_line_message
+        sse = SSE.new(response.stream)
+        sse.write("first line.\nsecond line.")
+      ensure
+        sse.close
+      end
     end
 
     tests SSETestController
@@ -87,6 +94,15 @@ module ActionController
       assert_match(/data: {\"name\":\"Ryan\"}/, second_response)
       assert_match(/id: 2/, second_response)
     end
+
+    def test_sse_with_multiple_line_message
+      get :sse_with_multiple_line_message
+
+      wait_for_response_stream_close
+      first_response, second_response = response.body.split("\n")
+      assert_match(/data: first line/, first_response)
+      assert_match(/data: second line/, second_response)
+    end
   end
 
   class LiveStreamTest < ActionController::TestCase
@@ -186,6 +202,39 @@ module ActionController
         response.stream.write ''
         response.stream.write params[:widget][:didnt_check_for_nil]
       end
+
+      def overfill_buffer_and_die
+        # Write until the buffer is full. It doesn't expose that
+        # information directly, so we must hard-code its size:
+        10.times do
+          response.stream.write '.'
+        end
+        # .. plus one more, because the #each frees up a slot:
+        response.stream.write '.'
+
+        latch.release
+
+        # This write will block, and eventually raise
+        response.stream.write 'x'
+
+        20.times do
+          response.stream.write '.'
+        end
+      end
+
+      def ignore_client_disconnect
+        response.stream.ignore_disconnect = true
+
+        response.stream.write '' # commit
+
+        # These writes will be ignored
+        15.times do
+          response.stream.write 'x'
+        end
+
+        logger.info 'Work complete'
+        latch.release
+      end
     end
 
     tests TestController
@@ -248,6 +297,62 @@ module ActionController
       assert t.join(3), 'timeout expired before the thread terminated'
     end
 
+    def test_abort_with_full_buffer
+      @controller.latch = ActiveSupport::Concurrency::Latch.new
+
+      @request.parameters[:format] = 'plain'
+      @controller.request  = @request
+      @controller.response = @response
+
+      got_error = ActiveSupport::Concurrency::Latch.new
+      @response.stream.on_error do
+        ActionController::Base.logger.warn 'Error while streaming'
+        got_error.release
+      end
+
+      t = Thread.new(@response) { |resp|
+        resp.await_commit
+        _, _, body = resp.to_a
+        body.each do |part|
+          @controller.latch.await
+          body.close
+          break
+        end
+      }
+
+      capture_log_output do |output|
+        @controller.process :overfill_buffer_and_die
+        t.join
+        got_error.await
+        assert_match 'Error while streaming', output.rewind && output.read
+      end
+    end
+
+    def test_ignore_client_disconnect
+      @controller.latch = ActiveSupport::Concurrency::Latch.new
+
+      @controller.request  = @request
+      @controller.response = @response
+
+      t = Thread.new(@response) { |resp|
+        resp.await_commit
+        _, _, body = resp.to_a
+        body.each do |part|
+          body.close
+          break
+        end
+      }
+
+      capture_log_output do |output|
+        @controller.process :ignore_client_disconnect
+        t.join
+        Timeout.timeout(3) do
+          @controller.latch.await
+        end
+        assert_match 'Work complete', output.rewind && output.read
+      end
+    end
+
     def test_thread_locals_get_copied
       @controller.tc = self
       Thread.current[:originating_thread] = Thread.current.object_id