1
0
Fork 0
mirror of https://github.com/rails/rails.git synced 2022-11-09 12:12:34 -05:00

add broadcast_to and stream_for methods as per #26

This commit is contained in:
Lachlan Sylvester 2015-07-22 11:03:47 +10:00
parent 6ee8bb9310
commit 5954fd1e0a
11 changed files with 139 additions and 14 deletions

View file

@ -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

View file

@ -68,6 +68,8 @@ module ActionCable
include Callbacks
include PeriodicTimers
include Streams
include Naming
include Broadcasting
on_subscribe :subscribed
on_unsubscribe :unsubscribed

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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