diff options
Diffstat (limited to 'actioncable/lib/action_cable')
32 files changed, 260 insertions, 280 deletions
diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb index 845b747fc5..a866044f95 100644 --- a/actioncable/lib/action_cable/channel/base.rb +++ b/actioncable/lib/action_cable/channel/base.rb @@ -1,4 +1,4 @@ -require 'set' +require "set" module ActionCable module Channel @@ -144,13 +144,14 @@ module ActionCable # When a channel is streaming via pubsub, we want to delay the confirmation # transmission until pubsub subscription is confirmed. - @defer_subscription_confirmation = false + # + # The counter starts at 1 because it's awaiting a call to #subscribe_to_channel + @defer_subscription_confirmation_counter = Concurrent::AtomicFixnum.new(1) @reject_subscription = nil @subscription_confirmation_sent = nil delegate_connection_identifiers - subscribe_to_channel end # Extract the action name from the passed data and process it via the channel. The process will ensure @@ -169,6 +170,17 @@ module ActionCable end end + # This method is called after subscription has been added to the connection + # and confirms or rejects the subscription. + def subscribe_to_channel + run_callbacks :subscribe do + subscribed + end + + reject_subscription if subscription_rejected? + ensure_confirmation_sent + end + # Called by the cable connection when it's cut, so the channel has a chance to cleanup with callbacks. # This method is not intended to be called directly by the user. Instead, overwrite the #unsubscribed callback. def unsubscribe_from_channel # :nodoc: @@ -177,7 +189,6 @@ module ActionCable end end - protected # Called once a consumer has become a subscriber of the channel. Usually the place to setup any streams # you want this channel to be sending to the subscriber. @@ -202,12 +213,18 @@ module ActionCable end end + def ensure_confirmation_sent + return if subscription_rejected? + @defer_subscription_confirmation_counter.decrement + transmit_subscription_confirmation unless defer_subscription_confirmation? + end + def defer_subscription_confirmation! - @defer_subscription_confirmation = true + @defer_subscription_confirmation_counter.increment end def defer_subscription_confirmation? - @defer_subscription_confirmation + @defer_subscription_confirmation_counter.value > 0 end def subscription_confirmation_sent? @@ -231,24 +248,12 @@ module ActionCable end end - def subscribe_to_channel - run_callbacks :subscribe do - subscribed - end - - if subscription_rejected? - reject_subscription - else - transmit_subscription_confirmation unless defer_subscription_confirmation? - end - end - def extract_action(data) - (data['action'].presence || :receive).to_sym + (data["action"].presence || :receive).to_sym end def processable_action?(action) - self.class.action_methods.include?(action.to_s) + self.class.action_methods.include?(action.to_s) unless subscription_rejected? end def dispatch_action(action, data) @@ -263,7 +268,7 @@ module ActionCable def action_signature(action, data) "#{self.class.name}##{action}".tap do |signature| - if (arguments = data.except('action')).any? + if (arguments = data.except("action")).any? signature << "(#{arguments.inspect})" end end diff --git a/actioncable/lib/action_cable/channel/broadcasting.rb b/actioncable/lib/action_cable/channel/broadcasting.rb index afc23d7d1a..23ed4ec943 100644 --- a/actioncable/lib/action_cable/channel/broadcasting.rb +++ b/actioncable/lib/action_cable/channel/broadcasting.rb @@ -1,4 +1,4 @@ -require 'active_support/core_ext/object/to_param' +require "active_support/core_ext/object/to_param" module ActionCable module Channel @@ -16,7 +16,7 @@ module ActionCable def broadcasting_for(model) #:nodoc: case when model.is_a?(Array) - model.map { |m| broadcasting_for(m) }.join(':') + model.map { |m| broadcasting_for(m) }.join(":") when model.respond_to?(:to_gid_param) model.to_gid_param else diff --git a/actioncable/lib/action_cable/channel/callbacks.rb b/actioncable/lib/action_cable/channel/callbacks.rb index 295d750e86..c740132c94 100644 --- a/actioncable/lib/action_cable/channel/callbacks.rb +++ b/actioncable/lib/action_cable/channel/callbacks.rb @@ -1,4 +1,4 @@ -require 'active_support/callbacks' +require "active_support/callbacks" module ActionCable module Channel diff --git a/actioncable/lib/action_cable/channel/naming.rb b/actioncable/lib/action_cable/channel/naming.rb index 4c9d53b15a..b7e88bf73d 100644 --- a/actioncable/lib/action_cable/channel/naming.rb +++ b/actioncable/lib/action_cable/channel/naming.rb @@ -10,8 +10,9 @@ module ActionCable # # ChatChannel.channel_name # => 'chat' # Chats::AppearancesChannel.channel_name # => 'chats:appearances' + # FooChats::BarAppearancesChannel.channel_name # => 'foo_chats:bar_appearances' def channel_name - @channel_name ||= name.sub(/Channel$/, '').gsub('::',':').underscore + @channel_name ||= name.sub(/Channel$/, "").gsub("::",":").underscore end end diff --git a/actioncable/lib/action_cable/channel/periodic_timers.rb b/actioncable/lib/action_cable/channel/periodic_timers.rb index dab604440f..c9daa0bcd3 100644 --- a/actioncable/lib/action_cable/channel/periodic_timers.rb +++ b/actioncable/lib/action_cable/channel/periodic_timers.rb @@ -30,7 +30,7 @@ module ActionCable def periodically(callback_or_method_name = nil, every:, &block) callback = if block_given? - raise ArgumentError, 'Pass a block or provide a callback arg, not both' if callback_or_method_name + raise ArgumentError, "Pass a block or provide a callback arg, not both" if callback_or_method_name block else case callback_or_method_name @@ -64,9 +64,7 @@ module ActionCable def start_periodic_timer(callback, every:) connection.server.event_loop.timer every do - connection.worker_pool.async_invoke connection do - instance_exec(&callback) - end + connection.worker_pool.async_exec self, connection: connection, &callback end end diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb index 200c9d053c..dbba333353 100644 --- a/actioncable/lib/action_cable/channel/streams.rb +++ b/actioncable/lib/action_cable/channel/streams.rb @@ -19,14 +19,14 @@ module ActionCable # end # # Based on the above example, the subscribers of this channel will get whatever data is put into the, - # let's say, `comments_for_45` broadcasting as soon as it's put there. + # let's say, <tt>comments_for_45</tt> broadcasting as soon as it's put there. # # An example broadcasting for this channel looks like so: # # ActionCable.server.broadcast "comments_for_45", author: 'DHH', content: 'Rails is just swell' # # 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 subscribe to a broadcasting like `comments:Z2lkOi8vVGVzdEFwcC9Qb3N0LzE` + # The following example would subscribe to a broadcasting like <tt>comments:Z2lkOi8vVGVzdEFwcC9Qb3N0LzE</tt>. # # class CommentsChannel < ApplicationCable::Channel # def subscribed @@ -69,8 +69,8 @@ module ActionCable # Start streaming from the named <tt>broadcasting</tt> pubsub queue. 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. - # Pass `coder: ActiveSupport::JSON` to decode messages as JSON before passing to the callback. - # Defaults to `coder: nil` which does no decoding, passes raw messages. + # Pass <tt>coder: ActiveSupport::JSON</tt> to decode messages as JSON before passing to the callback. + # Defaults to <tt>coder: nil</tt> which does no decoding, passes raw messages. def stream_from(broadcasting, callback = nil, coder: nil, &block) broadcasting = String(broadcasting) @@ -84,7 +84,7 @@ module ActionCable connection.server.event_loop.post do pubsub.subscribe(broadcasting, handler, lambda do - transmit_subscription_confirmation + ensure_confirmation_sent logger.info "#{self.class.name} is streaming from #{broadcasting}" end) end @@ -94,8 +94,8 @@ module ActionCable # <tt>callback</tt> that'll be used instead of the default of just transmitting the updates straight # to the subscriber. # - # Pass `coder: ActiveSupport::JSON` to decode messages as JSON before passing to the callback. - # Defaults to `coder: nil` which does no decoding, passes raw messages. + # Pass <tt>coder: ActiveSupport::JSON</tt> to decode messages as JSON before passing to the callback. + # Defaults to <tt>coder: nil</tt> which does no decoding, passes raw messages. def stream_for(model, callback = nil, coder: nil, &block) stream_from(broadcasting_for([ channel_name, model ]), callback || block, coder: coder) end @@ -138,7 +138,7 @@ module ActionCable end # May be overridden to change the default stream handling behavior - # which decodes JSON and transmits to client. + # which decodes JSON and transmits to the client. # # TODO: Tests demonstrating this. # diff --git a/actioncable/lib/action_cable/connection.rb b/actioncable/lib/action_cable/connection.rb index 5f813cf8e0..902efb07e2 100644 --- a/actioncable/lib/action_cable/connection.rb +++ b/actioncable/lib/action_cable/connection.rb @@ -8,8 +8,6 @@ module ActionCable autoload :ClientSocket autoload :Identification autoload :InternalChannel - autoload :FayeClientSocket - autoload :FayeEventLoop autoload :MessageBuffer autoload :Stream autoload :StreamEventLoop diff --git a/actioncable/lib/action_cable/connection/authorization.rb b/actioncable/lib/action_cable/connection/authorization.rb index 070a70e4e2..85df206445 100644 --- a/actioncable/lib/action_cable/connection/authorization.rb +++ b/actioncable/lib/action_cable/connection/authorization.rb @@ -10,4 +10,4 @@ module ActionCable end end end -end
\ No newline at end of file +end diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb index cc4e0f8c8b..06f4f5edd3 100644 --- a/actioncable/lib/action_cable/connection/base.rb +++ b/actioncable/lib/action_cable/connection/base.rb @@ -1,8 +1,8 @@ -require 'action_dispatch' +require "action_dispatch" module ActionCable module Connection - # For every WebSocket the Action Cable server accepts, a Connection object will be instantiated. This instance becomes the parent + # For every WebSocket connection the Action Cable server accepts, a Connection object will be instantiated. This instance becomes the parent # of all of the channel subscriptions that are created from there on. Incoming messages are then routed to these channel subscriptions # based on an identifier sent by the Action Cable consumer. The Connection itself does not deal with any specific application logic beyond # authentication and authorization. @@ -57,7 +57,7 @@ module ActionCable @worker_pool = server.worker_pool @logger = new_tagged_logger - @websocket = ActionCable::Connection::WebSocket.new(env, self, event_loop, server.config.client_socket_class) + @websocket = ActionCable::Connection::WebSocket.new(env, self, event_loop) @subscriptions = ActionCable::Connection::Subscriptions.new(self) @message_buffer = ActionCable::Connection::MessageBuffer.new(self) @@ -105,14 +105,14 @@ module ActionCable worker_pool.async_invoke(self, method, *arguments) end - # Return a basic hash of statistics for the connection keyed with `identifier`, `started_at`, and `subscriptions`. + # Return a basic hash of statistics for the connection keyed with <tt>identifier</tt>, <tt>started_at</tt>, <tt>subscriptions</tt>, and <tt>request_id</tt>. # This can be returned by a health check against the connection. def statistics { identifier: connection_identifier, started_at: @started_at, subscriptions: subscriptions.identifiers, - request_id: @env['action_dispatch.request_id'] + request_id: @env["action_dispatch.request_id"] } end @@ -195,7 +195,7 @@ module ActionCable def allow_request_origin? return true if server.config.disable_request_forgery_protection - if Array(server.config.allowed_request_origins).any? { |allowed_origin| allowed_origin === env['HTTP_ORIGIN'] } + if Array(server.config.allowed_request_origins).any? { |allowed_origin| allowed_origin === env["HTTP_ORIGIN"] } true else logger.error("Request origin not allowed: #{env['HTTP_ORIGIN']}") @@ -213,7 +213,7 @@ module ActionCable logger.error invalid_request_message logger.info finished_request_message - [ 404, { 'Content-Type' => 'text/plain' }, [ 'Page not found' ] ] + [ 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. @@ -226,7 +226,7 @@ module ActionCable 'Started %s "%s"%s for %s at %s' % [ request.request_method, request.filtered_path, - websocket.possible? ? ' [WebSocket]' : '[non-WebSocket]', + websocket.possible? ? " [WebSocket]" : "[non-WebSocket]", request.ip, Time.now.to_s ] end @@ -234,19 +234,19 @@ module ActionCable def finished_request_message 'Finished "%s"%s for %s at %s' % [ request.filtered_path, - websocket.possible? ? ' [WebSocket]' : '[non-WebSocket]', + websocket.possible? ? " [WebSocket]" : "[non-WebSocket]", request.ip, Time.now.to_s ] end def invalid_request_message - 'Failed to upgrade to WebSocket (REQUEST_METHOD: %s, HTTP_CONNECTION: %s, HTTP_UPGRADE: %s)' % [ + "Failed to upgrade to WebSocket (REQUEST_METHOD: %s, HTTP_CONNECTION: %s, HTTP_UPGRADE: %s)" % [ env["REQUEST_METHOD"], env["HTTP_CONNECTION"], env["HTTP_UPGRADE"] ] end def successful_request_message - 'Successfully upgraded to WebSocket (REQUEST_METHOD: %s, HTTP_CONNECTION: %s, HTTP_UPGRADE: %s)' % [ + "Successfully upgraded to WebSocket (REQUEST_METHOD: %s, HTTP_CONNECTION: %s, HTTP_UPGRADE: %s)" % [ env["REQUEST_METHOD"], env["HTTP_CONNECTION"], env["HTTP_UPGRADE"] ] end diff --git a/actioncable/lib/action_cable/connection/client_socket.rb b/actioncable/lib/action_cable/connection/client_socket.rb index 6f29f32ea9..70a2bbecb1 100644 --- a/actioncable/lib/action_cable/connection/client_socket.rb +++ b/actioncable/lib/action_cable/connection/client_socket.rb @@ -1,4 +1,4 @@ -require 'websocket/driver' +require "websocket/driver" module ActionCable module Connection @@ -8,16 +8,16 @@ module ActionCable # Copyright (c) 2010-2015 James Coglan class ClientSocket # :nodoc: def self.determine_url(env) - scheme = secure_request?(env) ? 'wss:' : 'ws:' + scheme = secure_request?(env) ? "wss:" : "ws:" "#{ scheme }//#{ env['HTTP_HOST'] }#{ env['REQUEST_URI'] }" end def self.secure_request?(env) - return true if env['HTTPS'] == 'on' - return true if env['HTTP_X_FORWARDED_SSL'] == 'on' - return true if env['HTTP_X_FORWARDED_SCHEME'] == 'https' - return true if env['HTTP_X_FORWARDED_PROTO'] == 'https' - return true if env['rack.url_scheme'] == 'https' + return true if env["HTTPS"] == "on" + return true if env["HTTP_X_FORWARDED_SSL"] == "on" + return true if env["HTTP_X_FORWARDED_SCHEME"] == "https" + return true if env["HTTP_X_FORWARDED_PROTO"] == "https" + return true if env["rack.url_scheme"] == "https" return false end @@ -37,7 +37,7 @@ module ActionCable @url = ClientSocket.determine_url(@env) @driver = @driver_started = nil - @close_params = ['', 1006] + @close_params = ["", 1006] @ready_state = CONNECTING @@ -56,7 +56,7 @@ module ActionCable return if @driver.nil? || @driver_started @stream.hijack_rack_socket - if callback = @env['async.callback'] + if callback = @env["async.callback"] callback.call([101, {}, @stream]) end @@ -78,18 +78,18 @@ module ActionCable def transmit(message) return false if @ready_state > OPEN case message - when Numeric then @driver.text(message.to_s) - when String then @driver.text(message) - when Array then @driver.binary(message) + when Numeric then @driver.text(message.to_s) + when String then @driver.text(message) + when Array then @driver.binary(message) else false end end def close(code = nil, reason = nil) code ||= 1000 - reason ||= '' + reason ||= "" - unless code == 1000 or (code >= 3000 and code <= 4999) + unless code == 1000 || (code >= 3000 && code <= 4999) raise ArgumentError, "Failed to execute 'close' on WebSocket: " + "The code must be either 1000, or between 3000 and 4999. " + "#{code} is neither." diff --git a/actioncable/lib/action_cable/connection/faye_client_socket.rb b/actioncable/lib/action_cable/connection/faye_client_socket.rb deleted file mode 100644 index a4bfe7db17..0000000000 --- a/actioncable/lib/action_cable/connection/faye_client_socket.rb +++ /dev/null @@ -1,48 +0,0 @@ -require 'faye/websocket' - -module ActionCable - module Connection - class FayeClientSocket - def initialize(env, event_target, stream_event_loop, protocols) - @env = env - @event_target = event_target - @protocols = protocols - - @faye = nil - end - - def alive? - @faye && @faye.ready_state == Faye::WebSocket::API::OPEN - end - - def transmit(data) - connect - @faye.send data - end - - def close - @faye && @faye.close - end - - def protocol - @faye && @faye.protocol - end - - def rack_response - connect - @faye.rack_response - end - - private - def connect - return if @faye - @faye = Faye::WebSocket.new(@env, @protocols) - - @faye.on(:open) { |event| @event_target.on_open } - @faye.on(:message) { |event| @event_target.on_message(event.data) } - @faye.on(:close) { |event| @event_target.on_close(event.reason, event.code) } - @faye.on(:error) { |event| @event_target.on_error(event.message) } - end - end - end -end diff --git a/actioncable/lib/action_cable/connection/faye_event_loop.rb b/actioncable/lib/action_cable/connection/faye_event_loop.rb deleted file mode 100644 index 9c44b38bc3..0000000000 --- a/actioncable/lib/action_cable/connection/faye_event_loop.rb +++ /dev/null @@ -1,44 +0,0 @@ -require 'thread' - -require 'eventmachine' -EventMachine.epoll if EventMachine.epoll? -EventMachine.kqueue if EventMachine.kqueue? - -module ActionCable - module Connection - class FayeEventLoop - @@mutex = Mutex.new - - def timer(interval, &block) - ensure_reactor_running - EMTimer.new(::EM::PeriodicTimer.new(interval, &block)) - end - - def post(task = nil, &block) - task ||= block - - ensure_reactor_running - ::EM.next_tick(&task) - end - - private - def ensure_reactor_running - return if EventMachine.reactor_running? - @@mutex.synchronize do - Thread.new { EventMachine.run } unless EventMachine.reactor_running? - Thread.pass until EventMachine.reactor_running? - end - end - - class EMTimer - def initialize(inner) - @inner = inner - end - - def shutdown - @inner.cancel - end - end - end - end -end diff --git a/actioncable/lib/action_cable/connection/identification.rb b/actioncable/lib/action_cable/connection/identification.rb index 4a54044aff..c91a1d1fd7 100644 --- a/actioncable/lib/action_cable/connection/identification.rb +++ b/actioncable/lib/action_cable/connection/identification.rb @@ -1,4 +1,4 @@ -require 'set' +require "set" module ActionCable module Connection diff --git a/actioncable/lib/action_cable/connection/internal_channel.rb b/actioncable/lib/action_cable/connection/internal_channel.rb index f70d52f99b..8f0ec766c3 100644 --- a/actioncable/lib/action_cable/connection/internal_channel.rb +++ b/actioncable/lib/action_cable/connection/internal_channel.rb @@ -27,8 +27,8 @@ module ActionCable end def process_internal_message(message) - case message['type'] - when 'disconnect' + case message["type"] + when "disconnect" logger.info "Removing connection (#{connection_identifier})" websocket.close end diff --git a/actioncable/lib/action_cable/connection/stream.rb b/actioncable/lib/action_cable/connection/stream.rb index 0cf59091bc..d66e1b4e41 100644 --- a/actioncable/lib/action_cable/connection/stream.rb +++ b/actioncable/lib/action_cable/connection/stream.rb @@ -1,3 +1,5 @@ +require "thread" + module ActionCable module Connection #-- @@ -8,9 +10,13 @@ module ActionCable def initialize(event_loop, socket) @event_loop = event_loop @socket_object = socket - @stream_send = socket.env['stream.send'] + @stream_send = socket.env["stream.send"] @rack_hijack_io = nil + @write_lock = Mutex.new + + @write_head = nil + @write_buffer = Queue.new end def each(&callback) @@ -27,21 +33,71 @@ module ActionCable end def write(data) - return @rack_hijack_io.write(data) if @rack_hijack_io - return @stream_send.call(data) if @stream_send + if @stream_send + return @stream_send.call(data) + end + + if @write_lock.try_lock + begin + if @write_head.nil? && @write_buffer.empty? + written = @rack_hijack_io.write_nonblock(data, exception: false) + + case written + when :wait_writable + # proceed below + when data.bytesize + return data.bytesize + else + @write_head = data.byteslice(written, data.bytesize) + @event_loop.writes_pending @rack_hijack_io + + return data.bytesize + end + end + ensure + @write_lock.unlock + end + end + + @write_buffer << data + @event_loop.writes_pending @rack_hijack_io + + data.bytesize rescue EOFError, Errno::ECONNRESET @socket_object.client_gone end + def flush_write_buffer + @write_lock.synchronize do + loop do + if @write_head.nil? + return true if @write_buffer.empty? + @write_head = @write_buffer.pop + end + + written = @rack_hijack_io.write_nonblock(@write_head, exception: false) + case written + when :wait_writable + return false + when @write_head.bytesize + @write_head = nil + else + @write_head = @write_head.byteslice(written, @write_head.bytesize) + return false + end + end + end + end + def receive(data) @socket_object.parse(data) end def hijack_rack_socket - return unless @socket_object.env['rack.hijack'] + return unless @socket_object.env["rack.hijack"] - @socket_object.env['rack.hijack'].call - @rack_hijack_io = @socket_object.env['rack.hijack_io'] + @socket_object.env["rack.hijack"].call + @rack_hijack_io = @socket_object.env["rack.hijack_io"] @event_loop.attach(@rack_hijack_io, self) end @@ -50,6 +106,7 @@ module ActionCable def clean_rack_hijack return unless @rack_hijack_io @event_loop.detach(@rack_hijack_io, self) + @rack_hijack_io.close @rack_hijack_io = nil end end diff --git a/actioncable/lib/action_cable/connection/stream_event_loop.rb b/actioncable/lib/action_cable/connection/stream_event_loop.rb index 2abad09c03..eec24638b6 100644 --- a/actioncable/lib/action_cable/connection/stream_event_loop.rb +++ b/actioncable/lib/action_cable/connection/stream_event_loop.rb @@ -1,11 +1,11 @@ -require 'nio' -require 'thread' +require "nio" +require "thread" module ActionCable module Connection class StreamEventLoop def initialize - @nio = @thread = nil + @nio = @executor = @thread = nil @map = {} @stopping = false @todo = Queue.new @@ -20,13 +20,14 @@ module ActionCable def post(task = nil, &block) task ||= block - Concurrent.global_io_executor << task + spawn + @executor << task end def attach(io, stream) @todo << lambda do - @map[io] = stream - @nio.register(io, :r) + @map[io] = @nio.register(io, :r) + @map[io].value = stream end wakeup end @@ -39,6 +40,15 @@ module ActionCable wakeup end + def writes_pending(io) + @todo << lambda do + if monitor = @map[io] + monitor.interests = :rw + end + end + wakeup + end + def stop @stopping = true wakeup if @nio @@ -52,6 +62,13 @@ module ActionCable return if @thread && @thread.status @nio ||= NIO::Selector.new + + @executor ||= Concurrent::ThreadPoolExecutor.new( + min_threads: 1, + max_threads: 10, + max_queue: 0, + ) + @thread = Thread.new { run } return true @@ -77,12 +94,25 @@ module ActionCable monitors.each do |monitor| io = monitor.io - stream = @map[io] + stream = monitor.value begin - stream.receive io.read_nonblock(4096) - rescue IO::WaitReadable - next + if monitor.writable? + if stream.flush_write_buffer + monitor.interests = :r + end + next unless monitor.readable? + end + + incoming = io.read_nonblock(4096, exception: false) + case incoming + when :wait_readable + next + when nil + stream.close + else + stream.receive incoming + end rescue # We expect one of EOFError or Errno::ECONNRESET in # normal operation (when the client goes away). But if diff --git a/actioncable/lib/action_cable/connection/subscriptions.rb b/actioncable/lib/action_cable/connection/subscriptions.rb index 3742f248d1..00511aead5 100644 --- a/actioncable/lib/action_cable/connection/subscriptions.rb +++ b/actioncable/lib/action_cable/connection/subscriptions.rb @@ -1,4 +1,4 @@ -require 'active_support/core_ext/hash/indifferent_access' +require "active_support/core_ext/hash/indifferent_access" module ActionCable module Connection @@ -11,10 +11,10 @@ module ActionCable end def execute_command(data) - case data['command'] - when 'subscribe' then add data - when 'unsubscribe' then remove data - when 'message' then perform_action data + case data["command"] + when "subscribe" then add data + when "unsubscribe" then remove data + when "message" then perform_action data else logger.error "Received unrecognized command in #{data.inspect}" end @@ -23,21 +23,25 @@ module ActionCable end def add(data) - id_key = data['identifier'] + id_key = data["identifier"] id_options = ActiveSupport::JSON.decode(id_key).with_indifferent_access - subscription_klass = connection.server.channel_classes[id_options[:channel]] + return if subscriptions.key?(id_key) - if subscription_klass - subscriptions[id_key] ||= subscription_klass.new(connection, id_key, id_options) + subscription_klass = id_options[:channel].safe_constantize + + if subscription_klass && ActionCable::Channel::Base >= subscription_klass + subscription = subscription_klass.new(connection, id_key, id_options) + subscriptions[id_key] = subscription + subscription.subscribe_to_channel else - logger.error "Subscription class not found (#{data.inspect})" + logger.error "Subscription class not found: #{id_options[:channel].inspect}" end end def remove(data) logger.info "Unsubscribing from channel: #{data['identifier']}" - remove_subscription subscriptions[data['identifier']] + remove_subscription subscriptions[data["identifier"]] end def remove_subscription(subscription) @@ -46,7 +50,7 @@ module ActionCable end def perform_action(data) - find(data).perform_action ActiveSupport::JSON.decode(data['data']) + find(data).perform_action ActiveSupport::JSON.decode(data["data"]) end def identifiers @@ -64,7 +68,7 @@ module ActionCable delegate :logger, to: :connection def find(data) - if subscription = subscriptions[data['identifier']] + if subscription = subscriptions[data["identifier"]] subscription else raise "Unable to find subscription with identifier: #{data['identifier']}" diff --git a/actioncable/lib/action_cable/connection/web_socket.rb b/actioncable/lib/action_cable/connection/web_socket.rb index 11f28c37e8..382141b89f 100644 --- a/actioncable/lib/action_cable/connection/web_socket.rb +++ b/actioncable/lib/action_cable/connection/web_socket.rb @@ -1,11 +1,11 @@ -require 'websocket/driver' +require "websocket/driver" module ActionCable module Connection # Wrap the real socket to minimize the externally-presented API class WebSocket - def initialize(env, event_target, event_loop, client_socket_class, protocols: ActionCable::INTERNAL[:protocols]) - @websocket = ::WebSocket::Driver.websocket?(env) ? client_socket_class.new(env, event_target, event_loop, protocols) : nil + def initialize(env, event_target, event_loop, protocols: ActionCable::INTERNAL[:protocols]) + @websocket = ::WebSocket::Driver.websocket?(env) ? ClientSocket.new(env, event_target, event_loop, protocols) : nil end def possible? diff --git a/actioncable/lib/action_cable/engine.rb b/actioncable/lib/action_cable/engine.rb index 7dc541d00c..4c5c975cd8 100644 --- a/actioncable/lib/action_cable/engine.rb +++ b/actioncable/lib/action_cable/engine.rb @@ -4,7 +4,7 @@ require "action_cable/helpers/action_cable_helper" require "active_support/core_ext/hash/indifferent_access" module ActionCable - class Railtie < Rails::Engine # :nodoc: + class Engine < Rails::Engine # :nodoc: config.action_cable = ActiveSupport::OrderedOptions.new config.action_cable.mount_path = ActionCable::INTERNAL[:default_mount_path] @@ -22,7 +22,7 @@ module ActionCable initializer "action_cable.set_configs" do |app| options = app.config.action_cable - options.allowed_request_origins ||= "http://localhost:3000" if ::Rails.env.development? + options.allowed_request_origins ||= /https?:\/\/localhost:\d+/ if ::Rails.env.development? app.paths.add "config/cable", with: "config/cable.yml" @@ -31,11 +31,8 @@ module ActionCable self.cable = Rails.application.config_for(config_path).with_indifferent_access end - if 'ApplicationCable::Connection'.safe_constantize - self.connection_class = ApplicationCable::Connection - end - - self.channel_paths = Rails.application.paths['app/channels'].existent + previous_connection_class = self.connection_class + self.connection_class = -> { "ApplicationCable::Connection".safe_constantize || previous_connection_class.call } options.each { |k,v| send("#{k}=", v) } end diff --git a/actioncable/lib/action_cable/gem_version.rb b/actioncable/lib/action_cable/gem_version.rb index 5ca2f473b1..8ba0230d47 100644 --- a/actioncable/lib/action_cable/gem_version.rb +++ b/actioncable/lib/action_cable/gem_version.rb @@ -6,9 +6,9 @@ module ActionCable module VERSION MAJOR = 5 - MINOR = 0 + MINOR = 1 TINY = 0 - PRE = "beta4" + PRE = "alpha" STRING = [MAJOR, MINOR, TINY, PRE].compact.join(".") end diff --git a/actioncable/lib/action_cable/remote_connections.rb b/actioncable/lib/action_cable/remote_connections.rb index a528024427..720ba52d19 100644 --- a/actioncable/lib/action_cable/remote_connections.rb +++ b/actioncable/lib/action_cable/remote_connections.rb @@ -41,7 +41,7 @@ module ActionCable # Uses the internal channel to disconnect the connection. def disconnect - server.broadcast internal_channel, type: 'disconnect' + server.broadcast internal_channel, type: "disconnect" end # Returns all the identifiers that were applied to this connection. diff --git a/actioncable/lib/action_cable/server.rb b/actioncable/lib/action_cable/server.rb index bd6a3826a3..22f9353825 100644 --- a/actioncable/lib/action_cable/server.rb +++ b/actioncable/lib/action_cable/server.rb @@ -9,7 +9,7 @@ module ActionCable autoload :Configuration autoload :Worker - autoload :ActiveRecordConnectionManagement, 'action_cable/server/worker/active_record_connection_management' + autoload :ActiveRecordConnectionManagement, "action_cable/server/worker/active_record_connection_management" end end end diff --git a/actioncable/lib/action_cable/server/base.rb b/actioncable/lib/action_cable/server/base.rb index b1a0e11631..419eccd73c 100644 --- a/actioncable/lib/action_cable/server/base.rb +++ b/actioncable/lib/action_cable/server/base.rb @@ -1,4 +1,4 @@ -require 'monitor' +require "monitor" module ActionCable module Server @@ -19,13 +19,13 @@ module ActionCable def initialize @mutex = Monitor.new - @remote_connections = @event_loop = @worker_pool = @channel_classes = @pubsub = nil + @remote_connections = @event_loop = @worker_pool = @pubsub = nil end # Called by Rack to setup the server. def call(env) setup_heartbeat_timer - config.connection_class.new(self, env).process + config.connection_class.call.new(self, env).process end # Disconnect all the connections identified by `identifiers` on this server or any others via RemoteConnections. @@ -37,9 +37,13 @@ module ActionCable connections.each(&:close) @mutex.synchronize do - worker_pool.halt if @worker_pool - + # Shutdown the worker pool + @worker_pool.halt if @worker_pool @worker_pool = nil + + # Shutdown the pub/sub adapter + @pubsub.shutdown if @pubsub + @pubsub = nil end end @@ -49,34 +53,24 @@ module ActionCable end def event_loop - @event_loop || @mutex.synchronize { @event_loop ||= config.event_loop_class.new } + @event_loop || @mutex.synchronize { @event_loop ||= ActionCable::Connection::StreamEventLoop.new } end # The worker pool is where we run connection callbacks and channel actions. We do as little as possible on the server's main thread. # The worker pool is an executor service that's backed by a pool of threads working from a task queue. The thread pool size maxes out - # at 4 worker threads by default. Tune the size yourself with config.action_cable.worker_pool_size. + # at 4 worker threads by default. Tune the size yourself with <tt>config.action_cable.worker_pool_size</tt>. # # Using Active Record, Redis, etc within your channel actions means you'll get a separate connection from each thread in the worker pool. # Plan your deployment accordingly: 5 servers each running 5 Puma workers each running an 8-thread worker pool means at least 200 database # connections. # # Also, ensure that your database connection pool size is as least as large as your worker pool size. Otherwise, workers may oversubscribe - # the db connection pool and block while they wait for other workers to release their connections. Use a smaller worker pool or a larger - # db connection pool instead. + # the database connection pool and block while they wait for other workers to release their connections. Use a smaller worker pool or a larger + # database connection pool instead. def worker_pool @worker_pool || @mutex.synchronize { @worker_pool ||= ActionCable::Server::Worker.new(max_size: config.worker_pool_size) } end - # Requires and returns a hash of all of the channel class constants, which are keyed by name. - def channel_classes - @channel_classes || @mutex.synchronize do - @channel_classes ||= begin - config.channel_paths.each { |channel_path| require channel_path } - config.channel_class_names.each_with_object({}) { |name, hash| hash[name] = name.constantize } - end - end - end - # Adapter used for all streams/broadcasting. def pubsub @pubsub || @mutex.synchronize { @pubsub ||= config.pubsub_adapter.new(self) } @@ -84,7 +78,7 @@ module ActionCable # All of the identifiers applied to the connection class associated with this server. def connection_identifiers - config.connection_class.identifiers + config.connection_class.call.identifiers end end diff --git a/actioncable/lib/action_cable/server/broadcasting.rb b/actioncable/lib/action_cable/server/broadcasting.rb index 8f93564113..1fc58baa3e 100644 --- a/actioncable/lib/action_cable/server/broadcasting.rb +++ b/actioncable/lib/action_cable/server/broadcasting.rb @@ -39,8 +39,12 @@ module ActionCable def broadcast(message) server.logger.info "[ActionCable] Broadcasting to #{broadcasting}: #{message.inspect}" - encoded = coder ? coder.encode(message) : message - server.pubsub.broadcast broadcasting, encoded + + payload = { broadcasting: broadcasting, message: message, coder: coder } + ActiveSupport::Notifications.instrument("broadcast.action_cable", payload) do + encoded = coder ? coder.encode(message) : message + server.pubsub.broadcast broadcasting, encoded + end end end end diff --git a/actioncable/lib/action_cable/server/configuration.rb b/actioncable/lib/action_cable/server/configuration.rb index 0bb378cf03..dc146f07b0 100644 --- a/actioncable/lib/action_cable/server/configuration.rb +++ b/actioncable/lib/action_cable/server/configuration.rb @@ -4,32 +4,24 @@ module ActionCable # in a Rails config initializer. class Configuration attr_accessor :logger, :log_tags - attr_accessor :use_faye, :connection_class, :worker_pool_size + attr_accessor :connection_class, :worker_pool_size attr_accessor :disable_request_forgery_protection, :allowed_request_origins attr_accessor :cable, :url, :mount_path - attr_accessor :channel_paths # :nodoc: - def initialize @log_tags = [] - @connection_class = ActionCable::Connection::Base + @connection_class = -> { ActionCable::Connection::Base } @worker_pool_size = 4 @disable_request_forgery_protection = false end - def channel_class_names - @channel_class_names ||= channel_paths.collect do |channel_path| - Pathname.new(channel_path).basename.to_s.split('.').first.camelize - end - end - # Returns constant of subscription adapter specified in config/cable.yml. # If the adapter cannot be found, this will default to the Redis adapter. # Also makes sure proper dependencies are required. def pubsub_adapter - adapter = (cable.fetch('adapter') { 'redis' }) + adapter = (cable.fetch("adapter") { "redis" }) path_to_adapter = "action_cable/subscription_adapter/#{adapter}" begin require path_to_adapter @@ -40,25 +32,9 @@ module ActionCable end adapter = adapter.camelize - adapter = 'PostgreSQL' if adapter == 'Postgresql' + adapter = "PostgreSQL" if adapter == "Postgresql" "ActionCable::SubscriptionAdapter::#{adapter}".constantize end - - def event_loop_class - if use_faye - ActionCable::Connection::FayeEventLoop - else - ActionCable::Connection::StreamEventLoop - end - end - - def client_socket_class - if use_faye - ActionCable::Connection::FayeClientSocket - else - ActionCable::Connection::ClientSocket - end - end end end end diff --git a/actioncable/lib/action_cable/server/worker.rb b/actioncable/lib/action_cable/server/worker.rb index a638ff72e7..43639c27af 100644 --- a/actioncable/lib/action_cable/server/worker.rb +++ b/actioncable/lib/action_cable/server/worker.rb @@ -1,6 +1,6 @@ -require 'active_support/callbacks' -require 'active_support/core_ext/module/attribute_accessors_per_thread' -require 'concurrent' +require "active_support/callbacks" +require "active_support/core_ext/module/attribute_accessors_per_thread" +require "concurrent" module ActionCable module Server @@ -25,7 +25,7 @@ module ActionCable # Stop processing work: any work that has not already started # running will be discarded from the queue def halt - @executor.kill + @executor.shutdown end def stopping? @@ -42,16 +42,20 @@ module ActionCable self.connection = nil end - def async_invoke(receiver, method, *args, connection: receiver) + def async_exec(receiver, *args, connection:, &block) + async_invoke receiver, :instance_exec, *args, connection: connection, &block + end + + def async_invoke(receiver, method, *args, connection: receiver, &block) @executor.post do - invoke(receiver, method, *args, connection: connection) + invoke(receiver, method, *args, connection: connection, &block) end end - def invoke(receiver, method, *args, connection:) + def invoke(receiver, method, *args, connection:, &block) work(connection) do begin - receiver.send method, *args + receiver.send method, *args, &block rescue Exception => e logger.error "There was an exception - #{e.class}(#{e.message})" logger.error e.backtrace.join("\n") diff --git a/actioncable/lib/action_cable/subscription_adapter/async.rb b/actioncable/lib/action_cable/subscription_adapter/async.rb index 10b3ac8cd8..46819dbfec 100644 --- a/actioncable/lib/action_cable/subscription_adapter/async.rb +++ b/actioncable/lib/action_cable/subscription_adapter/async.rb @@ -1,4 +1,4 @@ -require 'action_cable/subscription_adapter/inline' +require "action_cable/subscription_adapter/inline" module ActionCable module SubscriptionAdapter diff --git a/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb b/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb index 4735a4bfa8..bcd46d2a0e 100644 --- a/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb +++ b/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb @@ -1,9 +1,9 @@ -require 'thread' +require "thread" -gem 'em-hiredis', '~> 0.3.0' -gem 'redis', '~> 3.0' -require 'em-hiredis' -require 'redis' +gem "em-hiredis", "~> 0.3.0" +gem "redis", "~> 3.0" +require "em-hiredis" +require "redis" EventMachine.epoll if EventMachine.epoll? EventMachine.kqueue if EventMachine.kqueue? diff --git a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb index 66c7852f6e..bdab5205ec 100644 --- a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb +++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb @@ -1,6 +1,6 @@ -gem 'pg', '~> 0.18' -require 'pg' -require 'thread' +gem "pg", "~> 0.18" +require "pg" +require "thread" module ActionCable module SubscriptionAdapter @@ -33,7 +33,7 @@ module ActionCable pg_conn = ar_conn.raw_connection unless pg_conn.is_a?(PG::Connection) - raise 'ActiveRecord database must be Postgres in order to use the Postgres ActionCable storage adapter' + raise "The Active Record database must be PostgreSQL in order to use the PostgreSQL Action Cable storage adapter" end yield pg_conn diff --git a/actioncable/lib/action_cable/subscription_adapter/redis.rb b/actioncable/lib/action_cable/subscription_adapter/redis.rb index 65434f7107..62bd284a6b 100644 --- a/actioncable/lib/action_cable/subscription_adapter/redis.rb +++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb @@ -1,7 +1,7 @@ -require 'thread' +require "thread" -gem 'redis', '~> 3.0' -require 'redis' +gem "redis", "~> 3.0" +require "redis" module ActionCable module SubscriptionAdapter @@ -72,7 +72,7 @@ module ActionCable conn.without_reconnect do original_client = conn.client - conn.subscribe('_action_cable_internal') do |on| + conn.subscribe("_action_cable_internal") do |on| on.subscribe do |chan, count| @subscription_lock.synchronize do if count == 1 @@ -111,7 +111,7 @@ module ActionCable return if @thread.nil? when_connected do - send_command('unsubscribe') + send_command("unsubscribe") @raw_client = nil end end @@ -123,13 +123,13 @@ module ActionCable @subscription_lock.synchronize do ensure_listener_running @subscribe_callbacks[channel] << on_success - when_connected { send_command('subscribe', channel) } + when_connected { send_command("subscribe", channel) } end end def remove_channel(channel) @subscription_lock.synchronize do - when_connected { send_command('unsubscribe', channel) } + when_connected { send_command("unsubscribe", channel) } end end diff --git a/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb b/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb index 37eed09793..4ec513e3ba 100644 --- a/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb +++ b/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb @@ -32,7 +32,11 @@ module ActionCable end def broadcast(channel, message) - list = @sync.synchronize { @subscribers[channel].dup } + list = @sync.synchronize do + return if !@subscribers.key?(channel) + @subscribers[channel].dup + end + list.each do |subscriber| invoke_callback(subscriber, message) end diff --git a/actioncable/lib/action_cable/version.rb b/actioncable/lib/action_cable/version.rb index e17877202b..d6081409f0 100644 --- a/actioncable/lib/action_cable/version.rb +++ b/actioncable/lib/action_cable/version.rb @@ -1,4 +1,4 @@ -require_relative 'gem_version' +require_relative "gem_version" module ActionCable # Returns the version of the currently loaded Action Cable as a <tt>Gem::Version</tt> |