From 354018bf9b5f5bf0fbbc6e6efddc719e7523b39d Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Sat, 4 Apr 2015 00:26:14 -0500 Subject: Separate connection and server classes --- lib/action_cable.rb | 4 + lib/action_cable/channel/redis.rb | 7 +- lib/action_cable/connection.rb | 133 ++++++++++++++++++++++++++++++++ lib/action_cable/connections.rb | 17 +++++ lib/action_cable/server.rb | 154 +++----------------------------------- 5 files changed, 165 insertions(+), 150 deletions(-) create mode 100644 lib/action_cable/connection.rb create mode 100644 lib/action_cable/connections.rb (limited to 'lib') diff --git a/lib/action_cable.rb b/lib/action_cable.rb index 0681b8bdde..62046c7717 100644 --- a/lib/action_cable.rb +++ b/lib/action_cable.rb @@ -7,10 +7,12 @@ require 'active_support' require 'active_support/json' require 'active_support/concern' require 'active_support/core_ext/hash/indifferent_access' +require 'active_support/core_ext/module/delegation' require 'active_support/callbacks' require 'faye/websocket' require 'celluloid' +require 'em-hiredis' require 'action_cable/engine' if defined?(Rails) @@ -20,4 +22,6 @@ module ActionCable autoload :Channel, 'action_cable/channel' autoload :Worker, 'action_cable/worker' autoload :Server, 'action_cable/server' + autoload :Connection, 'action_cable/connection' + autoload :Connections, 'action_cable/connections' end diff --git a/lib/action_cable/channel/redis.rb b/lib/action_cable/channel/redis.rb index bdbd3c95b1..2691a3b145 100644 --- a/lib/action_cable/channel/redis.rb +++ b/lib/action_cable/channel/redis.rb @@ -6,11 +6,10 @@ module ActionCable included do on_unsubscribe :unsubscribe_from_redis_channels + delegate :pubsub, to: :connection end def subscribe_to(redis_channel, callback = nil) - raise "`ActionCable::Server.pubsub` class method is not defined" unless connection.class.respond_to?(:pubsub) - callback ||= -> (message) { broadcast ActiveSupport::JSON.decode(message) } @_redis_channels ||= [] @_redis_channels << [ redis_channel, callback ] @@ -24,10 +23,6 @@ module ActionCable @_redis_channels.each { |channel, callback| pubsub.unsubscribe_proc(channel, callback) } end end - - def pubsub - connection.class.pubsub - end end end diff --git a/lib/action_cable/connection.rb b/lib/action_cable/connection.rb new file mode 100644 index 0000000000..00fb8ca817 --- /dev/null +++ b/lib/action_cable/connection.rb @@ -0,0 +1,133 @@ +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 + + end +end diff --git a/lib/action_cable/connections.rb b/lib/action_cable/connections.rb new file mode 100644 index 0000000000..e68cc6e7a4 --- /dev/null +++ b/lib/action_cable/connections.rb @@ -0,0 +1,17 @@ +module ActionCable + module Connections + class << self + def active + end + + def where(identification) + end + end + + def disconnect + end + + def reconnect + end + end +end diff --git a/lib/action_cable/server.rb b/lib/action_cable/server.rb index ebf98171c1..6e9265dc06 100644 --- a/lib/action_cable/server.rb +++ b/lib/action_cable/server.rb @@ -1,157 +1,23 @@ module ActionCable class Server - class_attribute :registered_channels - self.registered_channels = Set.new - - class_attribute :worker_pool_size - self.worker_pool_size = 100 - cattr_accessor(:logger, instance_reader: true) { Rails.logger } - PING_INTERVAL = 3 - - class << self - def register_channels(*channel_classes) - self.registered_channels += channel_classes - end - - def call(env) - new(env).process - end - - def worker_pool - @worker_pool ||= ActionCable::Worker.pool(size: worker_pool_size) - end - end - - attr_reader :env - - def initialize(env) - @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 + attr_accessor :registered_channels, :worker_pool - @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 + def initialize(redis_config:, channels:, worker_pool_size: 100, connection: Connection) + @redis_config = redis_config + @registered_channels = Set.new(channels) + @worker_pool = ActionCable::Worker.pool(size: worker_pool_size) + @connection_class = connection 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 + def call(env) + @connection_class.new(self, env).process end - def cleanup_subscriptions - @subscriptions.each do |id, channel| - channel.unsubscribe - end + def pubsub + @pubsub ||= EM::Hiredis.connect(@redis_config['url']).pubsub end - def broadcast(data) - logger.info "Sending data: #{data}" - @websocket.send data - end - - def worker_pool - self.class.worker_pool - 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 = 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 -- cgit v1.2.3 From eec92d0229a7bceb62d49d58c70b5629fe140d7f Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Mon, 6 Apr 2015 12:21:22 -0500 Subject: Add connection identifier and an internal redis channel --- lib/action_cable.rb | 2 +- lib/action_cable/connection.rb | 133 +----------------------------- lib/action_cable/connection/base.rb | 139 ++++++++++++++++++++++++++++++++ lib/action_cable/connection/registry.rb | 65 +++++++++++++++ lib/action_cable/connection_proxy.rb | 14 ++++ lib/action_cable/connections.rb | 17 ---- 6 files changed, 222 insertions(+), 148 deletions(-) create mode 100644 lib/action_cable/connection/base.rb create mode 100644 lib/action_cable/connection/registry.rb create mode 100644 lib/action_cable/connection_proxy.rb delete mode 100644 lib/action_cable/connections.rb (limited to 'lib') diff --git a/lib/action_cable.rb b/lib/action_cable.rb index 62046c7717..fd42dc6cdd 100644 --- a/lib/action_cable.rb +++ b/lib/action_cable.rb @@ -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 diff --git a/lib/action_cable/connection.rb b/lib/action_cable/connection.rb index 00fb8ca817..102903c6ef 100644 --- a/lib/action_cable/connection.rb +++ b/lib/action_cable/connection.rb @@ -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 diff --git a/lib/action_cable/connection/base.rb b/lib/action_cable/connection/base.rb new file mode 100644 index 0000000000..c6c9899094 --- /dev/null +++ b/lib/action_cable/connection/base.rb @@ -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 diff --git a/lib/action_cable/connection/registry.rb b/lib/action_cable/connection/registry.rb new file mode 100644 index 0000000000..121f87f9f0 --- /dev/null +++ b/lib/action_cable/connection/registry.rb @@ -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 diff --git a/lib/action_cable/connection_proxy.rb b/lib/action_cable/connection_proxy.rb new file mode 100644 index 0000000000..980e037ff3 --- /dev/null +++ b/lib/action_cable/connection_proxy.rb @@ -0,0 +1,14 @@ +module ActionCable + module ConnectionProxy + class << self + def active + end + + def where(identification) + end + end + + def disconnect + end + end +end diff --git a/lib/action_cable/connections.rb b/lib/action_cable/connections.rb deleted file mode 100644 index e68cc6e7a4..0000000000 --- a/lib/action_cable/connections.rb +++ /dev/null @@ -1,17 +0,0 @@ -module ActionCable - module Connections - class << self - def active - end - - def where(identification) - end - end - - def disconnect - end - - def reconnect - end - end -end -- cgit v1.2.3 From 1c9d82dbf0e743534c5fa9be936eaa46c5b07523 Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Mon, 6 Apr 2015 13:17:02 -0500 Subject: Add remote connection to talk over internal redis channel --- lib/action_cable.rb | 2 +- lib/action_cable/connection.rb | 3 ++- lib/action_cable/connection/identifier.rb | 19 ++++++++++++++++ lib/action_cable/connection_proxy.rb | 14 ------------ lib/action_cable/remote_connection.rb | 38 +++++++++++++++++++++++++++++++ lib/action_cable/server.rb | 12 ++++++++-- 6 files changed, 70 insertions(+), 18 deletions(-) create mode 100644 lib/action_cable/connection/identifier.rb delete mode 100644 lib/action_cable/connection_proxy.rb create mode 100644 lib/action_cable/remote_connection.rb (limited to 'lib') diff --git a/lib/action_cable.rb b/lib/action_cable.rb index fd42dc6cdd..159ee2bcc0 100644 --- a/lib/action_cable.rb +++ b/lib/action_cable.rb @@ -23,5 +23,5 @@ module ActionCable autoload :Worker, 'action_cable/worker' autoload :Server, 'action_cable/server' autoload :Connection, 'action_cable/connection' - autoload :ConnectionProxy, 'action_cable/connection_proxy' + autoload :RemoteConnection, 'action_cable/remote_connection' end diff --git a/lib/action_cable/connection.rb b/lib/action_cable/connection.rb index 102903c6ef..91fc73713c 100644 --- a/lib/action_cable/connection.rb +++ b/lib/action_cable/connection.rb @@ -1,6 +1,7 @@ module ActionCable module Connection autoload :Base, 'action_cable/connection/base' - autoload :Registry, 'action_cable/connection/Registry' + autoload :Registry, 'action_cable/connection/registry' + autoload :Identifier, 'action_cable/connection/identifier' end end diff --git a/lib/action_cable/connection/identifier.rb b/lib/action_cable/connection/identifier.rb new file mode 100644 index 0000000000..9bfd773ab1 --- /dev/null +++ b/lib/action_cable/connection/identifier.rb @@ -0,0 +1,19 @@ +module ActionCable + module Connection + module Identifier + + 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 + + end + end +end diff --git a/lib/action_cable/connection_proxy.rb b/lib/action_cable/connection_proxy.rb deleted file mode 100644 index 980e037ff3..0000000000 --- a/lib/action_cable/connection_proxy.rb +++ /dev/null @@ -1,14 +0,0 @@ -module ActionCable - module ConnectionProxy - class << self - def active - end - - def where(identification) - end - end - - def disconnect - end - end -end diff --git a/lib/action_cable/remote_connection.rb b/lib/action_cable/remote_connection.rb new file mode 100644 index 0000000000..e2cb2d932c --- /dev/null +++ b/lib/action_cable/remote_connection.rb @@ -0,0 +1,38 @@ +module ActionCable + class RemoteConnection + class InvalidIdentifiersError < StandardError; end + + include Connection::Identifier + + delegate :redis, to: :server + + def initialize(server, ids) + @server = server + set_identifier_instance_vars(ids) + end + + def disconnect + message = { type: 'disconnect' }.to_json + redis.publish(internal_redis_channel, message) + end + + def identifiers + @server.connection_identifiers + end + + def redis + @redis ||= Redis.new(@server.redis_config) + end + + private + def set_identifier_instance_vars(ids) + raise InvalidIdentifiersError unless valid_identifiers?(ids) + ids.each { |k,v| instance_variable_set("@#{k}", v) } + end + + def valid_identifiers?(ids) + keys = ids.keys + identifiers.all? { |id| keys.include?(id) } + end + end +end diff --git a/lib/action_cable/server.rb b/lib/action_cable/server.rb index 6e9265dc06..51e246c232 100644 --- a/lib/action_cable/server.rb +++ b/lib/action_cable/server.rb @@ -2,12 +2,12 @@ module ActionCable class Server cattr_accessor(:logger, instance_reader: true) { Rails.logger } - attr_accessor :registered_channels, :worker_pool + attr_accessor :registered_channels, :redis_config def initialize(redis_config:, channels:, worker_pool_size: 100, connection: Connection) @redis_config = redis_config @registered_channels = Set.new(channels) - @worker_pool = ActionCable::Worker.pool(size: worker_pool_size) + @worker_pool_size = worker_pool_size @connection_class = connection end @@ -15,9 +15,17 @@ module ActionCable @connection_class.new(self, env).process end + def worker_pool + @worker_pool ||= ActionCable::Worker.pool(size: @worker_pool_size) + end + def pubsub @pubsub ||= EM::Hiredis.connect(@redis_config['url']).pubsub end + def connection_identifiers + @connection_class.identifiers + end + end end -- cgit v1.2.3 From 9501dd2f6d2f0ece938e2a15a31429e9b8cf1001 Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Mon, 6 Apr 2015 13:17:14 -0500 Subject: Be sure to cleanup internal redis subscriptions on close --- lib/action_cable/connection/base.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/action_cable/connection/base.rb b/lib/action_cable/connection/base.rb index c6c9899094..c5b982acf8 100644 --- a/lib/action_cable/connection/base.rb +++ b/lib/action_cable/connection/base.rb @@ -41,7 +41,7 @@ module ActionCable @websocket.on(:close) do |event| worker_pool.async.invoke(self, :cleanup_subscriptions) - worker_pool.async.invoke(self, :cleanup_subscriptions) + worker_pool.async.invoke(self, :cleanup_internal_redis_subscriptions) worker_pool.async.invoke(self, :disconnect) if respond_to?(:disconnect) EventMachine.cancel_timer(@ping_timer) if @ping_timer -- cgit v1.2.3 From 0c143c03441cf2a66557ec7ba2f5d3d2f889fb5d Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Mon, 6 Apr 2015 13:17:41 -0500 Subject: Remove a puts message --- lib/action_cable/connection/registry.rb | 1 - 1 file changed, 1 deletion(-) (limited to 'lib') diff --git a/lib/action_cable/connection/registry.rb b/lib/action_cable/connection/registry.rb index 121f87f9f0..03a0bf4fe9 100644 --- a/lib/action_cable/connection/registry.rb +++ b/lib/action_cable/connection/registry.rb @@ -22,7 +22,6 @@ module ActionCable pubsub.subscribe(internal_redis_channel, &callback) logger.info "[ActionCable] Registered connection (#{connection_identifier})" - puts "[ActionCable] Registered connection: #{connection_identifier}(#{internal_redis_channel})" end end -- cgit v1.2.3 From 90566fce53a71660c746e23499f8aa134b457335 Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Mon, 6 Apr 2015 13:31:17 -0500 Subject: Remote connections API for the server --- lib/action_cable.rb | 1 + lib/action_cable/remote_connections.rb | 13 +++++++++++++ lib/action_cable/server.rb | 4 ++++ 3 files changed, 18 insertions(+) create mode 100644 lib/action_cable/remote_connections.rb (limited to 'lib') diff --git a/lib/action_cable.rb b/lib/action_cable.rb index 159ee2bcc0..3352453491 100644 --- a/lib/action_cable.rb +++ b/lib/action_cable.rb @@ -24,4 +24,5 @@ module ActionCable autoload :Server, 'action_cable/server' autoload :Connection, 'action_cable/connection' autoload :RemoteConnection, 'action_cable/remote_connection' + autoload :RemoteConnections, 'action_cable/remote_connections' end diff --git a/lib/action_cable/remote_connections.rb b/lib/action_cable/remote_connections.rb new file mode 100644 index 0000000000..f9d7c49a27 --- /dev/null +++ b/lib/action_cable/remote_connections.rb @@ -0,0 +1,13 @@ +module ActionCable + class RemoteConnections + attr_reader :server + + def initialize(server) + @server = server + end + + def where(identifier) + RemoteConnection.new(server, identifier) + end + end +end diff --git a/lib/action_cable/server.rb b/lib/action_cable/server.rb index 51e246c232..222c77fd51 100644 --- a/lib/action_cable/server.rb +++ b/lib/action_cable/server.rb @@ -23,6 +23,10 @@ module ActionCable @pubsub ||= EM::Hiredis.connect(@redis_config['url']).pubsub end + def remote_connections + @remote_connections ||= RemoteConnections.new(self) + end + def connection_identifiers @connection_class.identifiers end -- cgit v1.2.3 From 8c4c782c78b45d94e97ee9a26a05c44cae7cd428 Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Mon, 6 Apr 2015 15:38:22 -0500 Subject: Catch exceptions when subscribing to a channel and processing a message --- lib/action_cable/connection/base.rb | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) (limited to 'lib') diff --git a/lib/action_cable/connection/base.rb b/lib/action_cable/connection/base.rb index c5b982acf8..ea5d52e99c 100644 --- a/lib/action_cable/connection/base.rb +++ b/lib/action_cable/connection/base.rb @@ -105,23 +105,29 @@ module ActionCable 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}" + logger.info "[ActionCable] 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}" + logger.error "[ActionCable] Subscription class not found (#{data.inspect})" end + rescue Exception => e + logger.error "[ActionCable] Could not subscribe to channel (#{data.inspect})" + logger.error e.backtrace.join("\n") 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}" + logger.error "[ActionCable] Unable to process message because no subscription found (#{message.inspect})" end + rescue Exception => e + logger.error "[ActionCable] Could not process message (#{data.inspect})" + logger.error e.backtrace.join("\n") end def unsubscribe_channel(data) - logger.info "Unsubscribing from channel: #{data['identifier']}" + logger.info "[ActionCable] Unsubscribing from channel: #{data['identifier']}" @subscriptions[data['identifier']].unsubscribe @subscriptions.delete(data['identifier']) end -- cgit v1.2.3 From fb797ad1f1c3b0d96968c5feef783a2b8fe07eed Mon Sep 17 00:00:00 2001 From: Pratik Naik Date: Mon, 6 Apr 2015 15:43:23 -0500 Subject: Fix an error message --- lib/action_cable/connection/base.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/action_cable/connection/base.rb b/lib/action_cable/connection/base.rb index ea5d52e99c..4ad1e7d065 100644 --- a/lib/action_cable/connection/base.rb +++ b/lib/action_cable/connection/base.rb @@ -119,7 +119,7 @@ module ActionCable if @subscriptions[message['identifier']] @subscriptions[message['identifier']].receive_data(ActiveSupport::JSON.decode message['data']) else - logger.error "[ActionCable] Unable to process message because no subscription found (#{message.inspect})" + logger.error "[ActionCable] Unable to process message because no subscription was found (#{message.inspect})" end rescue Exception => e logger.error "[ActionCable] Could not process message (#{data.inspect})" -- cgit v1.2.3