Creating an SSE class to be used with ActionController::Live.

actionpack/lib/action_controller/metal/live.rb

@@ -1,5 +1,6 @@
 require 'action_dispatch/http/response'
 require 'delegate'
+require 'active_support/json'
 
 module ActionController
   # Mix this module in to your controller, and all actions in that controller
@@ -32,6 +33,79 @@ module ActionController
   # the main thread. Make sure your actions are thread safe, and this shouldn't
   # be a problem (don't share state across threads, etc).
   module Live
+    # This class provides the ability to write an SSE (Server Sent Event)
+    # to an IO stream. The class is initialized with a stream and can be used
+    # to either write a JSON string or an object which can be converted to JSON.
+    #
+    # Writing an object will convert it into standard SSE format with whatever
+    # options you have configured. You may choose to set the following options:
+    #
+    #   1) Event. If specified, an event with this name will be dispatched on
+    #   the browser.
+    #   2) Retry. The reconnection time in milliseconds used when attempting
+    #   to send the event.
+    #   3) Id. If the connection dies while sending an SSE to the browser, then
+    #   the server will receive a +Last-Event-ID+ header with value equal to +id+.
+    #
+    # After setting an option in the constructor of the SSE object, all future
+    # SSEs sent accross the stream will use those options unless overridden.
+    #
+    # Example Usage:
+    #
+    #   class MyController < ActionController::Base
+    #     include ActionController::Live
+    #
+    #     def index
+    #       response.headers['Content-Type'] = 'text/event-stream'
+    #       sse = SSE.new(response.stream, retry: 300, event: "event-name")
+    #       sse.write({ name: 'John'})
+    #       sse.write({ name: 'John'}, id: 10)
+    #       sse.write({ name: 'John'}, id: 10, event: "other-event")
+    #       sse.write({ name: 'John'}, id: 10, event: "other-event", retry: 500)
+    #     ensure
+    #       sse.close
+    #     end
+    #   end
+    #
+    # Note: SSEs are not currently supported by IE. However, they are supported
+    # by Chrome, Firefox, Opera, and Safari.
+    class SSE
+
+      WHITELISTED_OPTIONS = %w( retry event id )
+
+      def initialize(stream, options = {})
+        @stream = stream
+        @options = options
+      end
+
+      def close
+        @stream.close
+      end
+
+      def write(object, options = {})
+        case object
+        when String
+          perform_write(object, options)
+        else
+          perform_write(ActiveSupport::JSON.encode(object), options)
+        end
+      end
+
+      private
+
+        def perform_write(json, options)
+          current_options = @options.merge(options).stringify_keys
+
+          WHITELISTED_OPTIONS.each do |option_name|
+            if (option_value = current_options[option_name])
+              @stream.write "#{option_name}: #{option_value}\n"
+            end
+          end
+
+          @stream.write "data: #{json}\n\n"
+        end
+    end
+
     class Buffer < ActionDispatch::Response::Buffer #:nodoc:
       def initialize(response)
         @error_callback = nil

actionpack/test/controller/live_stream_test.rb

@@ -2,6 +2,94 @@
 require 'active_support/concurrency/latch'
 
 module ActionController
+  class SSETest < ActionController::TestCase
+    class SSETestController < ActionController::Base
+      include ActionController::Live
+
+      def basic_sse
+        response.headers['Content-Type'] = 'text/event-stream'
+        sse = SSE.new(response.stream)
+        sse.write("{\"name\":\"John\"}")
+        sse.write({ name: "Ryan" })
+      ensure
+        sse.close
+      end
+
+      def sse_with_event
+        sse = SSE.new(response.stream, event: "send-name")
+        sse.write("{\"name\":\"John\"}")
+        sse.write({ name: "Ryan" })
+      ensure
+        sse.close
+      end
+
+      def sse_with_retry
+        sse = SSE.new(response.stream, retry: 1000)
+        sse.write("{\"name\":\"John\"}")
+        sse.write({ name: "Ryan" }, retry: 1500)
+      ensure
+        sse.close
+      end
+
+      def sse_with_id
+        sse = SSE.new(response.stream)
+        sse.write("{\"name\":\"John\"}", id: 1)
+        sse.write({ name: "Ryan" }, id: 2)
+      ensure
+        sse.close
+      end
+    end
+
+    tests SSETestController
+
+    def wait_for_response_stream_close
+      while !response.stream.closed?
+        sleep 0.01
+      end
+    end
+
+    def test_basic_sse
+      get :basic_sse
+
+      wait_for_response_stream_close
+      assert_match(/data: {\"name\":\"John\"}/, response.body)
+      assert_match(/data: {\"name\":\"Ryan\"}/, response.body)
+    end
+
+    def test_sse_with_event_name
+      get :sse_with_event
+
+      wait_for_response_stream_close
+      assert_match(/data: {\"name\":\"John\"}/, response.body)
+      assert_match(/data: {\"name\":\"Ryan\"}/, response.body)
+      assert_match(/event: send-name/, response.body)
+    end
+
+    def test_sse_with_retry
+      get :sse_with_retry
+
+      wait_for_response_stream_close
+      first_response, second_response = response.body.split("\n\n")
+      assert_match(/data: {\"name\":\"John\"}/, first_response)
+      assert_match(/retry: 1000/, first_response)
+
+      assert_match(/data: {\"name\":\"Ryan\"}/, second_response)
+      assert_match(/retry: 1500/, second_response)
+    end
+
+    def test_sse_with_id
+      get :sse_with_id
+
+      wait_for_response_stream_close
+      first_response, second_response = response.body.split("\n\n")
+      assert_match(/data: {\"name\":\"John\"}/, first_response)
+      assert_match(/id: 1/, first_response)
+
+      assert_match(/data: {\"name\":\"Ryan\"}/, second_response)
+      assert_match(/id: 2/, second_response)
+    end
+  end
+
   class LiveStreamTest < ActionController::TestCase
     class TestController < ActionController::Base
       include ActionController::Live