diff options
Diffstat (limited to 'actioncable')
39 files changed, 479 insertions, 231 deletions
diff --git a/actioncable/CHANGELOG.md b/actioncable/CHANGELOG.md index e671a07563..946fdfb3fc 100644 --- a/actioncable/CHANGELOG.md +++ b/actioncable/CHANGELOG.md @@ -1,3 +1,22 @@ +* Safely support autoloading and class unloading, by preventing concurrent + loads, and disconnecting all cables during reload. + + *Matthew Draper* + +* Ensure ActionCable behaves correctly for non-string queue names. + + *Jay Hayes* + +## Rails 5.0.0.beta3 (February 24, 2016) ## + +* Added `em_redis_connector` and `redis_connector` to + `ActionCable::SubscriptionAdapter::EventedRedis` and added `redis_connector` + to `ActionCable::SubscriptionAdapter::Redis`, so you can overwrite with your + own initializers. This is used when you want to use different-than-standard + Redis adapters, like for Makara distributed Redis. + + *DHH* + ## Rails 5.0.0.beta2 (February 01, 2016) ## * Support PostgreSQL pubsub adapter. diff --git a/actioncable/README.md b/actioncable/README.md index ac57532b62..595830feb0 100644 --- a/actioncable/README.md +++ b/actioncable/README.md @@ -17,7 +17,7 @@ The client of a WebSocket connection is called the consumer. Each consumer can in turn subscribe to multiple cable channels. Each channel encapsulates a logical unit of work, similar to what a controller does in a regular MVC setup. For example, -you could have a `ChatChannel` and a `AppearancesChannel`, and a consumer could be subscribed to either +you could have a `ChatChannel` and an `AppearancesChannel`, and a consumer could be subscribed to either or to both of these channels. At the very least, a consumer should be subscribed to one channel. When the consumer is subscribed to a channel, they act as a subscriber. The connection between @@ -39,7 +39,7 @@ reflections of each unit. ### A full-stack example The first thing you must do is define your `ApplicationCable::Connection` class in Ruby. This -is the place where you authorize the incoming connection, and proceed to establish it +is the place where you authorize the incoming connection, and proceed to establish it, if all is well. Here's the simplest example starting with the server-side connection class: ```ruby @@ -73,7 +73,7 @@ use that to set the `current_user`. By identifying the connection by this same c you're also ensuring that you can later retrieve all open connections by a given user (and potentially disconnect them all if the user is deleted or deauthorized). -Then you should define your `ApplicationCable::Channel` class in Ruby. This is the place where you put +Next, you should define your `ApplicationCable::Channel` class in Ruby. This is the place where you put shared logic between your channels. ```ruby @@ -94,7 +94,7 @@ The client-side needs to setup a consumer instance of this connection. That's do App.cable = ActionCable.createConsumer("ws://cable.example.com") ``` -The ws://cable.example.com address must point to your set of Action Cable servers, and it +The `ws://cable.example.com` address must point to your Action Cable server(s), and it must share a cookie namespace with the rest of the application (which may live under http://example.com). This ensures that the signed cookie will be correctly sent. @@ -105,8 +105,8 @@ is defined by declaring channels on the server and allowing the consumer to subs ### Channel example 1: User appearances -Here's a simple example of a channel that tracks whether a user is online or not and what page they're on. -(This is useful for creating presence features like showing a green dot next to a user name if they're online). +Here's a simple example of a channel that tracks whether a user is online or not, and also what page they are currently on. +(This is useful for creating presence features like showing a green dot next to a user's name if they're online). First you declare the server-side channel: @@ -180,7 +180,7 @@ App.cable.subscriptions.create "AppearanceChannel", Simply calling `App.cable.subscriptions.create` will setup the subscription, which will call `AppearanceChannel#subscribed`, which in turn is linked to original `App.cable` -> `ApplicationCable::Connection` instances. -We then link the client-side `appear` method to `AppearanceChannel#appear(data)`. This is possible because the server-side +Next, we link the client-side `appear` method to `AppearanceChannel#appear(data)`. This is possible because the server-side channel instance will automatically expose the public methods declared on the class (minus the callbacks), so that these can be reached as remote procedure calls via a subscription's `perform` method. @@ -188,7 +188,7 @@ can be reached as remote procedure calls via a subscription's `perform` method. The appearance example was all about exposing server functionality to client-side invocation over the WebSocket connection. But the great thing about WebSockets is that it's a two-way street. So now let's show an example where the server invokes -action on the client. +an action on the client. This is a web notification channel that allows you to trigger client-side web notifications when you broadcast to the right streams: @@ -215,7 +215,7 @@ ActionCable.server.broadcast \ "web_notifications_#{current_user.id}", { title: 'New things!', body: 'All the news that is fit to print' } ``` -The `ActionCable.server.broadcast` call places a message in the Redis' pubsub queue under a separate broadcasting name for each user. For a user with an ID of 1, the broadcasting name would be `web_notifications_1`. +The `ActionCable.server.broadcast` call places a message in the Action Cable pubsub queue under a separate broadcasting name for each user. For a user with an ID of 1, the broadcasting name would be `web_notifications_1`. The channel has been instructed to stream everything that arrives at `web_notifications_1` directly to the client by invoking the `#received(data)` callback. The data is the hash sent as the second parameter to the server-side broadcast call, JSON encoded for the trip across the wire, and unpacked for the data argument arriving to `#received`. @@ -234,7 +234,7 @@ class ChatChannel < ApplicationCable::Channel end ``` -Pass an object as the first argument to `subscriptions.create`, and that object will become your params hash in your cable channel. The keyword `channel` is required. +If you pass an object as the first argument to `subscriptions.create`, that object will become the params hash in your cable channel. The keyword `channel` is required. ```coffeescript # Client-side, which assumes you've already requested the right to send web notifications @@ -293,27 +293,29 @@ The rebroadcast will be received by all connected clients, _including_ the clien ### More complete examples -See the [rails/actioncable-examples](http://github.com/rails/actioncable-examples) repository for a full example of how to setup Action Cable in a Rails app and adding channels. +See the [rails/actioncable-examples](http://github.com/rails/actioncable-examples) repository for a full example of how to setup Action Cable in a Rails app, and how to add channels. ## Configuration -Action Cable has three required configurations: the Redis connection, allowed request origins, and the cable server url (which can optionally be set on the client side). +Action Cable has three required configurations: a subscription adapter, allowed request origins, and the cable server URL (which can optionally be set on the client side). ### Redis By default, `ActionCable::Server::Base` will look for a configuration file in `Rails.root.join('config/cable.yml')`. -This file must specify a Redis url for each Rails environment. It may use the following format: +This file must specify an adapter and a URL for each Rails environment. It may use the following format: ```yaml production: &production + adapter: redis url: redis://10.10.3.153:6381 development: &development + adapter: redis url: redis://localhost:6379 test: *development ``` -You can also change the location of the Redis config file in a Rails initializer with something like: +You can also change the location of the Action Cable config file in a Rails initializer with something like: ```ruby Rails.application.paths.add "config/cable", with: "somewhere/else/cable.yml" @@ -324,7 +326,7 @@ Rails.application.paths.add "config/cable", with: "somewhere/else/cable.yml" Action Cable will only accept requests from specified origins, which are passed to the server config as an array. The origins can be instances of strings or regular expressions, against which a check for match will be performed. ```ruby -ActionCable.server.config.allowed_request_origins = ['http://rubyonrails.com', /http:\/\/ruby.*/] +Rails.application.config.action_cable.allowed_request_origins = ['http://rubyonrails.com', /http:\/\/ruby.*/] ``` When running in the development environment, this defaults to "http://localhost:3000". @@ -332,26 +334,26 @@ When running in the development environment, this defaults to "http://localhost: To disable and allow requests from any origin: ```ruby -ActionCable.server.config.disable_request_forgery_protection = true +Rails.application.config.action_cable.disable_request_forgery_protection = true ``` ### Consumer Configuration -Once you have decided how to run your cable server (see below), you must provide the server url (or path) to your client-side setup. +Once you have decided how to run your cable server (see below), you must provide the server URL (or path) to your client-side setup. There are two ways you can do this. The first is to simply pass it in when creating your consumer. For a standalone server, this would be something like: `App.cable = ActionCable.createConsumer("ws://example.com:28080")`, and for an in-app server, something like: `App.cable = ActionCable.createConsumer("/cable")`. -The second option is to pass the server url through the `action_cable_meta_tag` in your layout. -This uses a url or path typically set via `config.action_cable.url` in the environment configuration files, or defaults to "/cable". +The second option is to pass the server URL through the `action_cable_meta_tag` in your layout. +This uses a URL or path typically set via `config.action_cable.url` in the environment configuration files, or defaults to "/cable". -This method is especially useful if your websocket url might change between environments. If you host your production server via https, you will need to use the wss scheme -for your ActionCable server, but development might remain http and use the ws scheme. You might use localhost in development and your +This method is especially useful if your WebSocket URL might change between environments. If you host your production server via https, you will need to use the wss scheme +for your Action Cable server, but development might remain http and use the ws scheme. You might use localhost in development and your domain in production. -In any case, to vary the websocket url between environments, add the following configuration to each environment: +In any case, to vary the WebSocket URL between environments, add the following configuration to each environment: ```ruby config.action_cable.url = "ws://example.com:28080" @@ -374,7 +376,7 @@ App.cable = ActionCable.createConsumer() The other common option to configure is the log tags applied to the per-connection logger. Here's close to what we're using in Basecamp: ```ruby -ActionCable.server.config.log_tags = [ +Rails.application.config.action_cable.log_tags = [ -> request { request.env['bc.account_id'] || "no-account" }, :action_cable, -> request { request.uuid } @@ -389,7 +391,7 @@ Also note that your server must provide at least the same number of database con ## Running the cable server ### Standalone -The cable server(s) is separated from your normal application server. It's still a rack application, but it is its own rack +The cable server(s) is separated from your normal application server. It's still a Rack application, but it is its own Rack application. The recommended basic setup is as follows: ```ruby @@ -410,7 +412,7 @@ The above will start a cable server on port 28080. ### In app -If you are using a threaded server like Puma or Thin, the current implementation of ActionCable can run side-along with your Rails application. For example, to listen for WebSocket requests on `/cable`, mount the server at that path: +If you are using a server that supports the [Rack socket hijacking API](http://www.rubydoc.info/github/rack/rack/file/SPEC#Hijacking), Action Cable can run alongside your Rails application. For example, to listen for WebSocket requests on `/cable`, mount the server at that path: ```ruby # config/routes.rb @@ -419,7 +421,7 @@ Example::Application.routes.draw do end ``` -For every instance of your server you create and for every worker your server spawns, you will also have a new instance of ActionCable, but the use of Redis keeps messages synced across connections. +For every instance of your server you create and for every worker your server spawns, you will also have a new instance of Action Cable, but the use of Redis keeps messages synced across connections. ### Notes @@ -431,25 +433,27 @@ The WebSocket server doesn't have access to the session, but it has access to th ## Dependencies -Action Cable is currently tied to Redis through its use of the pubsub feature to route -messages back and forth over the WebSocket cable connection. This dependency may well -be alleviated in the future, but for the moment that's what it is. So be sure to have -Redis installed and running. +Action Cable provides a subscription adapter interface to process its pubsub internals. By default, asynchronous, inline, PostgreSQL, evented Redis, and non-evented Redis adapters are included. The default adapter in new Rails applications is the asynchronous (`async`) adapter. To create your own adapter, you can look at `ActionCable::SubscriptionAdapter::Base` for all methods that must be implemented, and any of the adapters included within Action Cable as example implementations. -The Ruby side of things is built on top of [faye-websocket](https://github.com/faye/faye-websocket-ruby) and [concurrent-ruby](https://github.com/ruby-concurrency/concurrent-ruby). +The Ruby side of things is built on top of [websocket-driver](https://github.com/faye/websocket-driver-ruby), [nio4r](https://github.com/celluloid/nio4r), and [concurrent-ruby](https://github.com/ruby-concurrency/concurrent-ruby). ## Deployment -Action Cable is powered by a combination of websockets and threads. All of the +Action Cable is powered by a combination of WebSockets and threads. All of the connection management is handled internally by utilizing Ruby’s native thread support, which means you can use all your regular Rails models with no problems as long as you haven’t committed any thread-safety sins. -But this also means that Action Cable needs to run in its own server process. -So you'll have one set of server processes for your normal web work, and another -set of server processes for the Action Cable. The former can be single-threaded, -like Unicorn, but the latter must be multi-threaded, like Puma. +The Action Cable server does _not_ need to be a multi-threaded application server. +This is because Action Cable uses the [Rack socket hijacking API](http://www.rubydoc.info/github/rack/rack/file/SPEC#Hijacking) +to take over control of connections from the application server. Action Cable +then manages connections internally, in a multithreaded manner, regardless of +whether the application server is multi-threaded or not. So Action Cable works +with all the popular application servers -- Unicorn, Puma and Passenger. + +Action Cable does not work with WEBrick, because WEBrick does not support the +Rack socket hijacking API. ## License diff --git a/actioncable/app/assets/javascripts/action_cable.coffee.erb b/actioncable/app/assets/javascripts/action_cable.coffee.erb index 18a48c0610..6a8b4eeb85 100644 --- a/actioncable/app/assets/javascripts/action_cable.coffee.erb +++ b/actioncable/app/assets/javascripts/action_cable.coffee.erb @@ -9,7 +9,7 @@ getConfig: (name) -> element = document.head.querySelector("meta[name='action-cable-#{name}']") - element?.getAttribute("content") + element?.getAttribute("content") ? '/cable' createWebSocketURL: (url) -> if url and not /^wss?:/i.test(url) @@ -21,3 +21,14 @@ a.href else url + + startDebugging: -> + @debugging = true + + stopDebugging: -> + @debugging = null + + log: (messages...) -> + if @debugging + messages.push(Date.now()) + console.log("[ActionCable]", messages...) diff --git a/actioncable/app/assets/javascripts/action_cable/connection.coffee b/actioncable/app/assets/javascripts/action_cable/connection.coffee index fbd7dbd35b..4244322a1e 100644 --- a/actioncable/app/assets/javascripts/action_cable/connection.coffee +++ b/actioncable/app/assets/javascripts/action_cable/connection.coffee @@ -6,9 +6,11 @@ class ActionCable.Connection @reopenDelay: 500 constructor: (@consumer) -> - @open() send: (data) -> + unless @isOpen() + @open() + if @isOpen() @webSocket.send(JSON.stringify(data)) true @@ -16,9 +18,12 @@ class ActionCable.Connection false open: => - if @webSocket and not @isState("closed") + if @isAlive() + ActionCable.log("Attemped to open WebSocket, but existing socket is #{@getState()}") throw new Error("Existing connection must be closed before opening") else + ActionCable.log("Opening WebSocket, current state is #{@getState()}") + @uninstallEventHandlers() if @webSocket? @webSocket = new WebSocket(@consumer.url) @installEventHandlers() true @@ -27,19 +32,26 @@ class ActionCable.Connection @webSocket?.close() reopen: -> - if @isState("closed") - @open() - else + ActionCable.log("Reopening WebSocket, current state is #{@getState()}") + if @isAlive() try @close() + catch error + ActionCable.log("Failed to reopen WebSocket", error) finally + ActionCable.log("Reopening WebSocket in #{@constructor.reopenDelay}ms") setTimeout(@open, @constructor.reopenDelay) + else + @open() isOpen: -> @isState("open") # Private + isAlive: -> + @webSocket? and not @isState("closing", "closed") + isState: (states...) -> @getState() in states @@ -53,6 +65,11 @@ class ActionCable.Connection @webSocket["on#{eventName}"] = handler return + uninstallEventHandlers: -> + for eventName of @events + @webSocket["on#{eventName}"] = -> + return + events: message: (event) -> {identifier, message, type} = JSON.parse(event.data) @@ -66,13 +83,16 @@ class ActionCable.Connection @consumer.subscriptions.notify(identifier, "received", message) open: -> + ActionCable.log("WebSocket onopen event") @disconnected = false @consumer.subscriptions.reload() close: -> + ActionCable.log("WebSocket onclose event") @disconnect() error: -> + ActionCable.log("WebSocket onerror event") @disconnect() disconnect: -> diff --git a/actioncable/app/assets/javascripts/action_cable/connection_monitor.coffee b/actioncable/app/assets/javascripts/action_cable/connection_monitor.coffee index 99b9a1c6d5..75a6f1fb07 100644 --- a/actioncable/app/assets/javascripts/action_cable/connection_monitor.coffee +++ b/actioncable/app/assets/javascripts/action_cable/connection_monitor.coffee @@ -17,6 +17,7 @@ class ActionCable.ConnectionMonitor @reset() @pingedAt = now() delete @disconnectedAt + ActionCable.log("ConnectionMonitor connected") disconnected: -> @disconnectedAt = now() @@ -33,10 +34,12 @@ class ActionCable.ConnectionMonitor @startedAt = now() @poll() document.addEventListener("visibilitychange", @visibilityDidChange) + ActionCable.log("ConnectionMonitor started, pollInterval is #{@getInterval()}ms") stop: -> @stoppedAt = now() document.removeEventListener("visibilitychange", @visibilityDidChange) + ActionCable.log("ConnectionMonitor stopped") poll: -> setTimeout => @@ -52,8 +55,12 @@ class ActionCable.ConnectionMonitor reconnectIfStale: -> if @connectionIsStale() + ActionCable.log("ConnectionMonitor detected stale connection, reconnectAttempts = #{@reconnectAttempts}") @reconnectAttempts++ - unless @disconnectedRecently() + if @disconnectedRecently() + ActionCable.log("ConnectionMonitor skipping reopen because recently disconnected at #{@disconnectedAt}") + else + ActionCable.log("ConnectionMonitor reopening") @consumer.connection.reopen() connectionIsStale: -> @@ -66,6 +73,7 @@ class ActionCable.ConnectionMonitor if document.visibilityState is "visible" setTimeout => if @connectionIsStale() or not @consumer.connection.isOpen() + ActionCable.log("ConnectionMonitor reopening stale connection after visibilitychange to #{document.visibilityState}") @consumer.connection.reopen() , 200 diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb index 874ebe2e71..714d9887d4 100644 --- a/actioncable/lib/action_cable/channel/base.rb +++ b/actioncable/lib/action_cable/channel/base.rb @@ -32,8 +32,8 @@ module ActionCable # # == Action processing # - # Unlike subclasses of ActionController::Base, channels do not follow a REST - # constraint form for their actions. Instead, ActionCable operates through a + # 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. @@ -63,10 +63,10 @@ module ActionCable # end # end # - # In this example, subscribed/unsubscribed are not callable methods, as they + # 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 as it's a private method. You'll see that appear accepts a data + # 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. # @@ -125,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 @@ -166,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 @@ -183,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 @@ -191,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 @@ -224,7 +224,6 @@ module ActionCable end end - def subscribe_to_channel run_callbacks :subscribe do subscribed @@ -237,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 56597d02d7..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:) diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb index 3158f30814..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,8 +39,8 @@ 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 @@ -70,7 +72,8 @@ module ActionCable # Start streaming from the named <tt>broadcasting</tt> pubsub queue. Optionally, you can pass a <tt>callback</tt> that'll be used # instead of the default of just transmitting the updates straight to the subscriber. 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) diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb index b5f898436a..afe0d958d7 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,12 +48,13 @@ module ActionCable include InternalChannel include Authorization - attr_reader :server, :env, :subscriptions, :logger - delegate :stream_event_loop, :worker_pool, :pubsub, to: :server + attr_reader :server, :env, :subscriptions, :logger, :worker_pool + delegate :stream_event_loop, :pubsub, to: :server def initialize(server, env) @server, @env = server, env + @worker_pool = server.worker_pool @logger = new_tagged_logger @websocket = ActionCable::Connection::WebSocket.new(env, self, stream_event_loop) @@ -65,8 +66,8 @@ module ActionCable 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? @@ -76,7 +77,7 @@ module ActionCable 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? @@ -88,7 +89,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 @@ -154,7 +155,7 @@ module ActionCable def handle_open connect if respond_to?(:connect) subscribe_to_internal_channel - beat + confirm_connection_monitor_subscription message_buffer.process! server.add_connection(self) @@ -173,6 +174,13 @@ 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 @@ -185,12 +193,14 @@ 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 @@ -205,7 +215,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 @@ -213,10 +223,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 index ef937d7c16..f6b11e93f0 100644 --- a/actioncable/lib/action_cable/connection/client_socket.rb +++ b/actioncable/lib/action_cable/connection/client_socket.rb @@ -50,14 +50,16 @@ module ActionCable @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 - end - def start_driver - return if @driver.nil? || @driver_started @driver_started = true @driver.start end @@ -132,11 +134,8 @@ module ActionCable @ready_state = CLOSING @close_params = [reason, code] - if @stream - @stream.shutdown - else - finalize_close - end + @stream.shutdown if @stream + finalize_close end def finalize_close diff --git a/actioncable/lib/action_cable/connection/identification.rb b/actioncable/lib/action_cable/connection/identification.rb index 885ff3f102..4a54044aff 100644 --- a/actioncable/lib/action_cable/connection/identification.rb +++ b/actioncable/lib/action_cable/connection/identification.rb @@ -12,7 +12,7 @@ module ActionCable 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. + # 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/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 index ace250cd16..2d97b28c09 100644 --- a/actioncable/lib/action_cable/connection/stream.rb +++ b/actioncable/lib/action_cable/connection/stream.rb @@ -4,15 +4,13 @@ module ActionCable # This class is heavily based on faye-websocket-ruby # # Copyright (c) 2010-2015 James Coglan - class Stream + 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 - - hijack_rack_socket end def each(&callback) @@ -39,16 +37,16 @@ module ActionCable @socket_object.parse(data) end - private - def hijack_rack_socket - return unless @socket_object.env['rack.hijack'] + 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'] + @socket_object.env['rack.hijack'].call + @rack_hijack_io = @socket_object.env['rack.hijack_io'] - @event_loop.attach(@rack_hijack_io, self) - end + @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) diff --git a/actioncable/lib/action_cable/connection/subscriptions.rb b/actioncable/lib/action_cable/connection/subscriptions.rb index d7f95e6a62..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 = {} @@ -54,7 +54,7 @@ module ActionCable 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/engine.rb b/actioncable/lib/action_cable/engine.rb index f5e233e091..c90aadaf2c 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 @@ -31,8 +31,50 @@ module ActionCable self.cable = Rails.application.config_for(config_path).with_indifferent_access end + if 'ApplicationCable::Connection'.safe_constantize + self.connection_class = ApplicationCable::Connection + end + + self.channel_paths = Rails.application.paths['app/channels'].existent + 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 + + initializer "action_cable.set_work_hooks" do |app| + ActiveSupport.on_load(:action_cable) do + ActionCable::Server::Worker.set_callback :work, :around, prepend: true do |_, inner| + app.executor.wrap do + # If we took a while to get the lock, we may have been halted + # in the meantime. As we haven't started doing any real work + # yet, we should pretend that we never made it off the queue. + unless stopping? + inner.call + end + end + end + + wrap = lambda do |_, inner| + app.executor.wrap(&inner) + end + ActionCable::Channel::Base.set_callback :subscribe, :around, prepend: true, &wrap + ActionCable::Channel::Base.set_callback :unsubscribe, :around, prepend: true, &wrap + + app.reloader.before_class_unload do + ActionCable.server.restart + end + end + end end end diff --git a/actioncable/lib/action_cable/gem_version.rb b/actioncable/lib/action_cable/gem_version.rb index a71603e61a..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 = "beta2" + 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/remote_connections.rb b/actioncable/lib/action_cable/remote_connections.rb index 7ec121308a..aeef8abc72 100644 --- a/actioncable/lib/action_cable/remote_connections.rb +++ b/actioncable/lib/action_cable/remote_connections.rb @@ -13,8 +13,8 @@ module ActionCable # 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. + # <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 diff --git a/actioncable/lib/action_cable/server/base.rb b/actioncable/lib/action_cable/server/base.rb index fe48c112df..d9a2653cc2 100644 --- a/actioncable/lib/action_cable/server/base.rb +++ b/actioncable/lib/action_cable/server/base.rb @@ -2,10 +2,10 @@ 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 @@ -19,11 +19,10 @@ module ActionCable 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,6 +33,16 @@ module ActionCable remote_connections.where(identifiers).disconnect end + def restart + connections.each(&:close) + + @mutex.synchronize do + worker_pool.halt if @worker_pool + + @worker_pool = nil + end + end + # Gateway to RemoteConnections. See that class for details. def remote_connections @remote_connections || @mutex.synchronize { @remote_connections ||= RemoteConnections.new(self) } @@ -48,7 +57,7 @@ module ActionCable @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 || @mutex.synchronize do @channel_classes ||= begin @@ -63,7 +72,7 @@ module ActionCable @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 7e8aef45f4..98025f27f2 100644 --- a/actioncable/lib/action_cable/server/broadcasting.rb +++ b/actioncable/lib/action_cable/server/broadcasting.rb @@ -1,6 +1,6 @@ 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 @@ -9,24 +9,24 @@ module ActionCable # end # end # - # # Somewhere in your app this is called, perhaps from a NewCommentJob + # # 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 + # # 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) + Broadcaster.new(self, String(broadcasting)) end private diff --git a/actioncable/lib/action_cable/server/configuration.rb b/actioncable/lib/action_cable/server/configuration.rb index 9a248933c4..9a7301287c 100644 --- a/actioncable/lib/action_cable/server/configuration.rb +++ b/actioncable/lib/action_cable/server/configuration.rb @@ -1,31 +1,24 @@ 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 :channel_load_paths attr_accessor :disable_request_forgery_protection, :allowed_request_origins - attr_accessor :cable, :url + attr_accessor :cable, :url, :mount_path + + attr_accessor :channel_paths # :nodoc: def initialize @log_tags = [] - @connection_class = ApplicationCable::Connection - @worker_pool_size = 100 - - @channel_load_paths = [Rails.root.join('app/channels')] + @connection_class = ActionCable::Connection::Base + @worker_pool_size = 100 @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 diff --git a/actioncable/lib/action_cable/server/connections.rb b/actioncable/lib/action_cable/server/connections.rb index 8671dd5ebd..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,7 +18,7 @@ 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 @heartbeat_timer ||= Concurrent::TimerTask.new(execution_interval: BEAT_INTERVAL) do diff --git a/actioncable/lib/action_cable/server/worker.rb b/actioncable/lib/action_cable/server/worker.rb index 3b6c6d44a1..49cbaec0c0 100644 --- a/actioncable/lib/action_cable/server/worker.rb +++ b/actioncable/lib/action_cable/server/worker.rb @@ -4,8 +4,8 @@ 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 thread_mattr_accessor :connection @@ -20,6 +20,26 @@ module ActionCable ) end + # Stop processing work: any work that has not already started + # running will be discarded from the queue + def halt + @pool.kill + end + + def stopping? + @pool.shuttingdown? + end + + def work(connection) + self.connection = connection + + run_callbacks :work do + yield + end + ensure + self.connection = nil + end + def async_invoke(receiver, method, *args) @pool.post do invoke(receiver, method, *args) @@ -27,19 +47,15 @@ module ActionCable end def invoke(receiver, method, *args) - begin - self.connection = receiver - - run_callbacks :work do + work(receiver) do + begin receiver.send method, *args - end - rescue Exception => e - logger.error "There was an exception - #{e.class}(#{e.message})" - logger.error e.backtrace.join("\n") + 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 + receiver.handle_exception if receiver.respond_to?(:handle_exception) + end end end @@ -50,14 +66,8 @@ module ActionCable 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 + work(channel.connection) do + callback.respond_to?(:call) ? channel.instance_exec(&callback) : channel.send(callback) 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 index ecece4e270..c1e4aa8103 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,6 @@ 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 @@ -13,10 +12,8 @@ module ActionCable 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 +end diff --git a/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb b/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb index d697548cbd..e6c862959b 100644 --- a/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb +++ b/actioncable/lib/action_cable/subscription_adapter/evented_redis.rb @@ -13,6 +13,14 @@ module ActionCable 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 @@ -41,7 +49,7 @@ module ActionCable 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_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 @@ -51,7 +59,7 @@ module ActionCable def redis_connection_for_broadcasts @redis_connection_for_broadcasts || @server.mutex.synchronize do - @redis_connection_for_broadcasts ||= ::Redis.new(@server.config.cable) + @redis_connection_for_broadcasts ||= self.class.redis_connector.call(@server.config.cable) end end diff --git a/actioncable/lib/action_cable/subscription_adapter/redis.rb b/actioncable/lib/action_cable/subscription_adapter/redis.rb index 7076383efe..6b4236e7d3 100644 --- a/actioncable/lib/action_cable/subscription_adapter/redis.rb +++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb @@ -6,6 +6,10 @@ 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 @@ -29,7 +33,7 @@ module ActionCable end def redis_connection_for_subscriptions - ::Redis.new(@server.config.cable) + redis_connection end private @@ -39,10 +43,14 @@ module ActionCable def redis_connection_for_broadcasts @redis_connection_for_broadcasts || @server.mutex.synchronize do - @redis_connection_for_broadcasts ||= ::Redis.new(@server.config.cable) + @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() diff --git a/actioncable/lib/rails/generators/channel/channel_generator.rb b/actioncable/lib/rails/generators/channel/channel_generator.rb index c5d398810a..6debe40c91 100644 --- a/actioncable/lib/rails/generators/channel/channel_generator.rb +++ b/actioncable/lib/rails/generators/channel/channel_generator.rb @@ -15,12 +15,29 @@ module Rails 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 diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb index 947efd96d4..526ea92e4f 100644 --- a/actioncable/test/channel/stream_test.rb +++ b/actioncable/test/channel/stream_test.rb @@ -14,7 +14,12 @@ class ActionCable::Channel::StreamTest < ActionCable::TestCase def send_confirmation transmit_subscription_confirmation end + end + class SymbolChannel < ActionCable::Channel::Base + def subscribed + stream_from :channel + end end test "streaming start and stop" do @@ -28,6 +33,17 @@ class ActionCable::Channel::StreamTest < ActionCable::TestCase end end + test "stream from non-string channel" do + run_in_eventmachine do + connection = TestConnection.new + connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("channel", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } + channel = SymbolChannel.new connection, "" + + connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) } + channel.unsubscribe_from_channel + end + end + test "stream_for" do run_in_eventmachine do connection = TestConnection.new diff --git a/actioncable/test/client/echo_channel.rb b/actioncable/test/client/echo_channel.rb index 63e35f194a..5a7bac25c5 100644 --- a/actioncable/test/client/echo_channel.rb +++ b/actioncable/test/client/echo_channel.rb @@ -3,6 +3,10 @@ class EchoChannel < ActionCable::Channel::Base stream_from "global" end + def unsubscribed + 'Goodbye from EchoChannel!' + end + def ding(data) transmit(dong: data['message']) end diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb index 199d2b90a3..a6619d3bd2 100644 --- a/actioncable/test/client_test.rb +++ b/actioncable/test/client_test.rb @@ -12,24 +12,15 @@ class ClientTest < ActionCable::TestCase WAIT_WHEN_NOT_EXPECTING_EVENT = 0.2 def setup - # TODO: ActionCable requires a *lot* of setup at the moment... - ::Object.const_set(:ApplicationCable, Module.new) - ::ApplicationCable.const_set(:Connection, Class.new(ActionCable::Connection::Base)) - - ::Object.const_set(:Rails, Module.new) - ::Rails.singleton_class.send(:define_method, :root) { Pathname.new(__dir__) } - ActionCable.instance_variable_set(:@server, nil) server = ActionCable.server - server.config = ActionCable::Server::Configuration.new - inner_logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } - server.config.logger = ActionCable::Connection::TaggedLoggerProxy.new(inner_logger, tags: []) + server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } server.config.cable = { adapter: 'async' }.with_indifferent_access # and now the "real" setup for our test: server.config.disable_request_forgery_protection = true - server.config.channel_load_paths = [File.expand_path('client', __dir__)] + server.config.channel_paths = [ File.expand_path('client/echo_channel.rb', __dir__) ] Thread.new { EventMachine.run } unless EventMachine.reactor_running? Thread.pass until EventMachine.reactor_running? @@ -40,15 +31,6 @@ class ClientTest < ActionCable::TestCase def teardown $VERBOSE = @previous_verbose - - begin - ::Object.send(:remove_const, :ApplicationCable) - rescue NameError - end - begin - ::Object.send(:remove_const, :Rails) - rescue NameError - end end def with_puma_server(rack_app = ActionCable.server, port = 3099) @@ -72,7 +54,7 @@ class ClientTest < ActionCable::TestCase @ws = Faye::WebSocket::Client.new("ws://127.0.0.1:#{port}/") @messages = Queue.new @closed = Concurrent::Event.new - @has_messages = Concurrent::Event.new + @has_messages = Concurrent::Semaphore.new(0) @pings = 0 open = Concurrent::Event.new @@ -97,7 +79,7 @@ class ClientTest < ActionCable::TestCase @pings += 1 else @messages << hash - @has_messages.set + @has_messages.release end end @@ -110,8 +92,7 @@ class ClientTest < ActionCable::TestCase end def read_message - @has_messages.wait(WAIT_WHEN_EXPECTING_EVENT) if @messages.empty? - @has_messages.reset if @messages.size < 2 + @has_messages.try_acquire(1, WAIT_WHEN_EXPECTING_EVENT) msg = @messages.pop(true) raise msg if msg.is_a?(Exception) @@ -122,9 +103,11 @@ class ClientTest < ActionCable::TestCase def read_messages(expected_size = 0) list = [] loop do - @has_messages.wait(list.size < expected_size ? WAIT_WHEN_EXPECTING_EVENT : WAIT_WHEN_NOT_EXPECTING_EVENT) - if @has_messages.set? - list << read_message + if @has_messages.try_acquire(1, list.size < expected_size ? WAIT_WHEN_EXPECTING_EVENT : WAIT_WHEN_NOT_EXPECTING_EVENT) + msg = @messages.pop(true) + raise msg if msg.is_a?(Exception) + + list << msg else break end @@ -144,8 +127,16 @@ class ClientTest < ActionCable::TestCase end @ws.close + wait_for_close + end + + def wait_for_close @closed.wait(WAIT_WHEN_EXPECTING_EVENT) end + + def closed? + @closed.set? + end end def faye_client(port) @@ -187,7 +178,7 @@ class ClientTest < ActionCable::TestCase def test_many_clients with_puma_server do |port| - clients = 200.times.map { faye_client(port) } + clients = 100.times.map { faye_client(port) } clients.map {|c| Concurrent::Future.execute { c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') @@ -216,4 +207,37 @@ class ClientTest < ActionCable::TestCase c.close # disappear before read end end + + def test_unsubscribe_client + with_puma_server do |port| + app = ActionCable.server + identifier = JSON.dump(channel: 'EchoChannel') + + c = faye_client(port) + c.send_message command: 'subscribe', identifier: identifier + assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "type"=>"confirm_subscription"}, c.read_message) + assert_equal(1, app.connections.count) + assert(app.remote_connections.where(identifier: identifier)) + + channel = app.connections.first.subscriptions.send(:subscriptions).first[1] + channel.expects(:unsubscribed) + c.close + sleep 0.1 # Data takes a moment to process + + # All data is removed: No more connection or subscription information! + assert_equal(0, app.connections.count) + end + end + + def test_server_restart + with_puma_server do |port| + c = faye_client(port) + c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') + assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "type"=>"confirm_subscription"}, c.read_message) + + ActionCable.server.restart + c.wait_for_close + assert c.closed? + end + end end diff --git a/actioncable/test/connection/base_test.rb b/actioncable/test/connection/base_test.rb index e2b017a9a1..fb11f9be64 100644 --- a/actioncable/test/connection/base_test.rb +++ b/actioncable/test/connection/base_test.rb @@ -56,7 +56,7 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase run_in_eventmachine do connection = open_connection - connection.websocket.expects(:transmit).with(regexp_matches(/\_ping/)) + connection.websocket.expects(:transmit).with({ identifier: "_ping", type: "confirm_subscription" }.to_json) connection.message_buffer.expects(:process!) connection.process @@ -108,6 +108,26 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase end end + test "rejecting a connection causes a 404" do + run_in_eventmachine do + class CallMeMaybe + def call(*) + raise 'Do not call me!' + end + end + + env = Rack::MockRequest.env_for( + "/test", + { 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket', + 'HTTP_ORIGIN' => 'http://rubyonrails.org', 'rack.hijack' => CallMeMaybe.new } + ) + + connection = ActionCable::Connection::Base.new(@server, env) + response = connection.process + assert_equal 404, response[0] + end + end + private def open_connection env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket', diff --git a/actioncable/test/server/broadcasting_test.rb b/actioncable/test/server/broadcasting_test.rb new file mode 100644 index 0000000000..3b4a7eaf90 --- /dev/null +++ b/actioncable/test/server/broadcasting_test.rb @@ -0,0 +1,15 @@ +require "test_helper" + +class BroadcastingTest < ActiveSupport::TestCase + class TestServer + include ActionCable::Server::Broadcasting + end + + test "fetching a broadcaster converts the broadcasting queue to a string" do + broadcasting = :test_queue + server = TestServer.new + broadcaster = server.broadcaster_for(broadcasting) + + assert_equal "test_queue", broadcaster.broadcasting + end +end diff --git a/actioncable/test/stubs/test_server.rb b/actioncable/test/stubs/test_server.rb index 56d132b30a..5916cf1e83 100644 --- a/actioncable/test/stubs/test_server.rb +++ b/actioncable/test/stubs/test_server.rb @@ -17,4 +17,8 @@ class TestServer def stream_event_loop @stream_event_loop ||= ActionCable::Connection::StreamEventLoop.new end + + def worker_pool + @worker_pool ||= ActionCable::Server::Worker.new(max_size: 5) + end end diff --git a/actioncable/test/subscription_adapter/base_test.rb b/actioncable/test/subscription_adapter/base_test.rb index 7a7ae131e6..256dce673f 100644 --- a/actioncable/test/subscription_adapter/base_test.rb +++ b/actioncable/test/subscription_adapter/base_test.rb @@ -43,7 +43,7 @@ class ActionCable::SubscriptionAdapter::BaseTest < ActionCable::TestCase assert_respond_to(SuccessAdapter.new(@server), :broadcast) - assert_nothing_raised NotImplementedError do + assert_nothing_raised do broadcast end end @@ -55,7 +55,7 @@ class ActionCable::SubscriptionAdapter::BaseTest < ActionCable::TestCase assert_respond_to(SuccessAdapter.new(@server), :subscribe) - assert_nothing_raised NotImplementedError do + assert_nothing_raised do subscribe end end @@ -66,7 +66,7 @@ class ActionCable::SubscriptionAdapter::BaseTest < ActionCable::TestCase assert_respond_to(SuccessAdapter.new(@server), :unsubscribe) - assert_nothing_raised NotImplementedError do + assert_nothing_raised do unsubscribe end end diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb index 361858784e..b31c2aa36c 100644 --- a/actioncable/test/subscription_adapter/common.rb +++ b/actioncable/test/subscription_adapter/common.rb @@ -9,20 +9,7 @@ module CommonSubscriptionAdapterTest WAIT_WHEN_NOT_EXPECTING_EVENT = 0.2 def setup - # TODO: ActionCable requires a *lot* of setup at the moment... - ::Object.const_set(:ApplicationCable, Module.new) - ::ApplicationCable.const_set(:Connection, Class.new(ActionCable::Connection::Base)) - - ::Object.const_set(:Rails, Module.new) - ::Rails.singleton_class.send(:define_method, :root) { Pathname.new(__dir__) } - server = ActionCable::Server::Base.new - server.config = ActionCable::Server::Configuration.new - inner_logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } - server.config.logger = ActionCable::Connection::TaggedLoggerProxy.new(inner_logger, tags: []) - - - # and now the "real" setup for our test: server.config.cable = cable_config.with_indifferent_access adapter_klass = server.config.pubsub_adapter @@ -34,15 +21,6 @@ module CommonSubscriptionAdapterTest def teardown @tx_adapter.shutdown if @tx_adapter && @tx_adapter != @rx_adapter @rx_adapter.shutdown if @rx_adapter - - begin - ::Object.send(:remove_const, :ApplicationCable) - rescue NameError - end - begin - ::Object.send(:remove_const, :Rails) - rescue NameError - end end diff --git a/actioncable/test/worker_test.rb b/actioncable/test/worker_test.rb index 4a699cde27..654f49821e 100644 --- a/actioncable/test/worker_test.rb +++ b/actioncable/test/worker_test.rb @@ -17,7 +17,9 @@ class WorkerTest < ActiveSupport::TestCase end def logger - ActionCable.server.logger + # Impersonating a connection requires a TaggedLoggerProxy'ied logger. + inner_logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } + ActionCable::Connection::TaggedLoggerProxy.new(inner_logger, tags: []) end end |