Get ready to merge into Rails

1 files changed, 114 insertions, 0 deletions

diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb
new file mode 100644
index 0000000000..b5ffa17f72
--- /dev/null
+++ b/actioncable/lib/action_cable/channel/streams.rb
@@ -0,0 +1,114 @@
+module ActionCable
+  module Channel
+    # Streams allow channels to route broadcastings to the subscriber. A broadcasting is, as discussed elsewhere, a pub/sub queue where any data
+    # put into it is automatically sent to the clients that are connected at that time. It's purely an online queue, though. If you're not
+    # streaming a broadcasting at the very moment it sends out an update, you'll not get that update when connecting later.
+    #
+    # Most commonly, the streamed broadcast is sent straight to the subscriber on the client-side. The channel just acts as a connector between
+    # the two parties (the broadcaster and the channel subscriber). Here's an example of a channel that allows subscribers to get all new
+    # comments on a given page:
+    #
+    #   class CommentsChannel < ApplicationCable::Channel
+    #     def follow(data)
+    #       stream_from "comments_for_#{data['recording_id']}"
+    #     end
+    #
+    #     def unfollow
+    #       stop_all_streams
+    #     end
+    #   end
+    #
+    # So the subscribers of this channel will get whatever data is put into the, let's say, `comments_for_45` broadcasting as soon as it's put there.
+    # That looks like so from that side of things:
+    #
+    #   ActionCable.server.broadcast "comments_for_45", author: 'DHH', content: 'Rails is just swell'
+    #
+    # If you have a stream that is related to a model, then the broadcasting used can be generated from the model and channel.
+    # The following example would subscribe to a broadcasting like `comments:Z2lkOi8vVGVzdEFwcC9Qb3N0LzE`
+    #
+    #   class CommentsChannel < ApplicationCable::Channel
+    #     def subscribed
+    #       post = Post.find(params[:id])
+    #       stream_for post
+    #     end
+    #   end
+    #
+    # You can then broadcast to this channel using:
+    #
+    #   CommentsChannel.broadcast_to(@post, @comment)
+    #
+    # If you don't just want to parlay the broadcast unfiltered to the subscriber, you can supply a callback that lets you alter what goes out.
+    # Example below shows how you can use this to provide performance introspection in the process:
+    #
+    #   class ChatChannel < ApplicationCable::Channel
+    #    def subscribed
+    #      @room = Chat::Room[params[:room_number]]
+    #
+    #      stream_for @room, -> (encoded_message) do
+    #        message = ActiveSupport::JSON.decode(encoded_message)
+    #
+    #        if message['originated_at'].present?
+    #          elapsed_time = (Time.now.to_f - message['originated_at']).round(2)
+    #
+    #          ActiveSupport::Notifications.instrument :performance, measurement: 'Chat.message_delay', value: elapsed_time, action: :timing
+    #          logger.info "Message took #{elapsed_time}s to arrive"
+    #        end
+    #
+    #        transmit message
+    #      end
+    #    end
+    #
+    # You can stop streaming from all broadcasts by calling #stop_all_streams.
+    module Streams
+      extend ActiveSupport::Concern
+
+      included do
+        on_unsubscribe :stop_all_streams
+      end
+
+      # Start streaming from the named <tt>broadcasting</tt> pubsub queue. Optionally, you can pass a <tt>callback</tt> that'll be used
+      # instead of the default of just transmitting the updates straight to the subscriber.
+      def stream_from(broadcasting, callback = nil)
+        # Hold off the confirmation until pubsub#subscribe is successful
+        defer_subscription_confirmation!
+
+        callback ||= default_stream_callback(broadcasting)
+        streams << [ broadcasting, callback ]
+
+        EM.next_tick do
+          pubsub.subscribe(broadcasting, &callback).callback do |reply|
+            transmit_subscription_confirmation
+            logger.info "#{self.class.name} is streaming from #{broadcasting}"
+          end
+        end
+      end
+
+      # Start streaming the pubsub queue for the <tt>model</tt> in this channel. Optionally, you can pass a
+      # <tt>callback</tt> that'll be used instead of the default of just transmitting the updates straight
+      # to the subscriber.
+      def stream_for(model, callback = nil)
+        stream_from(broadcasting_for([ channel_name, model ]), callback)
+      end
+
+      def stop_all_streams
+        streams.each do |broadcasting, callback|
+          pubsub.unsubscribe_proc broadcasting, callback
+          logger.info "#{self.class.name} stopped streaming from #{broadcasting}"
+        end.clear
+      end
+
+      private
+        delegate :pubsub, to: :connection
+
+        def streams
+          @_streams ||= []
+        end
+
+        def default_stream_callback(broadcasting)
+          -> (message) do
+            transmit ActiveSupport::JSON.decode(message), via: "streamed from #{broadcasting}"
+          end
+        end
+    end
+  end
+end