add broadcast_to and stream_for methods as per #26

11 files changed, 139 insertions, 14 deletions

diff --git a/lib/action_cable/channel.rb b/lib/action_cable/channel.rb
index 9e4d3d3f93..3b973ba0a7 100644
--- a/lib/action_cable/channel.rb
+++ b/lib/action_cable/channel.rb
@@ -1,7 +1,9 @@
 module ActionCable
   module Channel
     autoload :Base, 'action_cable/channel/base'
+    autoload :Broadcasting, 'action_cable/channel/broadcasting'
     autoload :Callbacks, 'action_cable/channel/callbacks'
+    autoload :Naming, 'action_cable/channel/naming'
     autoload :PeriodicTimers, 'action_cable/channel/periodic_timers'
     autoload :Streams, 'action_cable/channel/streams'
   end
diff --git a/lib/action_cable/channel/base.rb b/lib/action_cable/channel/base.rb
index 87ae3a1211..c83c3b74fd 100644
--- a/lib/action_cable/channel/base.rb
+++ b/lib/action_cable/channel/base.rb
@@ -68,6 +68,8 @@ module ActionCable
       include Callbacks
       include PeriodicTimers
       include Streams
+      include Naming
+      include Broadcasting
 
       on_subscribe   :subscribed
       on_unsubscribe :unsubscribed
diff --git a/lib/action_cable/channel/broadcasting.rb b/lib/action_cable/channel/broadcasting.rb
new file mode 100644
index 0000000000..ee4117bc0a
--- /dev/null
+++ b/lib/action_cable/channel/broadcasting.rb
@@ -0,0 +1,27 @@
+module ActionCable
+  module Channel
+    module Broadcasting
+      extend ActiveSupport::Concern
+
+      delegate :broadcasting_for, to: :class
+
+      class_methods do
+        # Broadcast a hash to a unique broadcasting for this <tt>model</tt> in this channel.
+        def broadcast_to(model, message)
+          ActionCable.server.broadcast(broadcasting_for([ channel_name, model ]), message)
+        end
+
+        def broadcasting_for(model) #:nodoc:
+          case
+          when model.is_a?(Array)
+            model.map { |m| broadcasting_for(m) }.join(':')
+          when model.respond_to?(:to_gid_param)
+            model.to_gid_param
+          else
+            model.to_param
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/action_cable/channel/naming.rb b/lib/action_cable/channel/naming.rb
new file mode 100644
index 0000000000..4c9d53b15a
--- /dev/null
+++ b/lib/action_cable/channel/naming.rb
@@ -0,0 +1,22 @@
+module ActionCable
+  module Channel
+    module Naming
+      extend ActiveSupport::Concern
+
+      class_methods do
+        # Returns the name of the channel, underscored, without the <tt>Channel</tt> ending.
+        # If the channel is in a namespace, then the namespaces are represented by single
+        # colon separators in the channel name.
+        #
+        #   ChatChannel.channel_name # => 'chat'
+        #   Chats::AppearancesChannel.channel_name # => 'chats:appearances'
+        def channel_name
+          @channel_name ||= name.sub(/Channel$/, '').gsub('::',':').underscore
+        end
+      end
+
+      # Delegates to the class' <tt>channel_name</tt>
+      delegate :channel_name, to: :class
+    end
+  end
+end
diff --git a/lib/action_cable/channel/streams.rb b/lib/action_cable/channel/streams.rb
index 6a3dc76c1d..f711b065ca 100644
--- a/lib/action_cable/channel/streams.rb
+++ b/lib/action_cable/channel/streams.rb
@@ -1,7 +1,7 @@
 module ActionCable
   module Channel
     # Streams allow channels to route broadcastings to the subscriber. A broadcasting is an 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 
+    # 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
@@ -12,7 +12,7 @@ module ActionCable
     #     def follow(data)
     #       stream_from "comments_for_#{data['recording_id']}"
     #     end
-    #   
+    #
     #     def unfollow
     #       stop_all_streams
     #     end
@@ -23,23 +23,37 @@ module ActionCable
     #
     #   ActionCable.server.broadcast "comments_for_45", author: 'DHH', content: 'Rails is just swell'
     #
-    # If you don't just want to parlay the broadcast unfiltered to the subscriber, you can supply a callback that let's you alter what goes out. 
+    # 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 to subscribe to a broadcasting that would be something 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)
+    #
+    # If you don't just want to parlay the broadcast unfiltered to the subscriber, you can supply a callback that let's 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_from @room.channel, -> (message) do
+    #
+    #      stream_for @room, -> (message) do
     #        message = ActiveSupport::JSON.decode(m)
-    #  
+    #
     #        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
@@ -63,6 +77,13 @@ module ActionCable
         logger.info "#{self.class.name} is streaming from #{broadcasting}"
       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
