diff options
Diffstat (limited to 'actioncable/lib')
44 files changed, 2474 insertions, 0 deletions
diff --git a/actioncable/lib/action_cable.rb b/actioncable/lib/action_cable.rb new file mode 100644 index 0000000000..1dc66ef3ad --- /dev/null +++ b/actioncable/lib/action_cable.rb @@ -0,0 +1,51 @@ +#-- +# 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 +# "Software"), to deal in the Software without restriction, including +# without limitation the rights to use, copy, modify, merge, publish, +# distribute, sublicense, and/or sell copies of the Software, and to +# permit persons to whom the Software is furnished to do so, subject to +# the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +#++ + +require 'active_support' +require 'active_support/rails' +require 'action_cable/version' + +module ActionCable + extend ActiveSupport::Autoload + + INTERNAL = { + identifiers: { + ping: '_ping'.freeze + }, + message_types: { + confirmation: 'confirm_subscription'.freeze, + rejection: 'reject_subscription'.freeze + } + } + + # Singleton instance of the server + module_function def server + @server ||= ActionCable::Server::Base.new + end + + autoload :Server + autoload :Connection + autoload :Channel + autoload :RemoteConnections + autoload :SubscriptionAdapter +end diff --git a/actioncable/lib/action_cable/channel.rb b/actioncable/lib/action_cable/channel.rb new file mode 100644 index 0000000000..7ae262ce5f --- /dev/null +++ b/actioncable/lib/action_cable/channel.rb @@ -0,0 +1,14 @@ +module ActionCable + module Channel + extend ActiveSupport::Autoload + + eager_autoload do + autoload :Base + autoload :Broadcasting + autoload :Callbacks + autoload :Naming + autoload :PeriodicTimers + autoload :Streams + end + end +end diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb new file mode 100644 index 0000000000..874ebe2e71 --- /dev/null +++ b/actioncable/lib/action_cable/channel/base.rb @@ -0,0 +1,286 @@ +require 'set' + +module ActionCable + module Channel + # The channel provides the basic structure of grouping behavior into logical units when communicating over the WebSocket connection. + # You can think of a channel like a form of controller, but one that's capable of pushing content to the subscriber in addition to simply + # responding to the subscriber's direct requests. + # + # Channel instances are long-lived. A channel object will be instantiated when the cable consumer becomes a subscriber, and then + # lives until the consumer disconnects. This may be seconds, minutes, hours, or even days. That means you have to take special care + # not to do anything silly in a channel that would balloon its memory footprint or whatever. The references are forever, so they won't be released + # as is normally the case with a controller instance that gets thrown away after every request. + # + # Long-lived channels (and connections) also mean you're responsible for ensuring that the data is fresh. If you hold a reference to a user + # record, but the name is changed while that reference is held, you may be sending stale data if you don't take precautions to avoid it. + # + # The upside of long-lived channel instances is that you can use instance variables to keep reference to objects that future subscriber requests + # can interact with. Here's a quick example: + # + # class ChatChannel < ApplicationCable::Channel + # def subscribed + # @room = Chat::Room[params[:room_number]] + # end + # + # def speak(data) + # @room.speak data, user: current_user + # end + # end + # + # The #speak action simply uses the Chat::Room object that was created when the channel was first subscribed to by the consumer when that + # subscriber wants to say something in the room. + # + # == Action processing + # + # Unlike subclasses of ActionController::Base, channels do not follow a REST + # constraint form for their actions. Instead, ActionCable 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: + # + # class AppearanceChannel < ApplicationCable::Channel + # def subscribed + # @connection_token = generate_connection_token + # end + # + # def unsubscribed + # current_user.disappear @connection_token + # end + # + # def appear(data) + # current_user.appear @connection_token, on: data['appearing_on'] + # end + # + # def away + # current_user.away @connection_token + # end + # + # private + # def generate_connection_token + # SecureRandom.hex(36) + # end + # end + # + # In this example, subscribed/unsubscribed 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 as 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, <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 the #reject method: + # + # class ChatChannel < ApplicationCable::Channel + # def subscribed + # @room = Chat::Room[params[:room_number]] + # reject unless current_user.can_access?(@room) + # end + # end + # + # 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 + include Streams + include Naming + include Broadcasting + + attr_reader :params, :connection, :identifier + delegate :logger, to: :connection + + class << self + # A list of method names that should be considered actions. This + # includes all public instance methods on a channel, less + # any internal methods (defined on Base), adding back in + # any methods that are internal, but still exist on the class + # itself. + # + # ==== Returns + # * <tt>Set</tt> - A set of all methods that should be considered actions. + def action_methods + @action_methods ||= begin + # All public instance methods of this class, including ancestors + methods = (public_instance_methods(true) - + # Except for public instance methods of Base and its ancestors + ActionCable::Channel::Base.public_instance_methods(true) + + # Be sure to include shadowed public instance methods of this class + public_instance_methods(false)).uniq.map(&:to_s) + methods.to_set + end + end + + 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 + def clear_action_methods! + @action_methods = nil + end + + # Refresh the cached action_methods when a new action_method is added. + def method_added(name) + super + clear_action_methods! + end + end + + def initialize(connection, identifier, params = {}) + @connection = connection + @identifier = identifier + @params = params + + # 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 + @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 + # that the action requested is a public method on the channel declared by the user (so not one of the callbacks + # like #subscribed). + def perform_action(data) + action = extract_action(data) + + if processable_action?(action) + dispatch_action(action, data) + else + logger.error "Unable to process #{action_signature(action, data)}" + end + end + + # Called by the cable connection when its 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 + run_callbacks :unsubscribe do + unsubscribed + 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. + def subscribed + # Override in subclasses + 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. + def unsubscribed + # Override in subclasses + end + + # 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 } + connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data) + end + + def defer_subscription_confirmation! + @defer_subscription_confirmation = true + end + + def defer_subscription_confirmation? + @defer_subscription_confirmation + end + + def subscription_confirmation_sent? + @subscription_confirmation_sent + end + + def reject + @reject_subscription = true + end + + def subscription_rejected? + @reject_subscription + end + + private + def delegate_connection_identifiers + connection.identifiers.each do |identifier| + define_singleton_method(identifier) do + connection.send(identifier) + end + 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 + end + + def processable_action?(action) + self.class.action_methods.include?(action.to_s) + end + + def dispatch_action(action, data) + logger.info action_signature(action, data) + + if method(action).arity == 1 + public_send action, data + else + public_send action + end + end + + def action_signature(action, data) + "#{self.class.name}##{action}".tap do |signature| + if (arguments = data.except('action')).any? + signature << "(#{arguments.inspect})" + end + end + end + + def transmit_subscription_confirmation + unless subscription_confirmation_sent? + logger.info "#{self.class.name} is transmitting the subscription confirmation" + connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:confirmation]) + @subscription_confirmation_sent = true + end + end + + def reject_subscription + connection.subscriptions.remove_subscription self + transmit_subscription_rejection + end + + def transmit_subscription_rejection + logger.info "#{self.class.name} is transmitting the subscription rejection" + connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:rejection]) + end + end + end +end diff --git a/actioncable/lib/action_cable/channel/broadcasting.rb b/actioncable/lib/action_cable/channel/broadcasting.rb new file mode 100644 index 0000000000..afc23d7d1a --- /dev/null +++ b/actioncable/lib/action_cable/channel/broadcasting.rb @@ -0,0 +1,29 @@ +require 'active_support/core_ext/object/to_param' + +module ActionCable + module Channel + module Broadcasting + extend ActiveSupport::Concern + + delegate :broadcasting_for, to: :class + + class_methods do + # Broadcast a hash to a unique broadcasting for this <tt>model</tt> in this channel. + def broadcast_to(model, message) + ActionCable.server.broadcast(broadcasting_for([ channel_name, model ]), message) + end + + def broadcasting_for(model) #:nodoc: + case + when model.is_a?(Array) + model.map { |m| broadcasting_for(m) }.join(':') + when model.respond_to?(:to_gid_param) + model.to_gid_param + else + model.to_param + end + end + end + end + end +end diff --git a/actioncable/lib/action_cable/channel/callbacks.rb b/actioncable/lib/action_cable/channel/callbacks.rb new file mode 100644 index 0000000000..295d750e86 --- /dev/null +++ b/actioncable/lib/action_cable/channel/callbacks.rb @@ -0,0 +1,35 @@ +require 'active_support/callbacks' + +module ActionCable + module Channel + module Callbacks + extend ActiveSupport::Concern + include ActiveSupport::Callbacks + + included do + define_callbacks :subscribe + define_callbacks :unsubscribe + end + + class_methods do + def before_subscribe(*methods, &block) + set_callback(:subscribe, :before, *methods, &block) + end + + def after_subscribe(*methods, &block) + set_callback(:subscribe, :after, *methods, &block) + end + alias_method :on_subscribe, :after_subscribe + + def before_unsubscribe(*methods, &block) + set_callback(:unsubscribe, :before, *methods, &block) + end + + def after_unsubscribe(*methods, &block) + set_callback(:unsubscribe, :after, *methods, &block) + end + alias_method :on_unsubscribe, :after_unsubscribe + end + end + end +end diff --git a/actioncable/lib/action_cable/channel/naming.rb b/actioncable/lib/action_cable/channel/naming.rb new file mode 100644 index 0000000000..4c9d53b15a --- /dev/null +++ b/actioncable/lib/action_cable/channel/naming.rb @@ -0,0 +1,22 @@ +module ActionCable + module Channel + module Naming + extend ActiveSupport::Concern + + class_methods do + # Returns the name of the channel, underscored, without the <tt>Channel</tt> ending. + # If the channel is in a namespace, then the namespaces are represented by single + # colon separators in the channel name. + # + # ChatChannel.channel_name # => 'chat' + # Chats::AppearancesChannel.channel_name # => 'chats:appearances' + def channel_name + @channel_name ||= name.sub(/Channel$/, '').gsub('::',':').underscore + end + end + + # Delegates to the class' <tt>channel_name</tt> + delegate :channel_name, to: :class + end + end +end diff --git a/actioncable/lib/action_cable/channel/periodic_timers.rb b/actioncable/lib/action_cable/channel/periodic_timers.rb new file mode 100644 index 0000000000..56597d02d7 --- /dev/null +++ b/actioncable/lib/action_cable/channel/periodic_timers.rb @@ -0,0 +1,41 @@ +module ActionCable + module Channel + module PeriodicTimers + extend ActiveSupport::Concern + + included do + class_attribute :periodic_timers, instance_reader: false + self.periodic_timers = [] + + after_subscribe :start_periodic_timers + after_unsubscribe :stop_periodic_timers + end + + module ClassMethods + # Allow 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:) + self.periodic_timers += [ [ callback, every: every ] ] + end + end + + private + def active_periodic_timers + @active_periodic_timers ||= [] + end + + def start_periodic_timers + self.class.periodic_timers.each do |callback, options| + 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.shutdown } + end + end + end +end diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb new file mode 100644 index 0000000000..3158f30814 --- /dev/null +++ b/actioncable/lib/action_cable/channel/streams.rb @@ -0,0 +1,116 @@ +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. + # + # 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 + # comments on a given page: + # + # class CommentsChannel < ApplicationCable::Channel + # def follow(data) + # stream_from "comments_for_#{data['recording_id']}" + # end + # + # def unfollow + # stop_all_streams + # 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: + # + # 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` + # + # class CommentsChannel < ApplicationCable::Channel + # def subscribed + # post = Post.find(params[:id]) + # stream_for post + # end + # end + # + # You can then broadcast to this channel using: + # + # 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: + # + # class ChatChannel < ApplicationCable::Channel + # def subscribed + # @room = Chat::Room[params[:room_number]] + # + # 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) + # + # 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 + # end + # + # You can stop streaming from all broadcasts by calling #stop_all_streams. + module Streams + extend ActiveSupport::Concern + + included do + on_unsubscribe :stop_all_streams + end + + # 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 + defer_subscription_confirmation! + + callback ||= default_stream_callback(broadcasting) + streams << [ broadcasting, callback ] + + 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 + + # Start streaming the pubsub queue for the <tt>model</tt> in this channel. 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_for(model, callback = nil) + 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 broadcasting, callback + logger.info "#{self.class.name} stopped streaming from #{broadcasting}" + end.clear + end + + private + delegate :pubsub, to: :connection + + def streams + @_streams ||= [] + end + + def default_stream_callback(broadcasting) + -> (message) do + transmit ActiveSupport::JSON.decode(message), via: "streamed from #{broadcasting}" + end + end + end + end +end diff --git a/actioncable/lib/action_cable/connection.rb b/actioncable/lib/action_cable/connection.rb new file mode 100644 index 0000000000..902efb07e2 --- /dev/null +++ b/actioncable/lib/action_cable/connection.rb @@ -0,0 +1,19 @@ +module ActionCable + module Connection + extend ActiveSupport::Autoload + + eager_autoload do + autoload :Authorization + autoload :Base + autoload :ClientSocket + autoload :Identification + autoload :InternalChannel + autoload :MessageBuffer + autoload :Stream + autoload :StreamEventLoop + autoload :Subscriptions + autoload :TaggedLoggerProxy + autoload :WebSocket + end + end +end diff --git a/actioncable/lib/action_cable/connection/authorization.rb b/actioncable/lib/action_cable/connection/authorization.rb new file mode 100644 index 0000000000..070a70e4e2 --- /dev/null +++ b/actioncable/lib/action_cable/connection/authorization.rb @@ -0,0 +1,13 @@ +module ActionCable + module Connection + module Authorization + class UnauthorizedError < StandardError; end + + private + def reject_unauthorized_connection + logger.error "An unauthorized connection attempt was rejected" + raise UnauthorizedError + end + end + end +end
\ No newline at end of file diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb new file mode 100644 index 0000000000..b5f898436a --- /dev/null +++ b/actioncable/lib/action_cable/connection/base.rb @@ -0,0 +1,222 @@ +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 + # authentication and authorization. + # + # Here's a basic example: + # + # module ApplicationCable + # class Connection < ActionCable::Connection::Base + # identified_by :current_user + # + # def connect + # self.current_user = find_verified_user + # logger.add_tags current_user.name + # end + # + # def disconnect + # # Any cleanup work needed when the cable connection is cut. + # end + # + # protected + # def find_verified_user + # if current_user = User.find_by_identity cookies.signed[:identity_id] + # current_user + # else + # reject_unauthorized_connection + # end + # end + # 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. + # + # 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. + # + # Finally, we add a tag to the connection-specific logger with name of the current user to easily distinguish their messages in the log. + # + # Pretty simple, eh? + class Base + include Identification + include InternalChannel + include Authorization + + 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, self, stream_event_loop) + @subscriptions = ActionCable::Connection::Subscriptions.new(self) + @message_buffer = ActionCable::Connection::MessageBuffer.new(self) + + @_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 + logger.info started_request_message + + if websocket.possible? && allow_request_origin? + 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. + # The data is routed to the proper channel that the connection has subscribed to. + def receive(data_in_json) + if websocket.alive? + subscriptions.execute_command ActiveSupport::JSON.decode(data_in_json) + else + logger.error "Received data without a live WebSocket (#{data_in_json.inspect})" + end + end + + # 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) + websocket.transmit data + end + + # Close the WebSocket connection. + def close + websocket.close + end + + # 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) + end + + # Return a basic hash of statistics for the connection keyed with `identifier`, `started_at`, and `subscriptions`. + # 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'] + } + end + + def beat + 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. + def request + @request ||= begin + environment = Rails.application.env_config.merge(env) if defined?(Rails.application) && Rails.application + ActionDispatch::Request.new(environment || env) + end + end + + # The cookies of the request that initiated the WebSocket connection. Useful for performing authorization checks. + def cookies + request.cookie_jar + end + + attr_reader :websocket + attr_reader :message_buffer + + private + def handle_open + connect if respond_to?(:connect) + subscribe_to_internal_channel + beat + + message_buffer.process! + server.add_connection(self) + rescue ActionCable::Connection::Authorization::UnauthorizedError + respond_to_invalid_request + end + + def handle_close + logger.info finished_request_message + + server.remove_connection(self) + + subscriptions.unsubscribe_from_all + unsubscribe_from_internal_channel + + disconnect if respond_to?(:disconnect) + end + + 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'] } + true + else + logger.error("Request origin not allowed: #{env['HTTP_ORIGIN']}") + false + end + end + + def respond_to_successful_request + websocket.rack_response + end + + def respond_to_invalid_request + close if websocket.alive? + + 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, + tags: server.config.log_tags.map { |tag| tag.respond_to?(:call) ? tag.call(request) : tag.to_s.camelize } + end + + def started_request_message + 'Started %s "%s"%s for %s at %s' % [ + request.request_method, + request.filtered_path, + websocket.possible? ? ' [WebSocket]' : '', + request.ip, + Time.now.to_s ] + end + + def finished_request_message + 'Finished "%s"%s for %s at %s' % [ + request.filtered_path, + websocket.possible? ? ' [WebSocket]' : '', + request.ip, + Time.now.to_s ] + 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..ef937d7c16 --- /dev/null +++ b/actioncable/lib/action_cable/connection/client_socket.rb @@ -0,0 +1,150 @@ +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) + + if callback = @env['async.callback'] + callback.call([101, {}, @stream]) + end + end + + def start_driver + return if @driver.nil? || @driver_started + @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] + + if @stream + @stream.shutdown + else + finalize_close + end + 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 new file mode 100644 index 0000000000..885ff3f102 --- /dev/null +++ b/actioncable/lib/action_cable/connection/identification.rb @@ -0,0 +1,46 @@ +require 'set' + +module ActionCable + module Connection + module Identification + extend ActiveSupport::Concern + + included do + class_attribute :identifiers + self.identifiers = Set.new + end + + class_methods do + # 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. + def identified_by(*identifiers) + Array(identifiers).each { |identifier| attr_accessor identifier } + self.identifiers += identifiers + end + end + + # Return a single connection identifier that combines the value of all the registered identifiers into a single gid. + def connection_identifier + unless defined? @connection_identifier + @connection_identifier = connection_gid identifiers.map { |id| instance_variable_get("@#{id}") }.compact + end + + @connection_identifier + end + + private + def connection_gid(ids) + ids.map do |o| + if o.respond_to? :to_gid_param + o.to_gid_param + else + o.to_s + end + end.sort.join(":") + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/internal_channel.rb b/actioncable/lib/action_cable/connection/internal_channel.rb new file mode 100644 index 0000000000..27826792b3 --- /dev/null +++ b/actioncable/lib/action_cable/connection/internal_channel.rb @@ -0,0 +1,45 @@ +module ActionCable + module Connection + # Makes it possible for the RemoteConnection to disconnect a specific connection. + module InternalChannel + extend ActiveSupport::Concern + + private + def internal_channel + "action_cable/#{connection_identifier}" + end + + def subscribe_to_internal_channel + if connection_identifier.present? + callback = -> (message) { process_internal_message(message) } + @_internal_subscriptions ||= [] + @_internal_subscriptions << [ internal_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_subscriptions.present? + @_internal_subscriptions.each { |channel, callback| Concurrent.global_io_executor.post { pubsub.unsubscribe(channel, callback) } } + end + end + + def process_internal_message(message) + message = ActiveSupport::JSON.decode(message) + + case message['type'] + when 'disconnect' + logger.info "Removing connection (#{connection_identifier})" + websocket.close + end + rescue Exception => e + logger.error "There was an exception - #{e.class}(#{e.message})" + logger.error e.backtrace.join("\n") + + close + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/message_buffer.rb b/actioncable/lib/action_cable/connection/message_buffer.rb new file mode 100644 index 0000000000..2f65a1e84a --- /dev/null +++ b/actioncable/lib/action_cable/connection/message_buffer.rb @@ -0,0 +1,54 @@ +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 + def initialize(connection) + @connection = connection + @buffered_messages = [] + end + + def append(message) + if valid? message + if processing? + receive message + else + buffer message + end + else + connection.logger.error "Couldn't handle non-string message: #{message.class}" + end + end + + def processing? + @processing + end + + def process! + @processing = true + receive_buffered_messages + end + + protected + attr_reader :connection + attr_accessor :buffered_messages + + private + def valid?(message) + message.is_a?(String) + end + + def receive(message) + connection.send_async :receive, message + end + + def buffer(message) + buffered_messages << message + end + + def receive_buffered_messages + receive buffered_messages.shift until buffered_messages.empty? + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/stream.rb b/actioncable/lib/action_cable/connection/stream.rb new file mode 100644 index 0000000000..ace250cd16 --- /dev/null +++ b/actioncable/lib/action_cable/connection/stream.rb @@ -0,0 +1,59 @@ +module ActionCable + module Connection + #-- + # This class is heavily based on faye-websocket-ruby + # + # Copyright (c) 2010-2015 James Coglan + class Stream + def initialize(event_loop, socket) + @event_loop = event_loop + @socket_object = socket + @stream_send = socket.env['stream.send'] + + @rack_hijack_io = nil + + hijack_rack_socket + 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 + + private + 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 + + 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 new file mode 100644 index 0000000000..d7f95e6a62 --- /dev/null +++ b/actioncable/lib/action_cable/connection/subscriptions.rb @@ -0,0 +1,75 @@ +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 + def initialize(connection) + @connection = connection + @subscriptions = {} + end + + def execute_command(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 + rescue Exception => e + logger.error "Could not execute command from #{data.inspect}) [#{e.class} - #{e.message}]: #{e.backtrace.first(5).join(" | ")}" + end + + def add(data) + id_key = data['identifier'] + id_options = ActiveSupport::JSON.decode(id_key).with_indifferent_access + + subscription_klass = connection.server.channel_classes[id_options[:channel]] + + if subscription_klass + subscriptions[id_key] ||= subscription_klass.new(connection, id_key, id_options) + else + logger.error "Subscription class not found (#{data.inspect})" + end + end + + def remove(data) + logger.info "Unsubscribing from channel: #{data['identifier']}" + remove_subscription subscriptions[data['identifier']] + end + + def remove_subscription(subscription) + subscription.unsubscribe_from_channel + subscriptions.delete(subscription.identifier) + end + + def perform_action(data) + 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 } + end + + protected + attr_reader :connection, :subscriptions + + private + delegate :logger, to: :connection + + def find(data) + if subscription = subscriptions[data['identifier']] + subscription + else + raise "Unable to find subscription with identifier: #{data['identifier']}" + end + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/tagged_logger_proxy.rb b/actioncable/lib/action_cable/connection/tagged_logger_proxy.rb new file mode 100644 index 0000000000..41afa9680a --- /dev/null +++ b/actioncable/lib/action_cable/connection/tagged_logger_proxy.rb @@ -0,0 +1,40 @@ +module ActionCable + module Connection + # Allows the use of per-connection tags against the server logger. This wouldn't work using the traditional + # <tt>ActiveSupport::TaggedLogging</tt> enhanced Rails.logger, as that logger will reset the tags between requests. + # The connection is long-lived, so it needs its own set of tags for its independent duration. + class TaggedLoggerProxy + attr_reader :tags + + def initialize(logger, tags:) + @logger = logger + @tags = tags.flatten + end + + def add_tags(*tags) + @tags += tags.flatten + @tags = @tags.uniq + end + + def tag(logger) + if logger.respond_to?(:tagged) + current_tags = tags - logger.formatter.current_tags + logger.tagged(*current_tags) { yield } + else + yield + end + end + + %i( debug info warn error fatal unknown ).each do |severity| + define_method(severity) do |message| + log severity, message + end + end + + protected + def log(type, message) + tag(@logger) { @logger.send type, message } + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/web_socket.rb b/actioncable/lib/action_cable/connection/web_socket.rb new file mode 100644 index 0000000000..5e89fb9b72 --- /dev/null +++ b/actioncable/lib/action_cable/connection/web_socket.rb @@ -0,0 +1,35 @@ +require 'websocket/driver' + +module ActionCable + module Connection + # Wrap the real socket to minimize the externally-presented API + class WebSocket + 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? + websocket + end + + def alive? + websocket && websocket.alive? + end + + def transmit(data) + websocket.transmit data + end + + def close + websocket.close + end + + def rack_response + websocket.rack_response + end + + protected + attr_reader :websocket + end + end +end diff --git a/actioncable/lib/action_cable/engine.rb b/actioncable/lib/action_cable/engine.rb new file mode 100644 index 0000000000..f5e233e091 --- /dev/null +++ b/actioncable/lib/action_cable/engine.rb @@ -0,0 +1,38 @@ +require "rails" +require "action_cable" +require "action_cable/helpers/action_cable_helper" +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.eager_load_namespaces << ActionCable + + initializer "action_cable.helpers" do + ActiveSupport.on_load(:action_view) do + include ActionCable::Helpers::ActionCableHelper + end + end + + initializer "action_cable.logger" do + ActiveSupport.on_load(:action_cable) { self.logger ||= ::Rails.logger } + end + + initializer "action_cable.set_configs" do |app| + options = app.config.action_cable + options.allowed_request_origins ||= "http://localhost:3000" if ::Rails.env.development? + + app.paths.add "config/cable", with: "config/cable.yml" + + ActiveSupport.on_load(:action_cable) do + if (config_path = Pathname.new(app.config.paths["config/cable"].first)).exist? + self.cable = Rails.application.config_for(config_path).with_indifferent_access + end + + options.each { |k,v| send("#{k}=", v) } + end + end + end +end diff --git a/actioncable/lib/action_cable/gem_version.rb b/actioncable/lib/action_cable/gem_version.rb new file mode 100644 index 0000000000..a71603e61a --- /dev/null +++ b/actioncable/lib/action_cable/gem_version.rb @@ -0,0 +1,15 @@ +module ActionCable + # Returns the version of the currently loaded Action Cable as a <tt>Gem::Version</tt>. + def self.gem_version + Gem::Version.new VERSION::STRING + end + + module VERSION + MAJOR = 5 + MINOR = 0 + TINY = 0 + PRE = "beta2" + + STRING = [MAJOR, MINOR, TINY, PRE].compact.join(".") + end +end diff --git a/actioncable/lib/action_cable/helpers/action_cable_helper.rb b/actioncable/lib/action_cable/helpers/action_cable_helper.rb new file mode 100644 index 0000000000..b82751468a --- /dev/null +++ b/actioncable/lib/action_cable/helpers/action_cable_helper.rb @@ -0,0 +1,29 @@ +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: + # + # <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. + # Your CoffeeScript can then connect to the server without needing to specify the + # url directly: + # + # #= require cable + # @App = {} + # App.cable = Cable.createConsumer() + # + # Make sure to specify the correct server location in each of your environments + # config file: + # + # 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 + end + end + end +end diff --git a/actioncable/lib/action_cable/remote_connections.rb b/actioncable/lib/action_cable/remote_connections.rb new file mode 100644 index 0000000000..7ec121308a --- /dev/null +++ b/actioncable/lib/action_cable/remote_connections.rb @@ -0,0 +1,66 @@ +module ActionCable + # 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 + # identified_by :current_user + # .... + # end + # end + # + # ActionCable.server.remote_connections.where(current_user: User.find(1)).disconnect + # + # 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 these servers are subscribed to. + class RemoteConnections + attr_reader :server + + def initialize(server) + @server = server + end + + def where(identifier) + RemoteConnection.new(server, identifier) + end + + private + # 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 + + include Connection::Identification, Connection::InternalChannel + + def initialize(server, ids) + @server = server + set_identifier_instance_vars(ids) + end + + # Uses the internal channel to disconnect the connection. + def disconnect + server.broadcast internal_channel, type: 'disconnect' + end + + # Returns all the identifiers that were applied to this connection. + def identifiers + server.connection_identifiers + end + + private + attr_reader :server + + def set_identifier_instance_vars(ids) + raise InvalidIdentifiersError unless valid_identifiers?(ids) + ids.each { |k,v| instance_variable_set("@#{k}", v) } + end + + def valid_identifiers?(ids) + keys = ids.keys + identifiers.all? { |id| keys.include?(id) } + end + end + end +end diff --git a/actioncable/lib/action_cable/server.rb b/actioncable/lib/action_cable/server.rb new file mode 100644 index 0000000000..bd6a3826a3 --- /dev/null +++ b/actioncable/lib/action_cable/server.rb @@ -0,0 +1,15 @@ +module ActionCable + module Server + extend ActiveSupport::Autoload + + eager_autoload do + autoload :Base + autoload :Broadcasting + autoload :Connections + autoload :Configuration + + autoload :Worker + 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 new file mode 100644 index 0000000000..fe48c112df --- /dev/null +++ b/actioncable/lib/action_cable/server/base.rb @@ -0,0 +1,74 @@ +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. + # + # Also, this is the server instance used for broadcasting. See Broadcasting for details. + class Base + include ActionCable::Server::Broadcasting + include ActionCable::Server::Connections + + cattr_accessor(:config, instance_accessor: true) { ActionCable::Server::Configuration.new } + + 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. + def call(env) + setup_heartbeat_timer + config.connection_class.new(self, env).process + end + + # Disconnect all the connections identified by `identifiers` on this server or any others via RemoteConnections. + def disconnect(identifiers) + remote_connections.where(identifiers).disconnect + end + + # Gateway to RemoteConnections. See that class for details. + def remote_connections + @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 || @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. + 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) } + end + + # All the identifiers applied to the connection class associated with this server. + def connection_identifiers + config.connection_class.identifiers + end + end + + ActiveSupport.run_load_hooks(:action_cable, Base.config) + end +end diff --git a/actioncable/lib/action_cable/server/broadcasting.rb b/actioncable/lib/action_cable/server/broadcasting.rb new file mode 100644 index 0000000000..7e8aef45f4 --- /dev/null +++ b/actioncable/lib/action_cable/server/broadcasting.rb @@ -0,0 +1,47 @@ +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 + # 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 + # + # # 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 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. + 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 + # may need multiple spots to transmit to a specific broadcasting over and over. + def broadcaster_for(broadcasting) + Broadcaster.new(self, broadcasting) + end + + private + class Broadcaster + attr_reader :server, :broadcasting + + def initialize(server, broadcasting) + @server, @broadcasting = server, broadcasting + end + + def broadcast(message) + server.logger.info "[ActionCable] Broadcasting to #{broadcasting}: #{message}" + server.pubsub.broadcast broadcasting, ActiveSupport::JSON.encode(message) + end + end + end + end +end diff --git a/actioncable/lib/action_cable/server/configuration.rb b/actioncable/lib/action_cable/server/configuration.rb new file mode 100644 index 0000000000..9a248933c4 --- /dev/null +++ b/actioncable/lib/action_cable/server/configuration.rb @@ -0,0 +1,55 @@ +module ActionCable + module Server + # An instance of this configuration object is available via ActionCable.server.config, which allows you to tweak the configuration points + # in a Rails config initializer. + class Configuration + attr_accessor :logger, :log_tags + attr_accessor :connection_class, :worker_pool_size + attr_accessor :channel_load_paths + attr_accessor :disable_request_forgery_protection, :allowed_request_origins + attr_accessor :cable, :url + + def initialize + @log_tags = [] + + @connection_class = ApplicationCable::Connection + @worker_pool_size = 100 + + @channel_load_paths = [Rails.root.join('app/channels')] + + @disable_request_forgery_protection = false + end + + def channel_paths + @channel_paths ||= channel_load_paths.flat_map do |path| + Dir["#{path}/**/*_channel.rb"] + end + 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 new file mode 100644 index 0000000000..8671dd5ebd --- /dev/null +++ b/actioncable/lib/action_cable/server/connections.rb @@ -0,0 +1,35 @@ +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 + BEAT_INTERVAL = 3 + + def connections + @connections ||= [] + end + + def add_connection(connection) + connections << connection + end + + def remove_connection(connection) + connections.delete connection + 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 + # disconnect. + def setup_heartbeat_timer + @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 + connections.map(&:statistics) + end + end + end +end diff --git a/actioncable/lib/action_cable/server/worker.rb b/actioncable/lib/action_cable/server/worker.rb new file mode 100644 index 0000000000..3b6c6d44a1 --- /dev/null +++ b/actioncable/lib/action_cable/server/worker.rb @@ -0,0 +1,71 @@ +require 'active_support/callbacks' +require 'active_support/core_ext/module/attribute_accessors_per_thread' +require 'concurrent' + +module ActionCable + module Server + # Worker used by Server.send_async to do connection work in threads. Only for internal use. + class Worker + include ActiveSupport::Callbacks + + 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) + begin + self.connection = receiver + + 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 + end + + def async_run_periodic_timer(channel, callback) + @pool.post do + run_periodic_timer(channel, callback) + end + end + + def run_periodic_timer(channel, callback) + begin + self.connection = channel.connection + + 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 + end + end +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 new file mode 100644 index 0000000000..ecece4e270 --- /dev/null +++ b/actioncable/lib/action_cable/server/worker/active_record_connection_management.rb @@ -0,0 +1,22 @@ +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. + module ActiveRecordConnectionManagement + extend ActiveSupport::Concern + + included do + if defined?(ActiveRecord::Base) + set_callback :work, :around, :with_database_connections + end + end + + def with_database_connections + connection.logger.tag(ActiveRecord::Base.logger) { yield } + ensure + ActiveRecord::Base.clear_active_connections! + end + end + end + end +end
\ No newline at end of file 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..d697548cbd --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb @@ -0,0 +1,67 @@ +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 + + 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 ||= EM::Hiredis.connect(@server.config.cable[:url]).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 ||= ::Redis.new(@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..7076383efe --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb @@ -0,0 +1,163 @@ +require 'thread' + +gem 'redis', '~> 3.0' +require 'redis' + +module ActionCable + module SubscriptionAdapter + class Redis < Base # :nodoc: + 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.new(@server.config.cable) + 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.new(@server.config.cable) + end + 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/action_cable/version.rb b/actioncable/lib/action_cable/version.rb new file mode 100644 index 0000000000..e17877202b --- /dev/null +++ b/actioncable/lib/action_cable/version.rb @@ -0,0 +1,8 @@ +require_relative 'gem_version' + +module ActionCable + # Returns the version of the currently loaded Action Cable as a <tt>Gem::Version</tt> + def self.version + gem_version + end +end diff --git a/actioncable/lib/rails/generators/channel/USAGE b/actioncable/lib/rails/generators/channel/USAGE new file mode 100644 index 0000000000..27a934c689 --- /dev/null +++ b/actioncable/lib/rails/generators/channel/USAGE @@ -0,0 +1,14 @@ +Description: +============ + Stubs out a new cable channel for the server (in Ruby) and client (in CoffeeScript). + Pass the channel name, either CamelCased or under_scored, and an optional list of channel actions as arguments. + + Note: Turn on the cable connection in app/assets/javascript/cable.coffee after generating any channels. + +Example: +======== + rails generate channel Chat speak + + creates a Chat channel class and CoffeeScript asset: + Channel: app/channels/chat_channel.rb + Assets: app/assets/javascript/channels/chat.coffee diff --git a/actioncable/lib/rails/generators/channel/channel_generator.rb b/actioncable/lib/rails/generators/channel/channel_generator.rb new file mode 100644 index 0000000000..c5d398810a --- /dev/null +++ b/actioncable/lib/rails/generators/channel/channel_generator.rb @@ -0,0 +1,26 @@ +module Rails + module Generators + class ChannelGenerator < NamedBase + source_root File.expand_path("../templates", __FILE__) + + 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") + + if options[:assets] + template "assets/channel.coffee", File.join('app/assets/javascripts/channels', class_path, "#{file_name}.coffee") + end + end + + protected + def file_name + @_file_name ||= super.gsub(/\_channel/i, '') + end + end + end +end diff --git a/actioncable/lib/rails/generators/channel/templates/assets/channel.coffee b/actioncable/lib/rails/generators/channel/templates/assets/channel.coffee new file mode 100644 index 0000000000..5467811aba --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/assets/channel.coffee @@ -0,0 +1,14 @@ +App.<%= class_name.underscore %> = App.cable.subscriptions.create "<%= class_name %>Channel", + connected: -> + # Called when the subscription is ready for use on the server + + disconnected: -> + # Called when the subscription has been terminated by the server + + received: (data) -> + # Called when there's incoming data on the websocket for this channel +<% actions.each do |action| -%> + + <%= action %>: -> + @perform '<%= action %>' +<% end -%> diff --git a/actioncable/lib/rails/generators/channel/templates/channel.rb b/actioncable/lib/rails/generators/channel/templates/channel.rb new file mode 100644 index 0000000000..6cf04ee61f --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/channel.rb @@ -0,0 +1,17 @@ +# Be sure to restart your server when you modify this file. Action Cable runs in an EventMachine loop that does not support auto reloading. +<% module_namespacing do -%> +class <%= class_name %>Channel < ApplicationCable::Channel + def subscribed + # stream_from "some_channel" + end + + def unsubscribed + # Any cleanup needed when channel is unsubscribed + end +<% actions.each do |action| -%> + + def <%= action %> + end +<% end -%> +end +<% end -%> |