aboutsummaryrefslogtreecommitdiffstats
path: root/actioncable/lib/action_cable/server/worker.rb
diff options
context:
space:
mode:
Diffstat (limited to 'actioncable/lib/action_cable/server/worker.rb')
-rw-r--r--actioncable/lib/action_cable/server/worker.rb55
1 files changed, 42 insertions, 13 deletions
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