diff --git a/lib/action_cable/server/base.rb b/lib/action_cable/server/base.rb
index fe7966e090..b09fbf6da4 100644
--- a/lib/action_cable/server/base.rb
+++ b/lib/action_cable/server/base.rb
@@ -9,7 +9,7 @@ module ActionCable
       include ActionCable::Server::Connections
 
       cattr_accessor(:config, instance_accessor: true) { ActionCable::Server::Configuration.new }
-      
+
       def self.logger; config.logger; end
       delegate :logger, to: :config
 
@@ -56,7 +56,7 @@ module ActionCable
             logger.info "[ActionCable] Redis reconnect failed."
             # logger.info "[ActionCable] Redis reconnected. Closing all the open connections."
             # @connections.map &:close
-          end            
+          end
         end
       end
 
diff --git a/lib/action_cable/server/broadcasting.rb b/lib/action_cable/server/broadcasting.rb
index 4f72ffd96f..037b98951e 100644
--- a/lib/action_cable/server/broadcasting.rb
+++ b/lib/action_cable/server/broadcasting.rb
@@ -32,7 +32,7 @@ module ActionCable
       # The redis instance used for broadcasting. Not intended for direct user use.
       def broadcasting_redis
         @broadcasting_redis ||= Redis.new(config.redis)
-      end      
+      end
 
       private
         class Broadcaster
diff --git a/test/channel/broadcasting_test.rb b/test/channel/broadcasting_test.rb
new file mode 100644
index 0000000000..1de04243e5
--- /dev/null
+++ b/test/channel/broadcasting_test.rb
@@ -0,0 +1,29 @@
+require 'test_helper'
+require 'stubs/test_connection'
+require 'stubs/room'
+
+class ActionCable::Channel::BroadcastingTest < ActiveSupport::TestCase
+  class ChatChannel < ActionCable::Channel::Base
+  end
+
+  setup do
+    @connection = TestConnection.new
+  end
+
+  test "broadcasts_to" do
+    ActionCable.stubs(:server).returns mock().tap { |m| m.expects(:broadcast).with('action_cable:channel:broadcasting_test:chat:Room#1-Campfire', "Hello World") }
+    ChatChannel.broadcast_to(Room.new(1), "Hello World")
+  end
+
+  test "broadcasting_for with an object" do
+    assert_equal "Room#1-Campfire", ChatChannel.broadcasting_for(Room.new(1))
+  end
+
+  test "broadcasting_for with an array" do
+    assert_equal "Room#1-Campfire:Room#2-Campfire", ChatChannel.broadcasting_for([ Room.new(1), Room.new(2) ])
+  end
+
+  test "broadcasting_for with a string" do
+    assert_equal "hello", ChatChannel.broadcasting_for("hello")
+  end
+end
diff --git a/test/channel/naming_test.rb b/test/channel/naming_test.rb
new file mode 100644
index 0000000000..89ef6ad8b0
--- /dev/null
+++ b/test/channel/naming_test.rb
@@ -0,0 +1,10 @@
+require 'test_helper'
+
+class ActionCable::Channel::NamingTest < ActiveSupport::TestCase
+  class ChatChannel < ActionCable::Channel::Base
+  end
+
+  test "channel_name" do
+    assert_equal "action_cable:channel:naming_test:chat", ChatChannel.channel_name
+  end
+end
diff --git a/test/channel/stream_test.rb b/test/channel/stream_test.rb
index 1a8c259d11..b0a6f49072 100644
--- a/test/channel/stream_test.rb
+++ b/test/channel/stream_test.rb
@@ -5,8 +5,10 @@ require 'stubs/room'
 class ActionCable::Channel::StreamTest < ActiveSupport::TestCase
   class ChatChannel < ActionCable::Channel::Base
     def subscribed
-      @room = Room.new params[:id]
-      stream_from "test_room_#{@room.id}"
+      if params[:id]
+        @room = Room.new params[:id]
+        stream_from "test_room_#{@room.id}"
+      end
     end
   end
 
@@ -15,10 +17,16 @@ class ActionCable::Channel::StreamTest < ActiveSupport::TestCase
   end
 
   test "streaming start and stop" do
-    @connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe) }
+    @connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1") }
     channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
 
     @connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe_proc) }
     channel.unsubscribe_from_channel
   end
+
+  test "stream_for" do
+    @connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire") }
+    channel = ChatChannel.new @connection, ""
+    channel.stream_for Room.new(1)
+  end
 end
diff --git a/test/stubs/room.rb b/test/stubs/room.rb
index 388b5ae33f..246d6a98af 100644
--- a/test/stubs/room.rb
+++ b/test/stubs/room.rb
@@ -9,4 +9,8 @@ class Room
   def to_global_id
     "Room##{id}-#{name}"
   end
+
+  def to_gid_param
+    to_global_id.to_param
+  end
 end