mirror of https://github.com/rails/rails
Add connection identifier and an internal redis channel
This commit is contained in:
Pratik Naik
parent
354018bf9b
commit
eec92d0229
|
|
@ -4,6 +4,7 @@ PATH
|
|||
action_cable (0.0.2)
|
||||
activesupport (>= 4.2.0)
|
||||
celluloid (~> 0.16.0)
|
||||
em-hiredis (~> 0.3.0)
|
||||
faye-websocket (~> 0.9.2)
|
||||
|
||||
GEM
|
||||
|
|
@ -17,10 +18,14 @@ GEM
|
|||
tzinfo (~> 1.1)
|
||||
celluloid (0.16.0)
|
||||
timers (~> 4.0.0)
|
||||
em-hiredis (0.3.0)
|
||||
eventmachine (~> 1.0)
|
||||
hiredis (~> 0.5.0)
|
||||
eventmachine (1.0.7)
|
||||
faye-websocket (0.9.2)
|
||||
eventmachine (>= 0.12.0)
|
||||
websocket-driver (>= 0.5.1)
|
||||
hiredis (0.5.2)
|
||||
hitimes (1.2.2)
|
||||
i18n (0.7.0)
|
||||
json (1.8.2)
|
||||
|
|
@ -34,9 +39,9 @@ GEM
|
|||
hitimes
|
||||
tzinfo (1.2.2)
|
||||
thread_safe (~> 0.1)
|
||||
websocket-driver (0.5.1)
|
||||
websocket-driver (0.5.3)
|
||||
websocket-extensions (>= 0.1.0)
|
||||
websocket-extensions (0.1.1)
|
||||
websocket-extensions (0.1.2)
|
||||
|
||||
PLATFORMS
|
||||
ruby
|
||||
|
|
|
|||
|
|
@ -23,5 +23,5 @@ module ActionCable
|
|||
autoload :Worker, 'action_cable/worker'
|
||||
autoload :Server, 'action_cable/server'
|
||||
autoload :Connection, 'action_cable/connection'
|
||||
autoload :Connections, 'action_cable/connections'
|
||||
autoload :ConnectionProxy, 'action_cable/connection_proxy'
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1,133 +1,6 @@
|
|||
module ActionCable
|
||||
class Connection
|
||||
PING_INTERVAL = 3
|
||||
|
||||
attr_reader :env, :server
|
||||
delegate :worker_pool, :pubsub, :logger, to: :server
|
||||
|
||||
def initialize(server, env)
|
||||
@server = server
|
||||
@env = env
|
||||
@accept_messages = false
|
||||
@pending_messages = []
|
||||
end
|
||||
|
||||
def process
|
||||
if Faye::WebSocket.websocket?(@env)
|
||||
@subscriptions = {}
|
||||
|
||||
@websocket = Faye::WebSocket.new(@env)
|
||||
|
||||
@websocket.on(:open) do |event|
|
||||
broadcast_ping_timestamp
|
||||
@ping_timer = EventMachine.add_periodic_timer(PING_INTERVAL) { broadcast_ping_timestamp }
|
||||
worker_pool.async.invoke(self, :initialize_client)
|
||||
end
|
||||
|
||||
@websocket.on(:message) do |event|
|
||||
message = event.data
|
||||
|
||||
if message.is_a?(String)
|
||||
if @accept_messages
|
||||
worker_pool.async.invoke(self, :received_data, message)
|
||||
else
|
||||
@pending_messages << message
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@websocket.on(:close) do |event|
|
||||
worker_pool.async.invoke(self, :cleanup_subscriptions)
|
||||
worker_pool.async.invoke(self, :disconnect) if respond_to?(:disconnect)
|
||||
|
||||
EventMachine.cancel_timer(@ping_timer) if @ping_timer
|
||||
end
|
||||
|
||||
@websocket.rack_response
|
||||
else
|
||||
invalid_request
|
||||
end
|
||||
end
|
||||
|
||||
def received_data(data)
|
||||
return unless websocket_alive?
|
||||
|
||||
data = ActiveSupport::JSON.decode data
|
||||
|
||||
case data['action']
|
||||
when 'subscribe'
|
||||
subscribe_channel(data)
|
||||
when 'unsubscribe'
|
||||
unsubscribe_channel(data)
|
||||
when 'message'
|
||||
process_message(data)
|
||||
end
|
||||
end
|
||||
|
||||
def cleanup_subscriptions
|
||||
@subscriptions.each do |id, channel|
|
||||
channel.unsubscribe
|
||||
end
|
||||
end
|
||||
|
||||
def broadcast(data)
|
||||
logger.info "Sending data: #{data}"
|
||||
@websocket.send data
|
||||
end
|
||||
|
||||
def handle_exception
|
||||
logger.error "[ActionCable] Closing connection"
|
||||
|
||||
@websocket.close
|
||||
end
|
||||
|
||||
private
|
||||
def initialize_client
|
||||
connect if respond_to?(:connect)
|
||||
@accept_messages = true
|
||||
|
||||
worker_pool.async.invoke(self, :received_data, @pending_messages.shift) until @pending_messages.empty?
|
||||
end
|
||||
|
||||
def broadcast_ping_timestamp
|
||||
broadcast({ identifier: '_ping', message: Time.now.to_i }.to_json)
|
||||
end
|
||||
|
||||
def subscribe_channel(data)
|
||||
id_key = data['identifier']
|
||||
id_options = ActiveSupport::JSON.decode(id_key).with_indifferent_access
|
||||
|
||||
subscription_klass = server.registered_channels.detect { |channel_klass| channel_klass.find_name == id_options[:channel] }
|
||||
|
||||
if subscription_klass
|
||||
logger.info "Subscribing to channel: #{id_key}"
|
||||
@subscriptions[id_key] = subscription_klass.new(self, id_key, id_options)
|
||||
else
|
||||
logger.error "Unable to subscribe to channel: #{id_key}"
|
||||
end
|
||||
end
|
||||
|
||||
def process_message(message)
|
||||
if @subscriptions[message['identifier']]
|
||||
@subscriptions[message['identifier']].receive_data(ActiveSupport::JSON.decode message['data'])
|
||||
else
|
||||
logger.error "Unable to process message: #{message}"
|
||||
end
|
||||
end
|
||||
|
||||
def unsubscribe_channel(data)
|
||||
logger.info "Unsubscribing from channel: #{data['identifier']}"
|
||||
@subscriptions[data['identifier']].unsubscribe
|
||||
@subscriptions.delete(data['identifier'])
|
||||
end
|
||||
|
||||
def invalid_request
|
||||
[404, {'Content-Type' => 'text/plain'}, ['Page not found']]
|
||||
end
|
||||
|
||||
def websocket_alive?
|
||||
@websocket && @websocket.ready_state == Faye::WebSocket::API::OPEN
|
||||
end
|
||||
|
||||
module Connection
|
||||
autoload :Base, 'action_cable/connection/base'
|
||||
autoload :Registry, 'action_cable/connection/Registry'
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -0,0 +1,139 @@
|
|||
module ActionCable
|
||||
module Connection
|
||||
class Base
|
||||
include Registry
|
||||
|
||||
PING_INTERVAL = 3
|
||||
|
||||
attr_reader :env, :server
|
||||
delegate :worker_pool, :pubsub, :logger, to: :server
|
||||
|
||||
def initialize(server, env)
|
||||
@server = server
|
||||
@env = env
|
||||
@accept_messages = false
|
||||
@pending_messages = []
|
||||
end
|
||||
|
||||
def process
|
||||
if Faye::WebSocket.websocket?(@env)
|
||||
@subscriptions = {}
|
||||
|
||||
@websocket = Faye::WebSocket.new(@env)
|
||||
|
||||
@websocket.on(:open) do |event|
|
||||
broadcast_ping_timestamp
|
||||
@ping_timer = EventMachine.add_periodic_timer(PING_INTERVAL) { broadcast_ping_timestamp }
|
||||
worker_pool.async.invoke(self, :initialize_client)
|
||||
end
|
||||
|
||||
@websocket.on(:message) do |event|
|
||||
message = event.data
|
||||
|
||||
if message.is_a?(String)
|
||||
if @accept_messages
|
||||
worker_pool.async.invoke(self, :received_data, message)
|
||||
else
|
||||
@pending_messages << message
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@websocket.on(:close) do |event|
|
||||
worker_pool.async.invoke(self, :cleanup_subscriptions)
|
||||
worker_pool.async.invoke(self, :cleanup_subscriptions)
|
||||
worker_pool.async.invoke(self, :disconnect) if respond_to?(:disconnect)
|
||||
|
||||
EventMachine.cancel_timer(@ping_timer) if @ping_timer
|
||||
end
|
||||
|
||||
@websocket.rack_response
|
||||
else
|
||||
invalid_request
|
||||
end
|
||||
end
|
||||
|
||||
def received_data(data)
|
||||
return unless websocket_alive?
|
||||
|
||||
data = ActiveSupport::JSON.decode data
|
||||
|
||||
case data['action']
|
||||
when 'subscribe'
|
||||
subscribe_channel(data)
|
||||
when 'unsubscribe'
|
||||
unsubscribe_channel(data)
|
||||
when 'message'
|
||||
process_message(data)
|
||||
end
|
||||
end
|
||||
|
||||
def cleanup_subscriptions
|
||||
@subscriptions.each do |id, channel|
|
||||
channel.unsubscribe
|
||||
end
|
||||
end
|
||||
|
||||
def broadcast(data)
|
||||
logger.info "Sending data: #{data}"
|
||||
@websocket.send data
|
||||
end
|
||||
|
||||
def handle_exception
|
||||
logger.error "[ActionCable] Closing connection"
|
||||
|
||||
@websocket.close
|
||||
end
|
||||
|
||||
private
|
||||
def initialize_client
|
||||
connect if respond_to?(:connect)
|
||||
register_connection
|
||||
|
||||
@accept_messages = true
|
||||
worker_pool.async.invoke(self, :received_data, @pending_messages.shift) until @pending_messages.empty?
|
||||
end
|
||||
|
||||
def broadcast_ping_timestamp
|
||||
broadcast({ identifier: '_ping', message: Time.now.to_i }.to_json)
|
||||
end
|
||||
|
||||
def subscribe_channel(data)
|
||||
id_key = data['identifier']
|
||||
id_options = ActiveSupport::JSON.decode(id_key).with_indifferent_access
|
||||
|
||||
subscription_klass = server.registered_channels.detect { |channel_klass| channel_klass.find_name == id_options[:channel] }
|
||||
|
||||
if subscription_klass
|
||||
logger.info "Subscribing to channel: #{id_key}"
|
||||
@subscriptions[id_key] = subscription_klass.new(self, id_key, id_options)
|
||||
else
|
||||
logger.error "Unable to subscribe to channel: #{id_key}"
|
||||
end
|
||||
end
|
||||
|
||||
def process_message(message)
|
||||
if @subscriptions[message['identifier']]
|
||||
@subscriptions[message['identifier']].receive_data(ActiveSupport::JSON.decode message['data'])
|
||||
else
|
||||
logger.error "Unable to process message: #{message}"
|
||||
end
|
||||
end
|
||||
|
||||
def unsubscribe_channel(data)
|
||||
logger.info "Unsubscribing from channel: #{data['identifier']}"
|
||||
@subscriptions[data['identifier']].unsubscribe
|
||||
@subscriptions.delete(data['identifier'])
|
||||
end
|
||||
|
||||
def invalid_request
|
||||
[404, {'Content-Type' => 'text/plain'}, ['Page not found']]
|
||||
end
|
||||
|
||||
def websocket_alive?
|
||||
@websocket && @websocket.ready_state == Faye::WebSocket::API::OPEN
|
||||
end
|
||||
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,65 @@
|
|||
module ActionCable
|
||||
module Connection
|
||||
module Registry
|
||||
extend ActiveSupport::Concern
|
||||
|
||||
included do
|
||||
class_attribute :identifiers
|
||||
self.identifiers = Set.new
|
||||
end
|
||||
|
||||
module ClassMethods
|
||||
def identified_by(*identifiers)
|
||||
self.identifiers += identifiers
|
||||
end
|
||||
end
|
||||
|
||||
def register_connection
|
||||
if connection_identifier.present?
|
||||
callback = -> (message) { process_registry_message(message) }
|
||||
@_internal_redis_subscriptions ||= []
|
||||
@_internal_redis_subscriptions << [ internal_redis_channel, callback ]
|
||||
|
||||
pubsub.subscribe(internal_redis_channel, &callback)
|
||||
logger.info "[ActionCable] Registered connection (#{connection_identifier})"
|
||||
puts "[ActionCable] Registered connection: #{connection_identifier}(#{internal_redis_channel})"
|
||||
end
|
||||
end
|
||||
|
||||
def internal_redis_channel
|
||||
"action_cable/#{connection_identifier}"
|
||||
end
|
||||
|
||||
def connection_identifier
|
||||
@connection_identifier ||= connection_gid identifiers.map { |id| instance_variable_get("@#{id}")}
|
||||
end
|
||||
|
||||
def connection_gid(ids)
|
||||
ids.map {|o| o.to_global_id.to_s }.sort.join(":")
|
||||
end
|
||||
|
||||
def cleanup_internal_redis_subscriptions
|
||||
if @_internal_redis_subscriptions.present?
|
||||
@_internal_redis_subscriptions.each { |channel, callback| pubsub.unsubscribe_proc(channel, callback) }
|
||||
end
|
||||
end
|
||||
|
||||
private
|
||||
def process_registry_message(message)
|
||||
message = ActiveSupport::JSON.decode(message)
|
||||
|
||||
case message['type']
|
||||
when 'disconnect'
|
||||
logger.info "[ActionCable] Removing connection (#{connection_identifier})"
|
||||
@websocket.close
|
||||
end
|
||||
rescue Exception => e
|
||||
logger.error "[ActionCable] There was an exception - #{e.class}(#{e.message})"
|
||||
logger.error e.backtrace.join("\n")
|
||||
|
||||
handle_exception
|
||||
end
|
||||
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
module ActionCable
|
||||
module Connections
|
||||
module ConnectionProxy
|
||||
class << self
|
||||
def active
|
||||
end
|
||||
|
|
@ -10,8 +10,5 @@ module ActionCable
|
|||
|
||||
def disconnect
|
||||
end
|
||||
|
||||
def reconnect
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Reference in New Issue