From 5954fd1e0aa907e07ffff932aedc51109d4ce56d Mon Sep 17 00:00:00 2001 From: Lachlan Sylvester Date: Wed, 22 Jul 2015 11:03:47 +1000 Subject: [PATCH] add broadcast_to and stream_for methods as per #26 --- lib/action_cable/channel.rb | 2 ++ lib/action_cable/channel/base.rb | 2 ++ lib/action_cable/channel/broadcasting.rb | 27 +++++++++++++++++ lib/action_cable/channel/naming.rb | 22 ++++++++++++++ lib/action_cable/channel/streams.rb | 37 +++++++++++++++++++----- lib/action_cable/server/base.rb | 4 +-- lib/action_cable/server/broadcasting.rb | 2 +- test/channel/broadcasting_test.rb | 29 +++++++++++++++++++ test/channel/naming_test.rb | 10 +++++++ test/channel/stream_test.rb | 14 +++++++-- test/stubs/room.rb | 4 +++ 11 files changed, 139 insertions(+), 14 deletions(-) create mode 100644 lib/action_cable/channel/broadcasting.rb create mode 100644 lib/action_cable/channel/naming.rb create mode 100644 test/channel/broadcasting_test.rb create mode 100644 test/channel/naming_test.rb diff --git a/lib/action_cable/channel.rb b/lib/action_cable/channel.rb index 9e4d3d3f931b4..3b973ba0a7234 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 87ae3a12118cf..c83c3b74fd563 100644 --- a/lib/action_cable/channel/base.rb +++ b/lib/action_cable/channel/base.rb @@ -68,6 +68,8 @@ class Base 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 0000000000000..ee4117bc0a28e --- /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 model 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 0000000000000..4c9d53b15acba --- /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 Channel 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' channel_name + 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 6a3dc76c1df17..f711b065ca221 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 Channel # def follow(data) # stream_from "comments_for_#{data['recording_id']}" # end - # + # # def unfollow # stop_all_streams # end @@ -23,23 +23,37 @@ module Channel # # 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 @@ def stream_from(broadcasting, callback = nil) logger.info "#{self.class.name} is streaming from #{broadcasting}" end + # Start streaming the pubsub queue for the model in this channel. Optionally, you can pass a + # callback 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 fe7966e0903b7..b09fbf6da4e03 100644 --- a/lib/action_cable/server/base.rb +++ b/lib/action_cable/server/base.rb @@ -9,7 +9,7 @@ class Base 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 @@ def redis 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 4f72ffd96fca4..037b98951ebc3 100644 --- a/lib/action_cable/server/broadcasting.rb +++ b/lib/action_cable/server/broadcasting.rb @@ -32,7 +32,7 @@ def broadcaster_for(broadcasting) # 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 0000000000000..1de04243e574d --- /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 0000000000000..89ef6ad8b06b4 --- /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 1a8c259d11285..b0a6f49072867 100644 --- a/test/channel/stream_test.rb +++ b/test/channel/stream_test.rb @@ -5,8 +5,10 @@ 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 @@ def subscribed 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 388b5ae33fde5..246d6a98af535 100644 --- a/test/stubs/room.rb +++ b/test/stubs/room.rb @@ -9,4 +9,8 @@ def initialize(id, name='Campfire') def to_global_id "Room##{id}-#{name}" end + + def to_gid_param + to_global_id.to_param + end end