diff options
Diffstat (limited to 'actioncable/lib')
44 files changed, 1148 insertions, 600 deletions
diff --git a/actioncable/lib/action_cable.rb b/actioncable/lib/action_cable.rb index f27698765c..1dc66ef3ad 100644 --- a/actioncable/lib/action_cable.rb +++ b/actioncable/lib/action_cable.rb @@ -1,5 +1,5 @@ #-- -# Copyright (c) 2015 Basecamp, LLC +# Copyright (c) 2015-2016 Basecamp, LLC # # Permission is hereby granted, free of charge, to any person obtaining # a copy of this software and associated documentation files (the @@ -47,4 +47,5 @@ module ActionCable autoload :Connection autoload :Channel autoload :RemoteConnections + autoload :SubscriptionAdapter end diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb index ce9d62635c..714d9887d4 100644 --- a/actioncable/lib/action_cable/channel/base.rb +++ b/actioncable/lib/action_cable/channel/base.rb @@ -32,8 +32,11 @@ module ActionCable # # == Action processing # - # Unlike Action Controllers, channels do not follow a REST constraint form for its actions. It's a remote-procedure call model. You can - # declare any public method on the channel (optionally taking a data argument), and this method is automatically exposed as callable to the client. + # Unlike subclasses of ActionController::Base, channels do not follow a RESTful + # constraint form for their actions. Instead, Action Cable operates through a + # remote-procedure call model. You can declare any public method on the + # channel (optionally taking a <tt>data</tt> argument), and this method is + # automatically exposed as callable to the client. # # Example: # @@ -60,18 +63,22 @@ module ActionCable # end # end # - # In this example, subscribed/unsubscribed are not callable methods, as they were already declared in ActionCable::Channel::Base, but #appear/away - # are. #generate_connection_token is also not callable as its a private method. You'll see that appear accepts a data parameter, which it then - # uses as part of its model call. #away does not, it's simply a trigger action. + # In this example, the subscribed and unsubscribed methods are not callable methods, as they + # were already declared in ActionCable::Channel::Base, but <tt>#appear</tt> + # and <tt>#away</tt> are. <tt>#generate_connection_token</tt> is also not + # callable, since it's a private method. You'll see that appear accepts a data + # parameter, which it then uses as part of its model call. <tt>#away</tt> + # does not, since it's simply a trigger action. # - # Also note that in this example, current_user is available because it was marked as an identifying attribute on the connection. - # All such identifiers will automatically create a delegation method of the same name on the channel instance. + # Also note that in this example, <tt>current_user</tt> is available because + # it was marked as an identifying attribute on the connection. All such + # identifiers will automatically create a delegation method of the same name + # on the channel instance. # # == Rejecting subscription requests # - # A channel can reject a subscription request in the #subscribed callback by invoking #reject! - # - # Example: + # A channel can reject a subscription request in the #subscribed callback by + # invoking the #reject method: # # class ChatChannel < ApplicationCable::Channel # def subscribed @@ -80,8 +87,10 @@ module ActionCable # end # end # - # In this example, the subscription will be rejected if the current_user does not have access to the chat room. - # On the client-side, Channel#rejected callback will get invoked when the server rejects the subscription request. + # In this example, the subscription will be rejected if the + # <tt>current_user</tt> does not have access to the chat room. On the + # client-side, the <tt>Channel#rejected</tt> callback will get invoked when + # the server rejects the subscription request. class Base include Callbacks include PeriodicTimers @@ -116,7 +125,7 @@ module ActionCable protected # action_methods are cached and there is sometimes need to refresh # them. ::clear_action_methods! allows you to do that, so next time - # you run action_methods, they will be recalculated + # you run action_methods, they will be recalculated. def clear_action_methods! @action_methods = nil end @@ -133,8 +142,8 @@ module ActionCable @identifier = identifier @params = params - # When a channel is streaming via redis pubsub, we want to delay the confirmation - # transmission until redis pubsub subscription is confirmed. + # When a channel is streaming via pubsub, we want to delay the confirmation + # transmission until pubsub subscription is confirmed. @defer_subscription_confirmation = false @reject_subscription = nil @@ -157,9 +166,9 @@ module ActionCable end end - # Called by the cable connection when its cut so the channel has a chance to cleanup with callbacks. + # 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 + def unsubscribe_from_channel # :nodoc: run_callbacks :unsubscribe do unsubscribed end @@ -174,7 +183,7 @@ module ActionCable end # Called once a consumer has cut its cable connection. Can be used for cleaning up connections or marking - # people as offline or the like. + # users as offline or the like. def unsubscribed # Override in subclasses end @@ -182,7 +191,7 @@ module ActionCable # Transmit a hash of data to the subscriber. The hash will automatically be wrapped in a JSON envelope with # the proper channel identifier marked as the recipient. def transmit(data, via: nil) - logger.info "#{self.class.name} transmitting #{data.inspect}".tap { |m| m << " (via #{via})" if via } + logger.info "#{self.class.name} transmitting #{data.inspect.truncate(300)}".tap { |m| m << " (via #{via})" if via } connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data) end @@ -215,7 +224,6 @@ module ActionCable end end - def subscribe_to_channel run_callbacks :subscribe do subscribed @@ -228,7 +236,6 @@ module ActionCable end end - def extract_action(data) (data['action'].presence || :receive).to_sym end diff --git a/actioncable/lib/action_cable/channel/periodic_timers.rb b/actioncable/lib/action_cable/channel/periodic_timers.rb index 25fe8e5e54..0f6e854520 100644 --- a/actioncable/lib/action_cable/channel/periodic_timers.rb +++ b/actioncable/lib/action_cable/channel/periodic_timers.rb @@ -12,7 +12,7 @@ module ActionCable end module ClassMethods - # Allow you to call a private method <tt>every</tt> so often seconds. This periodic timer can be useful + # Allows you to call a private method <tt>every</tt> so often seconds. This periodic timer can be useful # for sending a steady flow of updates to a client based off an object that was configured on subscription. # It's an alternative to using streams if the channel is able to do the work internally. def periodically(callback, every:) @@ -27,14 +27,14 @@ 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) + active_periodic_timers << Concurrent::TimerTask.new(execution_interval: options[:every]) do + connection.worker_pool.async_run_periodic_timer(self, callback) end end end def stop_periodic_timers - active_periodic_timers.each { |timer| timer.cancel } + active_periodic_timers.each { |timer| timer.shutdown } end end end diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb index b5ffa17f72..431a5c1063 100644 --- a/actioncable/lib/action_cable/channel/streams.rb +++ b/actioncable/lib/action_cable/channel/streams.rb @@ -1,8 +1,8 @@ module ActionCable module Channel - # Streams allow channels to route broadcastings to the subscriber. A broadcasting is, as discussed elsewhere, a pub/sub queue where any data - # put into it is automatically sent to the clients that are connected at that time. It's purely an online queue, though. If you're not - # streaming a broadcasting at the very moment it sends out an update, you'll not get that update when connecting later. + # Streams allow channels to route broadcastings to the subscriber. A broadcasting is, as discussed elsewhere, a pubsub queue where any data + # placed into it is automatically sent to the clients that are connected at that time. It's purely an online queue, though. If you're not + # streaming a broadcasting at the very moment it sends out an update, you will not get that update, if you connect after it has been sent. # # Most commonly, the streamed broadcast is sent straight to the subscriber on the client-side. The channel just acts as a connector between # the two parties (the broadcaster and the channel subscriber). Here's an example of a channel that allows subscribers to get all new @@ -18,8 +18,10 @@ module ActionCable # end # end # - # So 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. - # That looks like so from that side of things: + # 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. + # + # An example broadcasting for this channel looks like so: # # ActionCable.server.broadcast "comments_for_45", author: 'DHH', content: 'Rails is just swell' # @@ -37,26 +39,27 @@ module ActionCable # # CommentsChannel.broadcast_to(@post, @comment) # - # If you don't just want to parlay the broadcast unfiltered to the subscriber, you can supply a callback that lets you alter what goes out. - # Example below shows how you can use this to provide performance introspection in the process: + # If you don't just want to parlay the broadcast unfiltered to the subscriber, you can also supply a callback that lets you alter what is sent out. + # The below example shows how you can use this to provide performance introspection in the process: # # class ChatChannel < ApplicationCable::Channel - # def subscribed - # @room = Chat::Room[params[:room_number]] + # def subscribed + # @room = Chat::Room[params[:room_number]] # - # stream_for @room, -> (encoded_message) do - # message = ActiveSupport::JSON.decode(encoded_message) + # stream_for @room, -> (encoded_message) do + # message = ActiveSupport::JSON.decode(encoded_message) # - # if message['originated_at'].present? - # elapsed_time = (Time.now.to_f - message['originated_at']).round(2) + # if message['originated_at'].present? + # elapsed_time = (Time.now.to_f - message['originated_at']).round(2) # - # ActiveSupport::Notifications.instrument :performance, measurement: 'Chat.message_delay', value: elapsed_time, action: :timing - # logger.info "Message took #{elapsed_time}s to arrive" - # end + # ActiveSupport::Notifications.instrument :performance, measurement: 'Chat.message_delay', value: elapsed_time, action: :timing + # logger.info "Message took #{elapsed_time}s to arrive" + # end # - # transmit message - # end - # end + # transmit message + # end + # end + # end # # You can stop streaming from all broadcasts by calling #stop_all_streams. module Streams @@ -69,17 +72,18 @@ 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. def stream_from(broadcasting, callback = nil) - # Hold off the confirmation until pubsub#subscribe is successful + broadcasting = String(broadcasting) + # Don't send the confirmation until pubsub#subscribe is successful defer_subscription_confirmation! callback ||= default_stream_callback(broadcasting) streams << [ broadcasting, callback ] - EM.next_tick do - pubsub.subscribe(broadcasting, &callback).callback do |reply| + Concurrent.global_io_executor.post do + pubsub.subscribe(broadcasting, callback, lambda do transmit_subscription_confirmation logger.info "#{self.class.name} is streaming from #{broadcasting}" - end + end) end end @@ -90,9 +94,10 @@ module ActionCable stream_from(broadcasting_for([ channel_name, model ]), callback) end + # Unsubscribes all streams associated with this channel from the pubsub queue. def stop_all_streams streams.each do |broadcasting, callback| - pubsub.unsubscribe_proc broadcasting, callback + pubsub.unsubscribe broadcasting, callback logger.info "#{self.class.name} stopped streaming from #{broadcasting}" end.clear end diff --git a/actioncable/lib/action_cable/connection.rb b/actioncable/lib/action_cable/connection.rb index b672e00682..902efb07e2 100644 --- a/actioncable/lib/action_cable/connection.rb +++ b/actioncable/lib/action_cable/connection.rb @@ -5,12 +5,15 @@ module ActionCable eager_autoload do autoload :Authorization autoload :Base + autoload :ClientSocket autoload :Identification autoload :InternalChannel autoload :MessageBuffer - autoload :WebSocket + autoload :Stream + autoload :StreamEventLoop autoload :Subscriptions autoload :TaggedLoggerProxy + autoload :WebSocket end end end diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb index f7b18a85ae..60f3ad3e06 100644 --- a/actioncable/lib/action_cable/connection/base.rb +++ b/actioncable/lib/action_cable/connection/base.rb @@ -2,9 +2,9 @@ require 'action_dispatch' module ActionCable module Connection - # For every WebSocket the cable server is accepting, a Connection object will be instantiated. This instance becomes the parent - # of all 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 cable consumer. The Connection itself does not deal with any specific application logic beyond + # For every WebSocket 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. # # Here's a basic example: @@ -33,9 +33,9 @@ module ActionCable # end # end # - # First, we declare that this connection can be identified by its current_user. This allows us later to be able to find all connections - # established for that current_user (and potentially disconnect them if the user was removed from an account). You can declare as many - # identification indexes as you like. Declaring an identification means that a attr_accessor is automatically set for that key. + # First, we declare that this connection can be identified by its current_user. This allows us to later be able to find all connections + # established for that current_user (and potentially disconnect them). You can declare as many + # identification indexes as you like. Declaring an identification means that an attr_accessor is automatically set for that key. # # Second, we rely on the fact that the WebSocket connection is established with the cookies from the domain being sent along. This makes # it easy to use signed cookies that were set when logging in via a web interface to authorize the WebSocket connection. @@ -48,41 +48,35 @@ module ActionCable include InternalChannel include Authorization - attr_reader :server, :env, :subscriptions - delegate :worker_pool, :pubsub, to: :server - - attr_reader :logger + attr_reader :server, :env, :subscriptions, :logger + delegate :stream_event_loop, :worker_pool, :pubsub, to: :server def initialize(server, env) @server, @env = server, env @logger = new_tagged_logger - @websocket = ActionCable::Connection::WebSocket.new(env) + @websocket = ActionCable::Connection::WebSocket.new(env, self, stream_event_loop) @subscriptions = ActionCable::Connection::Subscriptions.new(self) @message_buffer = ActionCable::Connection::MessageBuffer.new(self) - @_internal_redis_subscriptions = nil + @_internal_subscriptions = nil @started_at = Time.now end # Called by the server when a new WebSocket connection is established. This configures the callbacks intended for overwriting by the user. - # This method should not be called directly. Rely on the #connect (and #disconnect) callback instead. - def process + # This method should not be called directly -- instead rely upon on the #connect (and #disconnect) callbacks. + def process # :nodoc: logger.info started_request_message if websocket.possible? && allow_request_origin? - websocket.on(:open) { |event| send_async :on_open } - websocket.on(:message) { |event| on_message event.data } - websocket.on(:close) { |event| send_async :on_close } - respond_to_successful_request else respond_to_invalid_request end end - # Data received over the cable is handled by this method. It's expected that everything inbound is JSON encoded. + # Data received over the WebSocket connection is handled by this method. It's expected that everything inbound is JSON encoded. # The data is routed to the proper channel that the connection has subscribed to. def receive(data_in_json) if websocket.alive? @@ -94,7 +88,7 @@ module ActionCable # Send raw data straight back down the WebSocket. This is not intended to be called directly. Use the #transmit available on the # Channel instead, as that'll automatically address the correct subscriber and wrap the message in JSON. - def transmit(data) + def transmit(data) # :nodoc: websocket.transmit data end @@ -105,7 +99,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,6 +117,21 @@ module ActionCable transmit ActiveSupport::JSON.encode(identifier: ActionCable::INTERNAL[:identifiers][:ping], message: Time.now.to_i) end + def on_open # :nodoc: + send_async :handle_open + end + + def on_message(message) # :nodoc: + message_buffer.append message + end + + def on_error(message) # :nodoc: + # ignore + end + + def on_close(reason, code) # :nodoc: + send_async :handle_close + end protected # The request that initiated the WebSocket connection is available here. This gives access to the environment, cookies, etc. @@ -138,16 +147,14 @@ module ActionCable request.cookie_jar end - - protected attr_reader :websocket attr_reader :message_buffer private - def on_open + def handle_open connect if respond_to?(:connect) subscribe_to_internal_channel - beat + confirm_connection_monitor_subscription message_buffer.process! server.add_connection(self) @@ -155,11 +162,7 @@ module ActionCable respond_to_invalid_request end - def on_message(message) - message_buffer.append message - end - - def on_close + def handle_close logger.info finished_request_message server.remove_connection(self) @@ -170,6 +173,12 @@ module ActionCable disconnect if respond_to?(:disconnect) end + def confirm_connection_monitor_subscription + # Send confirmation message to the internal connection monitor channel. + # This ensures the connection monitor state is reset after a successful + # websocket connection. + transmit ActiveSupport::JSON.encode(identifier: ActionCable::INTERNAL[:identifiers][:ping], type: ActionCable::INTERNAL[:message_types][:confirmation]) + end def allow_request_origin? return true if server.config.disable_request_forgery_protection @@ -183,17 +192,18 @@ module ActionCable end def respond_to_successful_request + logger.info successful_request_message websocket.rack_response end def respond_to_invalid_request close if websocket.alive? + logger.error invalid_request_message logger.info finished_request_message [ 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, @@ -204,7 +214,7 @@ module ActionCable 'Started %s "%s"%s for %s at %s' % [ request.request_method, request.filtered_path, - websocket.possible? ? ' [WebSocket]' : '', + websocket.possible? ? ' [WebSocket]' : '[non-WebSocket]', request.ip, Time.now.to_s ] end @@ -212,10 +222,22 @@ module ActionCable def finished_request_message 'Finished "%s"%s for %s at %s' % [ request.filtered_path, - websocket.possible? ? ' [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)' % [ + 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)' % [ + env["REQUEST_METHOD"], env["HTTP_CONNECTION"], env["HTTP_UPGRADE"] + ] + end end end end diff --git a/actioncable/lib/action_cable/connection/client_socket.rb b/actioncable/lib/action_cable/connection/client_socket.rb new file mode 100644 index 0000000000..f6b11e93f0 --- /dev/null +++ b/actioncable/lib/action_cable/connection/client_socket.rb @@ -0,0 +1,149 @@ +require 'websocket/driver' + +module ActionCable + module Connection + #-- + # This class is heavily based on faye-websocket-ruby + # + # Copyright (c) 2010-2015 James Coglan + class ClientSocket # :nodoc: + def self.determine_url(env) + 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 false + end + + CONNECTING = 0 + OPEN = 1 + CLOSING = 2 + CLOSED = 3 + + attr_reader :env, :url + + def initialize(env, event_target, stream_event_loop) + @env = env + @event_target = event_target + @stream_event_loop = stream_event_loop + + @url = ClientSocket.determine_url(@env) + + @driver = @driver_started = nil + @close_params = ['', 1006] + + @ready_state = CONNECTING + + # The driver calls +env+, +url+, and +write+ + @driver = ::WebSocket::Driver.rack(self) + + @driver.on(:open) { |e| open } + @driver.on(:message) { |e| receive_message(e.data) } + @driver.on(:close) { |e| begin_close(e.reason, e.code) } + @driver.on(:error) { |e| emit_error(e.message) } + + @stream = ActionCable::Connection::Stream.new(@stream_event_loop, self) + end + + def start_driver + return if @driver.nil? || @driver_started + @stream.hijack_rack_socket + + if callback = @env['async.callback'] + callback.call([101, {}, @stream]) + end + + @driver_started = true + @driver.start + end + + def rack_response + start_driver + [ -1, {}, [] ] + end + + def write(data) + @stream.write(data) + end + + 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) + else false + end + end + + def close(code = nil, reason = nil) + code ||= 1000 + reason ||= '' + + unless code == 1000 or (code >= 3000 and code <= 4999) + raise ArgumentError, "Failed to execute 'close' on WebSocket: " + + "The code must be either 1000, or between 3000 and 4999. " + + "#{code} is neither." + end + + @ready_state = CLOSING unless @ready_state == CLOSED + @driver.close(reason, code) + end + + def parse(data) + @driver.parse(data) + end + + def client_gone + finalize_close + end + + def alive? + @ready_state == OPEN + end + + private + def open + return unless @ready_state == CONNECTING + @ready_state = OPEN + + @event_target.on_open + end + + def receive_message(data) + return unless @ready_state == OPEN + + @event_target.on_message(data) + end + + def emit_error(message) + return if @ready_state >= CLOSING + + @event_target.on_error(message) + end + + def begin_close(reason, code) + return if @ready_state == CLOSED + @ready_state = CLOSING + @close_params = [reason, code] + + @stream.shutdown if @stream + finalize_close + end + + def finalize_close + return if @ready_state == CLOSED + @ready_state = CLOSED + + @event_target.on_close(*@close_params) + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/identification.rb b/actioncable/lib/action_cable/connection/identification.rb index 2d75ff8d6d..4a54044aff 100644 --- a/actioncable/lib/action_cable/connection/identification.rb +++ b/actioncable/lib/action_cable/connection/identification.rb @@ -11,8 +11,8 @@ 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. - # Common identifiers are current_user and current_account, but could be anything really. + # 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 # channel instances created off the connection. diff --git a/actioncable/lib/action_cable/connection/internal_channel.rb b/actioncable/lib/action_cable/connection/internal_channel.rb index c065a24ab7..27826792b3 100644 --- a/actioncable/lib/action_cable/connection/internal_channel.rb +++ b/actioncable/lib/action_cable/connection/internal_channel.rb @@ -5,24 +5,24 @@ module ActionCable extend ActiveSupport::Concern private - def internal_redis_channel + def internal_channel "action_cable/#{connection_identifier}" end def subscribe_to_internal_channel if connection_identifier.present? callback = -> (message) { process_internal_message(message) } - @_internal_redis_subscriptions ||= [] - @_internal_redis_subscriptions << [ internal_redis_channel, callback ] + @_internal_subscriptions ||= [] + @_internal_subscriptions << [ internal_channel, callback ] - EM.next_tick { pubsub.subscribe(internal_redis_channel, &callback) } + Concurrent.global_io_executor.post { pubsub.subscribe(internal_channel, callback) } logger.info "Registered connection (#{connection_identifier})" end end def unsubscribe_from_internal_channel - if @_internal_redis_subscriptions.present? - @_internal_redis_subscriptions.each { |channel, callback| EM.next_tick { pubsub.unsubscribe_proc(channel, callback) } } + if @_internal_subscriptions.present? + @_internal_subscriptions.each { |channel, callback| Concurrent.global_io_executor.post { pubsub.unsubscribe(channel, callback) } } end end diff --git a/actioncable/lib/action_cable/connection/message_buffer.rb b/actioncable/lib/action_cable/connection/message_buffer.rb index 2f65a1e84a..19f2e6e918 100644 --- a/actioncable/lib/action_cable/connection/message_buffer.rb +++ b/actioncable/lib/action_cable/connection/message_buffer.rb @@ -1,8 +1,7 @@ module ActionCable module Connection - # Allows us to buffer messages received from the WebSocket before the Connection has been fully initialized and is ready to receive them. - # Entirely internal operation and should not be used directly by the user. - class MessageBuffer + # Allows us to buffer messages received from the WebSocket before the Connection has been fully initialized, and is ready to receive them. + class MessageBuffer # :nodoc: def initialize(connection) @connection = connection @buffered_messages = [] diff --git a/actioncable/lib/action_cable/connection/stream.rb b/actioncable/lib/action_cable/connection/stream.rb new file mode 100644 index 0000000000..2d97b28c09 --- /dev/null +++ b/actioncable/lib/action_cable/connection/stream.rb @@ -0,0 +1,57 @@ +module ActionCable + module Connection + #-- + # This class is heavily based on faye-websocket-ruby + # + # Copyright (c) 2010-2015 James Coglan + class Stream # :nodoc: + def initialize(event_loop, socket) + @event_loop = event_loop + @socket_object = socket + @stream_send = socket.env['stream.send'] + + @rack_hijack_io = nil + end + + def each(&callback) + @stream_send ||= callback + end + + def close + shutdown + @socket_object.client_gone + end + + def shutdown + clean_rack_hijack + end + + def write(data) + return @rack_hijack_io.write(data) if @rack_hijack_io + return @stream_send.call(data) if @stream_send + rescue EOFError + @socket_object.client_gone + end + + def receive(data) + @socket_object.parse(data) + end + + def hijack_rack_socket + return unless @socket_object.env['rack.hijack'] + + @socket_object.env['rack.hijack'].call + @rack_hijack_io = @socket_object.env['rack.hijack_io'] + + @event_loop.attach(@rack_hijack_io, self) + end + + private + def clean_rack_hijack + return unless @rack_hijack_io + @event_loop.detach(@rack_hijack_io, self) + @rack_hijack_io = nil + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/stream_event_loop.rb b/actioncable/lib/action_cable/connection/stream_event_loop.rb new file mode 100644 index 0000000000..e6335082d2 --- /dev/null +++ b/actioncable/lib/action_cable/connection/stream_event_loop.rb @@ -0,0 +1,94 @@ +require 'nio' +require 'thread' + +module ActionCable + module Connection + class StreamEventLoop + def initialize + @nio = @thread = nil + @map = {} + @stopping = false + @todo = Queue.new + + @spawn_mutex = Mutex.new + spawn + end + + def attach(io, stream) + @todo << lambda do + @map[io] = stream + @nio.register(io, :r) + end + wakeup + end + + def detach(io, stream) + @todo << lambda do + @nio.deregister io + @map.delete io + end + wakeup + end + + def stop + @stopping = true + wakeup if @nio + end + + private + def spawn + return if @thread && @thread.status + + @spawn_mutex.synchronize do + return if @thread && @thread.status + + @nio ||= NIO::Selector.new + @thread = Thread.new { run } + + return true + end + end + + def wakeup + spawn || @nio.wakeup + end + + def run + loop do + if @stopping + @nio.close + break + end + + until @todo.empty? + @todo.pop(true).call + end + + next unless monitors = @nio.select + + monitors.each do |monitor| + io = monitor.io + stream = @map[io] + + begin + stream.receive io.read_nonblock(4096) + rescue IO::WaitReadable + next + rescue + # We expect one of EOFError or Errno::ECONNRESET in + # normal operation (when the client goes away). But if + # anything else goes wrong, this is still the best way + # to handle it. + begin + stream.close + rescue + @nio.deregister io + @map.delete io + end + end + end + end + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/subscriptions.rb b/actioncable/lib/action_cable/connection/subscriptions.rb index 65d6634bb0..3742f248d1 100644 --- a/actioncable/lib/action_cable/connection/subscriptions.rb +++ b/actioncable/lib/action_cable/connection/subscriptions.rb @@ -3,8 +3,8 @@ require 'active_support/core_ext/hash/indifferent_access' module ActionCable module Connection # Collection class for all the channel subscriptions established on a given connection. Responsible for routing incoming commands that arrive on - # the connection to the proper channel. Should not be used directly by the user. - class Subscriptions + # the connection to the proper channel. + class Subscriptions # :nodoc: def initialize(connection) @connection = connection @subscriptions = {} @@ -49,13 +49,12 @@ module ActionCable find(data).perform_action ActiveSupport::JSON.decode(data['data']) end - def identifiers subscriptions.keys end def unsubscribe_from_all - subscriptions.each { |id, channel| channel.unsubscribe_from_channel } + subscriptions.each { |id, channel| remove_subscription(channel) } end protected diff --git a/actioncable/lib/action_cable/connection/web_socket.rb b/actioncable/lib/action_cable/connection/web_socket.rb index 670d5690ae..5e89fb9b72 100644 --- a/actioncable/lib/action_cable/connection/web_socket.rb +++ b/actioncable/lib/action_cable/connection/web_socket.rb @@ -1,13 +1,11 @@ -require 'faye/websocket' +require 'websocket/driver' module ActionCable module Connection - # Decorate the Faye::WebSocket with helpers we need. + # Wrap the real socket to minimize the externally-presented API class WebSocket - delegate :rack_response, :close, :on, to: :websocket - - def initialize(env) - @websocket = Faye::WebSocket.websocket?(env) ? Faye::WebSocket.new(env) : nil + def initialize(env, event_target, stream_event_loop) + @websocket = ::WebSocket::Driver.websocket?(env) ? ClientSocket.new(env, event_target, stream_event_loop) : nil end def possible? @@ -15,11 +13,19 @@ module ActionCable end def alive? - websocket && websocket.ready_state == Faye::WebSocket::API::OPEN + websocket && websocket.alive? end def transmit(data) - websocket.send data + websocket.transmit data + end + + def close + websocket.close + end + + def rack_response + websocket.rack_response end protected diff --git a/actioncable/lib/action_cable/engine.rb b/actioncable/lib/action_cable/engine.rb index 2d3caa5b0a..ae0c59dccd 100644 --- a/actioncable/lib/action_cable/engine.rb +++ b/actioncable/lib/action_cable/engine.rb @@ -6,7 +6,7 @@ require "active_support/core_ext/hash/indifferent_access" module ActionCable class Railtie < Rails::Engine # :nodoc: config.action_cable = ActiveSupport::OrderedOptions.new - config.action_cable.url = '/cable' + config.action_cable.mount_path = '/cable' config.eager_load_namespaces << ActionCable @@ -24,15 +24,32 @@ module ActionCable options = app.config.action_cable options.allowed_request_origins ||= "http://localhost:3000" if ::Rails.env.development? - app.paths.add "config/redis/cable", with: "config/redis/cable.yml" + app.paths.add "config/cable", with: "config/cable.yml" ActiveSupport.on_load(:action_cable) do - if (redis_cable_path = Pathname.new(app.config.paths["config/redis/cable"].first)).exist? - self.redis = Rails.application.config_for(redis_cable_path).with_indifferent_access + if (config_path = Pathname.new(app.config.paths["config/cable"].first)).exist? + 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 + options.each { |k,v| send("#{k}=", v) } end end + + initializer "action_cable.routes" do + config.after_initialize do |app| + config = app.config + unless config.action_cable.mount_path.nil? + app.routes.prepend do + mount ActionCable.server => config.action_cable.mount_path, internal: true + end + end + end + end end end diff --git a/actioncable/lib/action_cable/gem_version.rb b/actioncable/lib/action_cable/gem_version.rb index b1286aea6f..67adeefaff 100644 --- a/actioncable/lib/action_cable/gem_version.rb +++ b/actioncable/lib/action_cable/gem_version.rb @@ -8,7 +8,7 @@ module ActionCable MAJOR = 5 MINOR = 0 TINY = 0 - PRE = "beta1" + PRE = "beta3" STRING = [MAJOR, MINOR, TINY, PRE].compact.join(".") end diff --git a/actioncable/lib/action_cable/helpers/action_cable_helper.rb b/actioncable/lib/action_cable/helpers/action_cable_helper.rb index b82751468a..2081a37db6 100644 --- a/actioncable/lib/action_cable/helpers/action_cable_helper.rb +++ b/actioncable/lib/action_cable/helpers/action_cable_helper.rb @@ -1,28 +1,39 @@ module ActionCable module Helpers module ActionCableHelper - # Returns an "action-cable-url" meta tag with the value of the url specified in your - # configuration. Ensure this is above your javascript tag: + # Returns an "action-cable-url" meta tag with the value of the URL specified in your + # configuration. Ensure this is above your JavaScript tag: # # <head> # <%= action_cable_meta_tag %> # <%= javascript_include_tag 'application', 'data-turbolinks-track' => true %> # </head> # - # This is then used by ActionCable to determine the url of your websocket server. + # This is then used by Action Cable to determine the URL of your WebSocket server. # Your CoffeeScript can then connect to the server without needing to specify the - # url directly: + # URL directly: # # #= require cable # @App = {} # App.cable = Cable.createConsumer() # - # Make sure to specify the correct server location in each of your environments - # config file: + # Make sure to specify the correct server location in each of your environment + # config files: + # + # config.action_cable.mount_path = "/cable123" + # <%= action_cable_meta_tag %> would render: + # => <meta name="action-cable-url" content="/cable123" /> + # + # config.action_cable.url = "ws://actioncable.com" + # <%= action_cable_meta_tag %> would render: + # => <meta name="action-cable-url" content="ws://actioncable.com" /> # - # config.action_cable.url = "ws://example.com:28080" def action_cable_meta_tag - tag "meta", name: "action-cable-url", content: Rails.application.config.action_cable.url + tag "meta", name: "action-cable-url", content: ( + ActionCable.server.config.url || + ActionCable.server.config.mount_path || + raise("No Action Cable URL configured -- please configure this at config.action_cable.url") + ) end end end diff --git a/actioncable/lib/action_cable/process/logging.rb b/actioncable/lib/action_cable/process/logging.rb deleted file mode 100644 index 72b1a080d1..0000000000 --- a/actioncable/lib/action_cable/process/logging.rb +++ /dev/null @@ -1,10 +0,0 @@ -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/remote_connections.rb b/actioncable/lib/action_cable/remote_connections.rb index 1230d905ad..aeef8abc72 100644 --- a/actioncable/lib/action_cable/remote_connections.rb +++ b/actioncable/lib/action_cable/remote_connections.rb @@ -1,6 +1,7 @@ module ActionCable - # If you need to disconnect a given connection, you go through the RemoteConnections. You find the connections you're looking for by - # searching the identifier declared on the connection. Example: + # If you need to disconnect a given connection, you can go through the + # RemoteConnections. You can find the connections you're looking for by + # searching for the identifier declared on the connection. For example: # # module ApplicationCable # class Connection < ActionCable::Connection::Base @@ -11,8 +12,9 @@ module ActionCable # # ActionCable.server.remote_connections.where(current_user: User.find(1)).disconnect # - # That will disconnect all the connections established for User.find(1) across all servers running on all machines (because it uses - # the internal channel that all these servers are subscribed to). + # This will disconnect all the connections established for + # <tt>User.find(1)</tt>, across all servers running on all machines, because + # it uses the internal channel that all of these servers are subscribed to. class RemoteConnections attr_reader :server @@ -25,7 +27,7 @@ module ActionCable end private - # Represents a single remote connection found via ActionCable.server.remote_connections.where(*). + # Represents a single remote connection found via <tt>ActionCable.server.remote_connections.where(*)</tt>. # Exists for the solely for the purpose of calling #disconnect on that connection. class RemoteConnection class InvalidIdentifiersError < StandardError; end @@ -39,7 +41,7 @@ module ActionCable # Uses the internal channel to disconnect the connection. def disconnect - server.broadcast internal_redis_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 a2a89d5f1e..bd6a3826a3 100644 --- a/actioncable/lib/action_cable/server.rb +++ b/actioncable/lib/action_cable/server.rb @@ -1,7 +1,3 @@ -require 'eventmachine' -EventMachine.epoll if EventMachine.epoll? -EventMachine.kqueue if EventMachine.kqueue? - module ActionCable module Server extend ActiveSupport::Autoload diff --git a/actioncable/lib/action_cable/server/base.rb b/actioncable/lib/action_cable/server/base.rb index 740e4b301e..c3b64299e3 100644 --- a/actioncable/lib/action_cable/server/base.rb +++ b/actioncable/lib/action_cable/server/base.rb @@ -1,14 +1,11 @@ -# FIXME: Cargo culted fix from https://github.com/celluloid/celluloid-pool/issues/10 -require 'celluloid/current' - -require 'em-hiredis' +require 'thread' module ActionCable module Server - # A singleton ActionCable::Server instance is available via ActionCable.server. It's used by the rack process that starts the cable server, but - # also by the user to reach the RemoteConnections instead for finding and disconnecting connections across all servers. + # A singleton ActionCable::Server instance is available via ActionCable.server. It's used by the Rack process that starts the Action Cable server, but + # is also used by the user to reach the RemoteConnections object, which is used for finding and disconnecting connections across all servers. # - # Also, this is the server instance used for broadcasting. See Broadcasting for details. + # Also, this is the server instance used for broadcasting. See Broadcasting for more information. class Base include ActionCable::Server::Broadcasting include ActionCable::Server::Connections @@ -18,10 +15,14 @@ module ActionCable def self.logger; config.logger; end delegate :logger, to: :config + attr_reader :mutex + def initialize + @mutex = Mutex.new + @remote_connections = @stream_event_loop = @worker_pool = @channel_classes = @pubsub = nil end - # Called by rack to setup the server. + # Called by Rack to setup the server. def call(env) setup_heartbeat_timer config.connection_class.new(self, env).process @@ -34,39 +35,34 @@ module ActionCable # Gateway to RemoteConnections. See that class for details. def remote_connections - @remote_connections ||= RemoteConnections.new(self) + @remote_connections || @mutex.synchronize { @remote_connections ||= RemoteConnections.new(self) } + end + + def stream_event_loop + @stream_event_loop || @mutex.synchronize { @stream_event_loop ||= ActionCable::Connection::StreamEventLoop.new } end # 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 || @mutex.synchronize { @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. + # Requires and returns a hash of all of the channel class constants, which are keyed by name. def channel_classes - @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 } + @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 - # The redis pubsub adapter used for all streams/broadcasting. + # Adapter used for all streams/broadcasting. def pubsub - @pubsub ||= redis.pubsub - end - - # The EventMachine Redis instance used by the pubsub adapter. - def redis - @redis ||= EM::Hiredis.connect(config.redis[:url]).tap do |redis| - redis.on(:reconnect_failed) do - logger.info "[ActionCable] Redis reconnect failed." - # logger.info "[ActionCable] Redis reconnected. Closing all the open connections." - # @connections.map &:close - end - end + @pubsub || @mutex.synchronize { @pubsub ||= config.pubsub_adapter.new(self) } end - # All the identifiers applied to the connection class associated with this server. + # All of the identifiers applied to the connection class associated with this server. def connection_identifiers config.connection_class.identifiers end diff --git a/actioncable/lib/action_cable/server/broadcasting.rb b/actioncable/lib/action_cable/server/broadcasting.rb index c759239a0e..98025f27f2 100644 --- a/actioncable/lib/action_cable/server/broadcasting.rb +++ b/actioncable/lib/action_cable/server/broadcasting.rb @@ -1,39 +1,32 @@ -require 'redis' - module ActionCable module Server - # Broadcasting is how other parts of your application can send messages to the channel subscribers. As explained in Channel, most of the time, these + # Broadcasting is how other parts of your application can send messages to a channel's subscribers. As explained in Channel, most of the time, these # broadcastings are streamed directly to the clients subscribed to the named broadcasting. Let's explain with a full-stack example: # # class WebNotificationsChannel < ApplicationCable::Channel - # def subscribed - # stream_from "web_notifications_#{current_user.id}" - # end - # end + # def subscribed + # stream_from "web_notifications_#{current_user.id}" + # end + # end # - # # Somewhere in your app this is called, perhaps from a NewCommentJob - # ActionCable.server.broadcast \ - # "web_notifications_1", { title: 'New things!', body: 'All shit fit for print' } + # # Somewhere in your app this is called, perhaps from a NewCommentJob: + # ActionCable.server.broadcast \ + # "web_notifications_1", { title: "New things!", body: "All that's fit for print" } # - # # Client-side coffescript, which assumes you've already requested the right to send web notifications - # App.cable.subscriptions.create "WebNotificationsChannel", - # received: (data) -> - # new Notification data['title'], body: data['body'] + # # Client-side CoffeeScript, which assumes you've already requested the right to send web notifications: + # App.cable.subscriptions.create "WebNotificationsChannel", + # received: (data) -> + # new Notification data['title'], body: data['body'] module Broadcasting - # Broadcast a hash directly to a named <tt>broadcasting</tt>. It'll automatically be JSON encoded. + # Broadcast a hash directly to a named <tt>broadcasting</tt>. This will later be JSON encoded. def broadcast(broadcasting, message) broadcaster_for(broadcasting).broadcast(message) end - # Returns a broadcaster for a named <tt>broadcasting</tt> that can be reused. Useful when you have a object that + # Returns a broadcaster for a named <tt>broadcasting</tt> that can be reused. Useful when you have an object that # may need multiple spots to transmit to a specific broadcasting over and over. def broadcaster_for(broadcasting) - Broadcaster.new(self, broadcasting) - end - - # The redis instance used for broadcasting. Not intended for direct user use. - def broadcasting_redis - @broadcasting_redis ||= Redis.new(config.redis) + Broadcaster.new(self, String(broadcasting)) end private @@ -46,7 +39,7 @@ module ActionCable def broadcast(message) server.logger.info "[ActionCable] Broadcasting to #{broadcasting}: #{message}" - server.broadcasting_redis.publish broadcasting, ActiveSupport::JSON.encode(message) + server.pubsub.broadcast broadcasting, ActiveSupport::JSON.encode(message) end end end diff --git a/actioncable/lib/action_cable/server/configuration.rb b/actioncable/lib/action_cable/server/configuration.rb index 935133cbba..9a7301287c 100644 --- a/actioncable/lib/action_cable/server/configuration.rb +++ b/actioncable/lib/action_cable/server/configuration.rb @@ -1,35 +1,48 @@ module ActionCable module Server - # An instance of this configuration object is available via ActionCable.server.config, which allows you to tweak the configuration points + # An instance of this configuration object is available via ActionCable.server.config, which allows you to tweak Action Cable configuration # in a Rails config initializer. class Configuration attr_accessor :logger, :log_tags attr_accessor :connection_class, :worker_pool_size - attr_accessor :redis, :channels_path attr_accessor :disable_request_forgery_protection, :allowed_request_origins - attr_accessor :url + attr_accessor :cable, :url, :mount_path + + attr_accessor :channel_paths # :nodoc: def initialize @log_tags = [] - @connection_class = ApplicationCable::Connection - @worker_pool_size = 100 - - @channels_path = Rails.root.join('app/channels') + @connection_class = ActionCable::Connection::Base + @worker_pool_size = 100 @disable_request_forgery_protection = false end - def channel_paths - @channels ||= Dir["#{channels_path}/**/*_channel.rb"] - 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' }) + path_to_adapter = "action_cable/subscription_adapter/#{adapter}" + begin + require path_to_adapter + rescue Gem::LoadError => e + raise Gem::LoadError, "Specified '#{adapter}' for Action Cable pubsub adapter, but the gem is not loaded. Add `gem '#{e.name}'` to your Gemfile (and ensure its version is at the minimum required by Action Cable)." + rescue LoadError => e + raise LoadError, "Could not load '#{path_to_adapter}'. Make sure that the adapter in config/cable.yml is valid. If you use an adapter other than 'postgresql' or 'redis' add the necessary adapter gem to the Gemfile.", e.backtrace + end + + adapter = adapter.camelize + adapter = 'PostgreSQL' if adapter == 'Postgresql' + "ActionCable::SubscriptionAdapter::#{adapter}".constantize + end end end end - diff --git a/actioncable/lib/action_cable/server/connections.rb b/actioncable/lib/action_cable/server/connections.rb index 47dcea8c20..4dc8934b25 100644 --- a/actioncable/lib/action_cable/server/connections.rb +++ b/actioncable/lib/action_cable/server/connections.rb @@ -1,9 +1,8 @@ module ActionCable module Server - # Collection class for all the connections that's been established on this specific server. Remember, usually you'll run many cable servers, so - # you can't use this collection as an full list of all the connections established against your application. Use RemoteConnections for that. - # As such, this is primarily for internal use. - module Connections + # Collection class for all the connections that have been established on this specific server. Remember, usually you'll run many Action Cable servers, so + # you can't use this collection as a full list of all of the connections established against your application. Instead, use RemoteConnections for that. + module Connections # :nodoc: BEAT_INTERVAL = 3 def connections @@ -19,14 +18,12 @@ module ActionCable end # WebSocket connection implementations differ on when they'll mark a connection as stale. We basically never want a connection to go stale, as you - # then can't rely on being able to receive and send to it. So there's a 3 second heartbeat running on all connections. If the beat fails, we automatically + # then can't rely on being able to communicate with the connection. To solve this, a 3 second heartbeat runs on all connections. If the beat fails, we automatically # disconnect. def setup_heartbeat_timer - EM.next_tick do - @heartbeat_timer ||= EventMachine.add_periodic_timer(BEAT_INTERVAL) do - EM.next_tick { connections.map(&:beat) } - end - end + @heartbeat_timer ||= Concurrent::TimerTask.new(execution_interval: BEAT_INTERVAL) do + Concurrent.global_io_executor.post { connections.map(&:beat) } + end.tap(&:execute) end def open_connections_statistics diff --git a/actioncable/lib/action_cable/server/worker.rb b/actioncable/lib/action_cable/server/worker.rb index e063b2a2e1..b920b880db 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 + # Worker used by Server.send_async to do connection work in threads. + class Worker # :nodoc: 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 diff --git a/actioncable/lib/action_cable/server/worker/active_record_connection_management.rb b/actioncable/lib/action_cable/server/worker/active_record_connection_management.rb index ecece4e270..1ac8934410 100644 --- a/actioncable/lib/action_cable/server/worker/active_record_connection_management.rb +++ b/actioncable/lib/action_cable/server/worker/active_record_connection_management.rb @@ -1,7 +1,7 @@ module ActionCable module Server class Worker - # Clear active connections between units of work so the long-running channel or connection processes do not hoard connections. + # Clear active connections between units of work so that way long-running channels or connection processes do not hoard connections. module ActiveRecordConnectionManagement extend ActiveSupport::Concern @@ -19,4 +19,4 @@ module ActionCable end end end -end
\ No newline at end of file +end diff --git a/actioncable/lib/action_cable/subscription_adapter.rb b/actioncable/lib/action_cable/subscription_adapter.rb new file mode 100644 index 0000000000..72e62f3daf --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter.rb @@ -0,0 +1,8 @@ +module ActionCable + module SubscriptionAdapter + extend ActiveSupport::Autoload + + autoload :Base + autoload :SubscriberMap + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter/async.rb b/actioncable/lib/action_cable/subscription_adapter/async.rb new file mode 100644 index 0000000000..cca6894289 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/async.rb @@ -0,0 +1,22 @@ +require 'action_cable/subscription_adapter/inline' + +module ActionCable + module SubscriptionAdapter + class Async < Inline # :nodoc: + private + def new_subscriber_map + AsyncSubscriberMap.new + end + + class AsyncSubscriberMap < SubscriberMap + def add_subscriber(*) + Concurrent.global_io_executor.post { super } + end + + def invoke_callback(*) + Concurrent.global_io_executor.post { super } + end + end + end + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter/base.rb b/actioncable/lib/action_cable/subscription_adapter/base.rb new file mode 100644 index 0000000000..796db5ffa3 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/base.rb @@ -0,0 +1,28 @@ +module ActionCable + module SubscriptionAdapter + class Base + attr_reader :logger, :server + + def initialize(server) + @server = server + @logger = @server.logger + end + + def broadcast(channel, payload) + raise NotImplementedError + end + + def subscribe(channel, message_callback, success_callback = nil) + raise NotImplementedError + end + + def unsubscribe(channel, message_callback) + raise NotImplementedError + end + + def shutdown + raise NotImplementedError + end + end + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb b/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb new file mode 100644 index 0000000000..e6c862959b --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb @@ -0,0 +1,75 @@ +require 'thread' + +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? + +module ActionCable + module SubscriptionAdapter + class EventedRedis < Base # :nodoc: + @@mutex = Mutex.new + + # Overwrite this factory method for EventMachine Redis connections if you want to use a different Redis connection library than EM::Hiredis. + # This is needed, for example, when using Makara proxies for distributed Redis. + cattr_accessor(:em_redis_connector) { ->(config) { EM::Hiredis.connect(config[:url]) } } + + # Overwrite this factory method for Redis connections if you want to use a different Redis connection library than Redis. + # This is needed, for example, when using Makara proxies for distributed Redis. + cattr_accessor(:redis_connector) { ->(config) { ::Redis.new(url: config[:url]) } } + + def initialize(*) + super + @redis_connection_for_broadcasts = @redis_connection_for_subscriptions = nil + end + + def broadcast(channel, payload) + redis_connection_for_broadcasts.publish(channel, payload) + end + + def subscribe(channel, message_callback, success_callback = nil) + redis_connection_for_subscriptions.pubsub.subscribe(channel, &message_callback).tap do |result| + result.callback { |reply| success_callback.call } if success_callback + end + end + + def unsubscribe(channel, message_callback) + redis_connection_for_subscriptions.pubsub.unsubscribe_proc(channel, message_callback) + end + + def shutdown + redis_connection_for_subscriptions.pubsub.close_connection + @redis_connection_for_subscriptions = nil + end + + private + def redis_connection_for_subscriptions + ensure_reactor_running + @redis_connection_for_subscriptions || @server.mutex.synchronize do + @redis_connection_for_subscriptions ||= self.class.em_redis_connector.call(@server.config.cable).tap do |redis| + redis.on(:reconnect_failed) do + @logger.info "[ActionCable] Redis reconnect failed." + end + end + end + end + + def redis_connection_for_broadcasts + @redis_connection_for_broadcasts || @server.mutex.synchronize do + @redis_connection_for_broadcasts ||= self.class.redis_connector.call(@server.config.cable) + end + end + + 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 + end + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter/inline.rb b/actioncable/lib/action_cable/subscription_adapter/inline.rb new file mode 100644 index 0000000000..81357faead --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/inline.rb @@ -0,0 +1,35 @@ +module ActionCable + module SubscriptionAdapter + class Inline < Base # :nodoc: + def initialize(*) + super + @subscriber_map = nil + end + + def broadcast(channel, payload) + subscriber_map.broadcast(channel, payload) + end + + def subscribe(channel, callback, success_callback = nil) + subscriber_map.add_subscriber(channel, callback, success_callback) + end + + def unsubscribe(channel, callback) + subscriber_map.remove_subscriber(channel, callback) + end + + def shutdown + # nothing to do + end + + private + def subscriber_map + @subscriber_map || @server.mutex.synchronize { @subscriber_map ||= new_subscriber_map } + end + + def new_subscriber_map + SubscriberMap.new + end + end + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb new file mode 100644 index 0000000000..abaeb92e54 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb @@ -0,0 +1,106 @@ +gem 'pg', '~> 0.18' +require 'pg' +require 'thread' + +module ActionCable + module SubscriptionAdapter + class PostgreSQL < Base # :nodoc: + def initialize(*) + super + @listener = nil + end + + def broadcast(channel, payload) + with_connection do |pg_conn| + pg_conn.exec("NOTIFY #{pg_conn.escape_identifier(channel)}, '#{pg_conn.escape_string(payload)}'") + end + end + + def subscribe(channel, callback, success_callback = nil) + listener.add_subscriber(channel, callback, success_callback) + end + + def unsubscribe(channel, callback) + listener.remove_subscriber(channel, callback) + end + + def shutdown + listener.shutdown + end + + def with_connection(&block) # :nodoc: + ActiveRecord::Base.connection_pool.with_connection do |ar_conn| + 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' + end + + yield pg_conn + end + end + + private + def listener + @listener || @server.mutex.synchronize { @listener ||= Listener.new(self) } + end + + class Listener < SubscriberMap + def initialize(adapter) + super() + + @adapter = adapter + @queue = Queue.new + + @thread = Thread.new do + Thread.current.abort_on_exception = true + listen + end + end + + def listen + @adapter.with_connection do |pg_conn| + catch :shutdown do + loop do + until @queue.empty? + action, channel, callback = @queue.pop(true) + + case action + when :listen + pg_conn.exec("LISTEN #{pg_conn.escape_identifier channel}") + Concurrent.global_io_executor << callback if callback + when :unlisten + pg_conn.exec("UNLISTEN #{pg_conn.escape_identifier channel}") + when :shutdown + throw :shutdown + end + end + + pg_conn.wait_for_notify(1) do |chan, pid, message| + broadcast(chan, message) + end + end + end + end + end + + def shutdown + @queue.push([:shutdown]) + Thread.pass while @thread.alive? + end + + def add_channel(channel, on_success) + @queue.push([:listen, channel, on_success]) + end + + def remove_channel(channel) + @queue.push([:unlisten, channel]) + end + + def invoke_callback(*) + Concurrent.global_io_executor.post { super } + end + end + end + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter/redis.rb b/actioncable/lib/action_cable/subscription_adapter/redis.rb new file mode 100644 index 0000000000..6b4236e7d3 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb @@ -0,0 +1,171 @@ +require 'thread' + +gem 'redis', '~> 3.0' +require 'redis' + +module ActionCable + module SubscriptionAdapter + class Redis < Base # :nodoc: + # Overwrite this factory method for redis connections if you want to use a different Redis library than Redis. + # This is needed, for example, when using Makara proxies for distributed Redis. + cattr_accessor(:redis_connector) { ->(config) { ::Redis.new(url: config[:url]) } } + + def initialize(*) + super + @listener = nil + @redis_connection_for_broadcasts = nil + end + + def broadcast(channel, payload) + redis_connection_for_broadcasts.publish(channel, payload) + end + + def subscribe(channel, callback, success_callback = nil) + listener.add_subscriber(channel, callback, success_callback) + end + + def unsubscribe(channel, callback) + listener.remove_subscriber(channel, callback) + end + + def shutdown + @listener.shutdown if @listener + end + + def redis_connection_for_subscriptions + redis_connection + end + + private + def listener + @listener || @server.mutex.synchronize { @listener ||= Listener.new(self) } + end + + def redis_connection_for_broadcasts + @redis_connection_for_broadcasts || @server.mutex.synchronize do + @redis_connection_for_broadcasts ||= redis_connection + end + end + + def redis_connection + self.class.redis_connector.call(@server.config.cable) + end + + class Listener < SubscriberMap + def initialize(adapter) + super() + + @adapter = adapter + + @subscribe_callbacks = Hash.new { |h, k| h[k] = [] } + @subscription_lock = Mutex.new + + @raw_client = nil + + @when_connected = [] + + @thread = nil + end + + def listen(conn) + conn.without_reconnect do + original_client = conn.client + + conn.subscribe('_action_cable_internal') do |on| + on.subscribe do |chan, count| + @subscription_lock.synchronize do + if count == 1 + @raw_client = original_client + + until @when_connected.empty? + @when_connected.shift.call + end + end + + if callbacks = @subscribe_callbacks[chan] + next_callback = callbacks.shift + Concurrent.global_io_executor << next_callback if next_callback + @subscribe_callbacks.delete(chan) if callbacks.empty? + end + end + end + + on.message do |chan, message| + broadcast(chan, message) + end + + on.unsubscribe do |chan, count| + if count == 0 + @subscription_lock.synchronize do + @raw_client = nil + end + end + end + end + end + end + + def shutdown + @subscription_lock.synchronize do + return if @thread.nil? + + when_connected do + send_command('unsubscribe') + @raw_client = nil + end + end + + Thread.pass while @thread.alive? + end + + def add_channel(channel, on_success) + @subscription_lock.synchronize do + ensure_listener_running + @subscribe_callbacks[channel] << on_success + when_connected { send_command('subscribe', channel) } + end + end + + def remove_channel(channel) + @subscription_lock.synchronize do + when_connected { send_command('unsubscribe', channel) } + end + end + + def invoke_callback(*) + Concurrent.global_io_executor.post { super } + end + + private + def ensure_listener_running + @thread ||= Thread.new do + Thread.current.abort_on_exception = true + + conn = @adapter.redis_connection_for_subscriptions + listen conn + end + end + + def when_connected(&block) + if @raw_client + block.call + else + @when_connected << block + end + end + + def send_command(*command) + @raw_client.write(command) + + very_raw_connection = + @raw_client.connection.instance_variable_defined?(:@connection) && + @raw_client.connection.instance_variable_get(:@connection) + + if very_raw_connection && very_raw_connection.respond_to?(:flush) + very_raw_connection.flush + end + end + end + end + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb b/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb new file mode 100644 index 0000000000..37eed09793 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb @@ -0,0 +1,53 @@ +module ActionCable + module SubscriptionAdapter + class SubscriberMap + def initialize + @subscribers = Hash.new { |h,k| h[k] = [] } + @sync = Mutex.new + end + + def add_subscriber(channel, subscriber, on_success) + @sync.synchronize do + new_channel = !@subscribers.key?(channel) + + @subscribers[channel] << subscriber + + if new_channel + add_channel channel, on_success + elsif on_success + on_success.call + end + end + end + + def remove_subscriber(channel, subscriber) + @sync.synchronize do + @subscribers[channel].delete(subscriber) + + if @subscribers[channel].empty? + @subscribers.delete channel + remove_channel channel + end + end + end + + def broadcast(channel, message) + list = @sync.synchronize { @subscribers[channel].dup } + list.each do |subscriber| + invoke_callback(subscriber, message) + end + end + + def add_channel(channel, on_success) + on_success.call if on_success + end + + def remove_channel(channel) + end + + def invoke_callback(callback, message) + callback.call message + end + end + end +end diff --git a/actioncable/lib/assets/javascripts/action_cable.coffee.erb b/actioncable/lib/assets/javascripts/action_cable.coffee.erb deleted file mode 100644 index 7daea4ebcd..0000000000 --- a/actioncable/lib/assets/javascripts/action_cable.coffee.erb +++ /dev/null @@ -1,23 +0,0 @@ -#= require_self -#= require action_cable/consumer - -@ActionCable = - INTERNAL: <%= ActionCable::INTERNAL.to_json %> - - createConsumer: (url = @getConfig("url")) -> - new ActionCable.Consumer @createWebSocketURL(url) - - getConfig: (name) -> - element = document.head.querySelector("meta[name='action-cable-#{name}']") - element?.getAttribute("content") - - createWebSocketURL: (url) -> - if url and not /^wss?:/i.test(url) - a = document.createElement("a") - a.href = url - # Fix populating Location properties in IE. Otherwise, protocol will be blank. - a.href = a.href - a.protocol = a.protocol.replace("http", "ws") - a.href - else - url diff --git a/actioncable/lib/assets/javascripts/action_cable/connection.coffee b/actioncable/lib/assets/javascripts/action_cable/connection.coffee deleted file mode 100644 index 2f69a9b26c..0000000000 --- a/actioncable/lib/assets/javascripts/action_cable/connection.coffee +++ /dev/null @@ -1,84 +0,0 @@ -# Encapsulate the cable connection held by the consumer. This is an internal class not intended for direct user manipulation. - -{message_types} = ActionCable.INTERNAL - -class ActionCable.Connection - @reopenDelay: 500 - - constructor: (@consumer) -> - @open() - - send: (data) -> - if @isOpen() - @webSocket.send(JSON.stringify(data)) - true - else - false - - open: => - if @webSocket and not @isState("closed") - throw new Error("Existing connection must be closed before opening") - else - @webSocket = new WebSocket(@consumer.url) - @installEventHandlers() - true - - close: -> - @webSocket?.close() - - reopen: -> - if @isState("closed") - @open() - else - try - @close() - finally - setTimeout(@open, @constructor.reopenDelay) - - isOpen: -> - @isState("open") - - # Private - - isState: (states...) -> - @getState() in states - - getState: -> - return state.toLowerCase() for state, value of WebSocket when value is @webSocket?.readyState - null - - installEventHandlers: -> - for eventName of @events - handler = @events[eventName].bind(this) - @webSocket["on#{eventName}"] = handler - return - - events: - message: (event) -> - {identifier, message, type} = JSON.parse(event.data) - - switch type - when message_types.confirmation - @consumer.subscriptions.notify(identifier, "connected") - when message_types.rejection - @consumer.subscriptions.reject(identifier) - else - @consumer.subscriptions.notify(identifier, "received", message) - - open: -> - @disconnected = false - @consumer.subscriptions.reload() - - close: -> - @disconnect() - - error: -> - @disconnect() - - disconnect: -> - return if @disconnected - @disconnected = true - @consumer.subscriptions.notifyAll("disconnected") - - toJSON: -> - state: @getState() diff --git a/actioncable/lib/assets/javascripts/action_cable/connection_monitor.coffee b/actioncable/lib/assets/javascripts/action_cable/connection_monitor.coffee deleted file mode 100644 index b594802be1..0000000000 --- a/actioncable/lib/assets/javascripts/action_cable/connection_monitor.coffee +++ /dev/null @@ -1,84 +0,0 @@ -# Responsible for ensuring the cable connection is in good health by validating the heartbeat pings sent from the server, and attempting -# revival reconnections if things go astray. Internal class, not intended for direct user manipulation. -class ActionCable.ConnectionMonitor - @pollInterval: - min: 3 - max: 30 - - @staleThreshold: 6 # Server::Connections::BEAT_INTERVAL * 2 (missed two pings) - - identifier: ActionCable.INTERNAL.identifiers.ping - - constructor: (@consumer) -> - @consumer.subscriptions.add(this) - @start() - - connected: -> - @reset() - @pingedAt = now() - delete @disconnectedAt - - disconnected: -> - @disconnectedAt = now() - - received: -> - @pingedAt = now() - - reset: -> - @reconnectAttempts = 0 - - start: -> - @reset() - delete @stoppedAt - @startedAt = now() - @poll() - document.addEventListener("visibilitychange", @visibilityDidChange) - - stop: -> - @stoppedAt = now() - document.removeEventListener("visibilitychange", @visibilityDidChange) - - poll: -> - setTimeout => - unless @stoppedAt - @reconnectIfStale() - @poll() - , @getInterval() - - getInterval: -> - {min, max} = @constructor.pollInterval - interval = 5 * Math.log(@reconnectAttempts + 1) - clamp(interval, min, max) * 1000 - - reconnectIfStale: -> - if @connectionIsStale() - @reconnectAttempts++ - unless @disconnectedRecently() - @consumer.connection.reopen() - - connectionIsStale: -> - secondsSince(@pingedAt ? @startedAt) > @constructor.staleThreshold - - disconnectedRecently: -> - @disconnectedAt and secondsSince(@disconnectedAt) < @constructor.staleThreshold - - visibilityDidChange: => - if document.visibilityState is "visible" - setTimeout => - if @connectionIsStale() or not @consumer.connection.isOpen() - @consumer.connection.reopen() - , 200 - - toJSON: -> - interval = @getInterval() - connectionIsStale = @connectionIsStale() - {@startedAt, @stoppedAt, @pingedAt, @reconnectAttempts, connectionIsStale, interval} - - now = -> - new Date().getTime() - - secondsSince = (time) -> - (now() - time) / 1000 - - clamp = (number, min, max) -> - Math.max(min, Math.min(max, number)) diff --git a/actioncable/lib/assets/javascripts/action_cable/consumer.coffee b/actioncable/lib/assets/javascripts/action_cable/consumer.coffee deleted file mode 100644 index 5cf8978d77..0000000000 --- a/actioncable/lib/assets/javascripts/action_cable/consumer.coffee +++ /dev/null @@ -1,31 +0,0 @@ -#= require action_cable/connection -#= require action_cable/connection_monitor -#= require action_cable/subscriptions -#= require action_cable/subscription - -# The ActionCable.Consumer establishes the connection to a server-side Ruby Connection object. Once established, -# the ActionCable.ConnectionMonitor will ensure that its properly maintained through heartbeats and checking for stale updates. -# The Consumer instance is also the gateway to establishing subscriptions to desired channels through the #createSubscription -# method. -# -# The following example shows how this can be setup: -# -# @App = {} -# App.cable = ActionCable.createConsumer "ws://example.com/accounts/1" -# App.appearance = App.cable.subscriptions.create "AppearanceChannel" -# -# For more details on how you'd configure an actual channel subscription, see ActionCable.Subscription. -class ActionCable.Consumer - constructor: (@url) -> - @subscriptions = new ActionCable.Subscriptions this - @connection = new ActionCable.Connection this - @connectionMonitor = new ActionCable.ConnectionMonitor this - - send: (data) -> - @connection.send(data) - - inspect: -> - JSON.stringify(this, null, 2) - - toJSON: -> - {@url, @subscriptions, @connection, @connectionMonitor} diff --git a/actioncable/lib/assets/javascripts/action_cable/subscription.coffee b/actioncable/lib/assets/javascripts/action_cable/subscription.coffee deleted file mode 100644 index 339d676933..0000000000 --- a/actioncable/lib/assets/javascripts/action_cable/subscription.coffee +++ /dev/null @@ -1,68 +0,0 @@ -# A new subscription is created through the ActionCable.Subscriptions instance available on the consumer. -# It provides a number of callbacks and a method for calling remote procedure calls on the corresponding -# Channel instance on the server side. -# -# An example demonstrates the basic functionality: -# -# App.appearance = App.cable.subscriptions.create "AppearanceChannel", -# connected: -> -# # Called once the subscription has been successfully completed -# -# appear: -> -# @perform 'appear', appearing_on: @appearingOn() -# -# away: -> -# @perform 'away' -# -# appearingOn: -> -# $('main').data 'appearing-on' -# -# The methods #appear and #away forward their intent to the remote AppearanceChannel instance on the server -# by calling the `@perform` method with the first parameter being the action (which maps to AppearanceChannel#appear/away). -# The second parameter is a hash that'll get JSON encoded and made available on the server in the data parameter. -# -# This is how the server component would look: -# -# class AppearanceChannel < ApplicationActionCable::Channel -# def subscribed -# current_user.appear -# end -# -# def unsubscribed -# current_user.disappear -# end -# -# def appear(data) -# current_user.appear on: data['appearing_on'] -# end -# -# def away -# current_user.away -# end -# end -# -# The "AppearanceChannel" name is automatically mapped between the client-side subscription creation and the server-side Ruby class name. -# The AppearanceChannel#appear/away public methods are exposed automatically to client-side invocation through the @perform method. -class ActionCable.Subscription - constructor: (@subscriptions, params = {}, mixin) -> - @identifier = JSON.stringify(params) - extend(this, mixin) - @subscriptions.add(this) - @consumer = @subscriptions.consumer - - # Perform a channel action with the optional data passed as an attribute - perform: (action, data = {}) -> - data.action = action - @send(data) - - send: (data) -> - @consumer.send(command: "message", identifier: @identifier, data: JSON.stringify(data)) - - unsubscribe: -> - @subscriptions.remove(this) - - extend = (object, properties) -> - if properties? - for key, value of properties - object[key] = value - object diff --git a/actioncable/lib/assets/javascripts/action_cable/subscriptions.coffee b/actioncable/lib/assets/javascripts/action_cable/subscriptions.coffee deleted file mode 100644 index 0316f76a24..0000000000 --- a/actioncable/lib/assets/javascripts/action_cable/subscriptions.coffee +++ /dev/null @@ -1,78 +0,0 @@ -# Collection class for creating (and internally managing) channel subscriptions. The only method intended to be triggered by the user -# us ActionCable.Subscriptions#create, and it should be called through the consumer like so: -# -# @App = {} -# App.cable = ActionCable.createConsumer "ws://example.com/accounts/1" -# App.appearance = App.cable.subscriptions.create "AppearanceChannel" -# -# For more details on how you'd configure an actual channel subscription, see ActionCable.Subscription. -class ActionCable.Subscriptions - constructor: (@consumer) -> - @subscriptions = [] - @history = [] - - create: (channelName, mixin) -> - channel = channelName - params = if typeof channel is "object" then channel else {channel} - new ActionCable.Subscription this, params, mixin - - # Private - - add: (subscription) -> - @subscriptions.push(subscription) - @notify(subscription, "initialized") - @sendCommand(subscription, "subscribe") - - remove: (subscription) -> - @forget(subscription) - - unless @findAll(subscription.identifier).length - @sendCommand(subscription, "unsubscribe") - - reject: (identifier) -> - for subscription in @findAll(identifier) - @forget(subscription) - @notify(subscription, "rejected") - - forget: (subscription) -> - @subscriptions = (s for s in @subscriptions when s isnt subscription) - - findAll: (identifier) -> - s for s in @subscriptions when s.identifier is identifier - - reload: -> - for subscription in @subscriptions - @sendCommand(subscription, "subscribe") - - notifyAll: (callbackName, args...) -> - for subscription in @subscriptions - @notify(subscription, callbackName, args...) - - notify: (subscription, callbackName, args...) -> - if typeof subscription is "string" - subscriptions = @findAll(subscription) - else - subscriptions = [subscription] - - for subscription in subscriptions - subscription[callbackName]?(args...) - - if callbackName in ["initialized", "connected", "disconnected", "rejected"] - {identifier} = subscription - @record(notification: {identifier, callbackName, args}) - - sendCommand: (subscription, command) -> - {identifier} = subscription - if identifier is ActionCable.INTERNAL.identifiers.ping - @consumer.connection.isOpen() - else - @consumer.send({command, identifier}) - - record: (data) -> - data.time = new Date() - @history = @history.slice(-19) - @history.push(data) - - toJSON: -> - history: @history - identifiers: (subscription.identifier for subscription in @subscriptions) diff --git a/actioncable/lib/rails/generators/channel/channel_generator.rb b/actioncable/lib/rails/generators/channel/channel_generator.rb index 2f37d8055b..6debe40c91 100644 --- a/actioncable/lib/rails/generators/channel/channel_generator.rb +++ b/actioncable/lib/rails/generators/channel/channel_generator.rb @@ -5,17 +5,39 @@ module Rails argument :actions, type: :array, default: [], banner: "method method" + class_option :assets, type: :boolean + check_class_collision suffix: "Channel" def create_channel_file template "channel.rb", File.join('app/channels', class_path, "#{file_name}_channel.rb") - template "assets/channel.coffee", File.join('app/assets/javascripts/channels', class_path, "#{file_name}.coffee") + + if options[:assets] + template "assets/channel.coffee", File.join('app/assets/javascripts/channels', class_path, "#{file_name}.coffee") + end + + generate_application_cable_files end protected def file_name @_file_name ||= super.gsub(/\_channel/i, '') end + + # FIXME: Change these files to symlinks once RubyGems 2.5.0 is required. + def generate_application_cable_files + return if self.behavior != :invoke + + files = [ + 'application_cable/channel.rb', + 'application_cable/connection.rb' + ] + + files.each do |name| + path = File.join('app/channels/', name) + template(name, path) if !File.exist?(path) + end + end end end end diff --git a/actioncable/lib/rails/generators/channel/templates/application_cable/channel.rb b/actioncable/lib/rails/generators/channel/templates/application_cable/channel.rb new file mode 100644 index 0000000000..d56fa30f4d --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/application_cable/channel.rb @@ -0,0 +1,5 @@ +# Be sure to restart your server when you modify this file. Action Cable runs in a loop that does not support auto reloading. +module ApplicationCable + class Channel < ActionCable::Channel::Base + end +end diff --git a/actioncable/lib/rails/generators/channel/templates/application_cable/connection.rb b/actioncable/lib/rails/generators/channel/templates/application_cable/connection.rb new file mode 100644 index 0000000000..b4f41389ad --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/application_cable/connection.rb @@ -0,0 +1,5 @@ +# Be sure to restart your server when you modify this file. Action Cable runs in a loop that does not support auto reloading. +module ApplicationCable + class Connection < ActionCable::Connection::Base + end +end diff --git a/actioncable/lib/rails/generators/channel/templates/channel.rb b/actioncable/lib/rails/generators/channel/templates/channel.rb index 6cf04ee61f..7bff3341c1 100644 --- a/actioncable/lib/rails/generators/channel/templates/channel.rb +++ b/actioncable/lib/rails/generators/channel/templates/channel.rb @@ -1,4 +1,4 @@ -# Be sure to restart your server when you modify this file. Action Cable runs in an EventMachine loop that does not support auto reloading. +# Be sure to restart your server when you modify this file. Action Cable runs in a loop that does not support auto reloading. <% module_namespacing do -%> class <%= class_name %>Channel < ApplicationCable::Channel def subscribed |