diff options
Diffstat (limited to 'actioncable/lib/action_cable')
7 files changed, 47 insertions, 32 deletions
diff --git a/actioncable/lib/action_cable/channel/periodic_timers.rb b/actioncable/lib/action_cable/channel/periodic_timers.rb index 25fe8e5e54..7f0fb37afc 100644 --- a/actioncable/lib/action_cable/channel/periodic_timers.rb +++ b/actioncable/lib/action_cable/channel/periodic_timers.rb @@ -28,7 +28,7 @@ module ActionCable def start_periodic_timers self.class.periodic_timers.each do |callback, options| active_periodic_timers << EventMachine::PeriodicTimer.new(options[:every]) do - connection.worker_pool.async.run_periodic_timer(self, callback) + connection.worker_pool.async_run_periodic_timer(self, callback) end end end diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb index f7b18a85ae..a8cfdf90f3 100644 --- a/actioncable/lib/action_cable/connection/base.rb +++ b/actioncable/lib/action_cable/connection/base.rb @@ -48,11 +48,9 @@ module ActionCable include InternalChannel include Authorization - attr_reader :server, :env, :subscriptions + attr_reader :server, :env, :subscriptions, :logger delegate :worker_pool, :pubsub, to: :server - attr_reader :logger - def initialize(server, env) @server, @env = server, env @@ -105,7 +103,7 @@ module ActionCable # Invoke a method on the connection asynchronously through the pool of thread workers. def send_async(method, *arguments) - worker_pool.async.invoke(self, method, *arguments) + worker_pool.async_invoke(self, method, *arguments) end # Return a basic hash of statistics for the connection keyed with `identifier`, `started_at`, and `subscriptions`. @@ -123,7 +121,6 @@ module ActionCable transmit ActiveSupport::JSON.encode(identifier: ActionCable::INTERNAL[:identifiers][:ping], message: Time.now.to_i) end - protected # The request that initiated the WebSocket connection is available here. This gives access to the environment, cookies, etc. def request @@ -138,8 +135,6 @@ module ActionCable request.cookie_jar end - - protected attr_reader :websocket attr_reader :message_buffer @@ -170,7 +165,6 @@ module ActionCable disconnect if respond_to?(:disconnect) end - def allow_request_origin? return true if server.config.disable_request_forgery_protection @@ -193,7 +187,6 @@ module ActionCable [ 404, { 'Content-Type' => 'text/plain' }, [ 'Page not found' ] ] end - # Tags are declared in the server but computed in the connection. This allows us per-connection tailored tags. def new_tagged_logger TaggedLoggerProxy.new server.logger, diff --git a/actioncable/lib/action_cable/connection/identification.rb b/actioncable/lib/action_cable/connection/identification.rb index 2d75ff8d6d..885ff3f102 100644 --- a/actioncable/lib/action_cable/connection/identification.rb +++ b/actioncable/lib/action_cable/connection/identification.rb @@ -11,7 +11,7 @@ module ActionCable end class_methods do - # Mark a key as being a connection identifier index that can then used to find the specific connection again later. + # Mark a key as being a connection identifier index that can then be used to find the specific connection again later. # Common identifiers are current_user and current_account, but could be anything really. # # Note that anything marked as an identifier will automatically create a delegate by the same name on any diff --git a/actioncable/lib/action_cable/connection/subscriptions.rb b/actioncable/lib/action_cable/connection/subscriptions.rb index 65d6634bb0..d7f95e6a62 100644 --- a/actioncable/lib/action_cable/connection/subscriptions.rb +++ b/actioncable/lib/action_cable/connection/subscriptions.rb @@ -49,7 +49,6 @@ module ActionCable find(data).perform_action ActiveSupport::JSON.decode(data['data']) end - def identifiers subscriptions.keys end diff --git a/actioncable/lib/action_cable/process/logging.rb b/actioncable/lib/action_cable/process/logging.rb index 72b1a080d1..dce637b3ca 100644 --- a/actioncable/lib/action_cable/process/logging.rb +++ b/actioncable/lib/action_cable/process/logging.rb @@ -1,10 +1,7 @@ require 'action_cable/server' require 'eventmachine' -require 'celluloid' EM.error_handler do |e| puts "Error raised inside the event loop: #{e.message}" puts e.backtrace.join("\n") end - -Celluloid.logger = ActionCable.server.logger diff --git a/actioncable/lib/action_cable/server/base.rb b/actioncable/lib/action_cable/server/base.rb index 740e4b301e..3785bbd154 100644 --- a/actioncable/lib/action_cable/server/base.rb +++ b/actioncable/lib/action_cable/server/base.rb @@ -1,6 +1,3 @@ -# FIXME: Cargo culted fix from https://github.com/celluloid/celluloid-pool/issues/10 -require 'celluloid/current' - require 'em-hiredis' module ActionCable @@ -39,7 +36,7 @@ module ActionCable # The thread worker pool for handling all the connection work on this server. Default size is set by config.worker_pool_size. def worker_pool - @worker_pool ||= ActionCable::Server::Worker.pool(size: config.worker_pool_size) + @worker_pool ||= ActionCable::Server::Worker.new(max_size: config.worker_pool_size) end # Requires and returns a hash of all the channel class constants keyed by name. diff --git a/actioncable/lib/action_cable/server/worker.rb b/actioncable/lib/action_cable/server/worker.rb index e063b2a2e1..3b6c6d44a1 100644 --- a/actioncable/lib/action_cable/server/worker.rb +++ b/actioncable/lib/action_cable/server/worker.rb @@ -1,39 +1,68 @@ -require 'celluloid' require 'active_support/callbacks' +require 'active_support/core_ext/module/attribute_accessors_per_thread' +require 'concurrent' module ActionCable module Server # Worker used by Server.send_async to do connection work in threads. Only for internal use. class Worker include ActiveSupport::Callbacks - include Celluloid - attr_reader :connection + thread_mattr_accessor :connection define_callbacks :work include ActiveRecordConnectionManagement + def initialize(max_size: 5) + @pool = Concurrent::ThreadPoolExecutor.new( + min_threads: 1, + max_threads: max_size, + max_queue: 0, + ) + end + + def async_invoke(receiver, method, *args) + @pool.post do + invoke(receiver, method, *args) + end + end + def invoke(receiver, method, *args) - @connection = receiver + begin + self.connection = receiver - run_callbacks :work do - receiver.send method, *args + run_callbacks :work do + receiver.send method, *args + end + rescue Exception => e + logger.error "There was an exception - #{e.class}(#{e.message})" + logger.error e.backtrace.join("\n") + + receiver.handle_exception if receiver.respond_to?(:handle_exception) + ensure + self.connection = nil end - rescue Exception => e - logger.error "There was an exception - #{e.class}(#{e.message})" - logger.error e.backtrace.join("\n") + end - receiver.handle_exception if receiver.respond_to?(:handle_exception) + def async_run_periodic_timer(channel, callback) + @pool.post do + run_periodic_timer(channel, callback) + end end def run_periodic_timer(channel, callback) - @connection = channel.connection + begin + self.connection = channel.connection - run_callbacks :work do - callback.respond_to?(:call) ? channel.instance_exec(&callback) : channel.send(callback) + run_callbacks :work do + callback.respond_to?(:call) ? channel.instance_exec(&callback) : channel.send(callback) + end + ensure + self.connection = nil end end private + def logger ActionCable.server.logger end |