diff options
Diffstat (limited to 'actioncable')
119 files changed, 8498 insertions, 0 deletions
diff --git a/actioncable/.babelrc b/actioncable/.babelrc new file mode 100644 index 0000000000..4f0c469c60 --- /dev/null +++ b/actioncable/.babelrc @@ -0,0 +1,8 @@ +{ + "presets": [ + ["env", { "modules": false, "loose": true } ] + ], + "plugins": [ + "external-helpers" + ] +} diff --git a/actioncable/.eslintrc b/actioncable/.eslintrc new file mode 100644 index 0000000000..3d9ecd4bce --- /dev/null +++ b/actioncable/.eslintrc @@ -0,0 +1,19 @@ +{ + "extends": "eslint:recommended", + "rules": { + "semi": ["error", "never"], + "quotes": ["error", "double"], + "no-unused-vars": ["error", { "vars": "all", "args": "none" }] + }, + "plugins": [ + "import" + ], + "env": { + "browser": true, + "es6": true + }, + "parserOptions": { + "ecmaVersion": 6, + "sourceType": "module" + } +} diff --git a/actioncable/.gitignore b/actioncable/.gitignore new file mode 100644 index 0000000000..7fa7c03e03 --- /dev/null +++ b/actioncable/.gitignore @@ -0,0 +1,4 @@ +/app/javascript/action_cable/internal.js +/src +/test/javascript/compiled/ +/tmp/ diff --git a/actioncable/CHANGELOG.md b/actioncable/CHANGELOG.md new file mode 100644 index 0000000000..162de0df0b --- /dev/null +++ b/actioncable/CHANGELOG.md @@ -0,0 +1,74 @@ +* The JavaScript WebSocket client will no longer try to reconnect + when you call `reject_unauthorized_connection` on the connection. + + *Mick Staugaard* + +* `ActionCable.Connection#getState` now references the configurable + `ActionCable.adapters.WebSocket` property rather than the `WebSocket` global + variable, matching the behavior of `ActionCable.Connection#open`. + + *Richard Macklin* + +* The ActionCable javascript package has been converted from CoffeeScript + to ES2015, and we now publish the source code in the npm distribution. + + This allows ActionCable users to depend on the javascript source code + rather than the compiled code, which can produce smaller javascript bundles. + + This change includes some breaking changes to optional parts of the + ActionCable javascript API: + + - Configuration of the WebSocket adapter and logger adapter have been moved + from properties of `ActionCable` to properties of `ActionCable.adapters`. + If you are currently configuring these adapters you will need to make + these changes when upgrading: + + ```diff + - ActionCable.WebSocket = MyWebSocket + + ActionCable.adapters.WebSocket = MyWebSocket + ``` + ```diff + - ActionCable.logger = myLogger + + ActionCable.adapters.logger = myLogger + ``` + + - The `ActionCable.startDebugging()` and `ActionCable.stopDebugging()` + methods have been removed and replaced with the property + `ActionCable.logger.enabled`. If you are currently using these methods you + will need to make these changes when upgrading: + + ```diff + - ActionCable.startDebugging() + + ActionCable.logger.enabled = true + ``` + ```diff + - ActionCable.stopDebugging() + + ActionCable.logger.enabled = false + ``` + + *Richard Macklin* + +* Add `id` option to redis adapter so now you can distinguish + ActionCable's redis connections among others. Also, you can set + custom id in options. + + Before: + ``` + $ redis-cli client list + id=669 addr=127.0.0.1:46442 fd=8 name= age=18 ... + ``` + + After: + ``` + $ redis-cli client list + id=673 addr=127.0.0.1:46516 fd=8 name=ActionCable-PID-19413 age=2 ... + ``` + + *Ilia Kasianenko* + +* Rails 6 requires Ruby 2.5.0 or newer. + + *Jeremy Daer*, *Kasper Timm Hansen* + + +Please check [5-2-stable](https://github.com/rails/rails/blob/5-2-stable/actioncable/CHANGELOG.md) for previous changes. diff --git a/actioncable/MIT-LICENSE b/actioncable/MIT-LICENSE new file mode 100644 index 0000000000..bdb42d988a --- /dev/null +++ b/actioncable/MIT-LICENSE @@ -0,0 +1,20 @@ +Copyright (c) 2015-2019 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. diff --git a/actioncable/README.md b/actioncable/README.md new file mode 100644 index 0000000000..a2783d6f45 --- /dev/null +++ b/actioncable/README.md @@ -0,0 +1,567 @@ +# Action Cable – Integrated WebSockets for Rails + +Action Cable seamlessly integrates WebSockets with the rest of your Rails application. +It allows for real-time features to be written in Ruby in the same style +and form as the rest of your Rails application, while still being performant +and scalable. It's a full-stack offering that provides both a client-side +JavaScript framework and a server-side Ruby framework. You have access to your full +domain model written with Active Record or your ORM of choice. + +## Terminology + +A single Action Cable server can handle multiple connection instances. It has one +connection instance per WebSocket connection. A single user may have multiple +WebSockets open to your application if they use multiple browser tabs or devices. +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 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 +the subscriber and the channel is, surprise-surprise, called a subscription. A consumer +can act as a subscriber to a given channel any number of times. For example, a consumer +could subscribe to multiple chat rooms at the same time. (And remember that a physical user may +have multiple consumers, one per tab/device open to your connection). + +Each channel can then again be streaming zero or more broadcastings. A broadcasting is a +pubsub link where anything transmitted by the broadcaster is sent directly to the channel +subscribers who are streaming that named broadcasting. + +As you can see, this is a fairly deep architectural stack. There's a lot of new terminology +to identify the new pieces, and on top of that, you're dealing with both client and server side +reflections of each unit. + +## Examples + +### 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, +if all is well. Here's the simplest example starting with the server-side connection class: + +```ruby +# app/channels/application_cable/connection.rb +module ApplicationCable + class Connection < ActionCable::Connection::Base + identified_by :current_user + + def connect + self.current_user = find_verified_user + end + + private + def find_verified_user + if verified_user = User.find_by(id: cookies.encrypted[:user_id]) + verified_user + else + reject_unauthorized_connection + end + end + end +end +``` +Here `identified_by` is a connection identifier that can be used to find the specific connection again or later. +Note that anything marked as an identifier will automatically create a delegate by the same name on any channel instances created off the connection. + +This relies on the fact that you will already have handled authentication of the user, and +that a successful authentication sets a signed cookie with the `user_id`. This cookie is then +automatically sent to the connection instance when a new connection is attempted, and you +use that to set the `current_user`. By identifying the connection by this same current_user, +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). + +Next, you should define your `ApplicationCable::Channel` class in Ruby. This is the place where you put +shared logic between your channels. + +```ruby +# app/channels/application_cable/channel.rb +module ApplicationCable + class Channel < ActionCable::Channel::Base + end +end +``` + +The client-side needs to setup a consumer instance of this connection. That's done like so: + +```js +// app/assets/javascripts/cable.js +//= require action_cable +//= require_self +//= require_tree ./channels + +(function() { + this.App || (this.App = {}); + + App.cable = ActionCable.createConsumer("ws://cable.example.com"); +}).call(this); +``` + +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. + +That's all you need to establish the connection! But of course, this isn't very useful in +itself. This just gives you the plumbing. To make stuff happen, you need content. That content +is defined by declaring channels on the server and allowing the consumer to subscribe to them. + + +### Channel example 1: User appearances + +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: + +```ruby +# app/channels/appearance_channel.rb +class AppearanceChannel < ApplicationCable::Channel + def subscribed + current_user.appear + end + + def unsubscribed + current_user.disappear + end + + def appear(data) + current_user.appear on: data['appearing_on'] + end + + def away + current_user.away + end +end +``` + +The `#subscribed` callback is invoked when, as we'll show below, a client-side subscription is initiated. In this case, +we take that opportunity to say "the current user has indeed appeared". That appear/disappear API could be backed by +Redis or a database or whatever else. Here's what the client-side of that looks like: + +```coffeescript +# app/assets/javascripts/cable/subscriptions/appearance.coffee +App.cable.subscriptions.create "AppearanceChannel", + # Called when the subscription is ready for use on the server + connected: -> + @install() + @appear() + + # Called when the WebSocket connection is closed + disconnected: -> + @uninstall() + + # Called when the subscription is rejected by the server + rejected: -> + @uninstall() + + appear: -> + # Calls `AppearanceChannel#appear(data)` on the server + @perform("appear", appearing_on: $("main").data("appearing-on")) + + away: -> + # Calls `AppearanceChannel#away` on the server + @perform("away") + + + buttonSelector = "[data-behavior~=appear_away]" + + install: -> + $(document).on "turbolinks:load.appearance", => + @appear() + + $(document).on "click.appearance", buttonSelector, => + @away() + false + + $(buttonSelector).show() + + uninstall: -> + $(document).off(".appearance") + $(buttonSelector).hide() +``` + +Simply calling `App.cable.subscriptions.create` will setup the subscription, which will call `AppearanceChannel#subscribed`, +which in turn is linked to the original `App.cable` -> `ApplicationCable::Connection` instances. + +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. + +### Channel example 2: Receiving new web notifications + +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 +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: + +```ruby +# app/channels/web_notifications_channel.rb +class WebNotificationsChannel < ApplicationCable::Channel + def subscribed + stream_from "web_notifications_#{current_user.id}" + end +end +``` + +```coffeescript +# Client-side, 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"] +``` + +```ruby +# Somewhere in your app this is called, perhaps from a NewCommentJob +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 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`. + + +### Passing Parameters to Channel + +You can pass parameters from the client side to the server side when creating a subscription. For example: + +```ruby +# app/channels/chat_channel.rb +class ChatChannel < ApplicationCable::Channel + def subscribed + stream_from "chat_#{params[:room]}" + end +end +``` + +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 +App.cable.subscriptions.create { channel: "ChatChannel", room: "Best Room" }, + received: (data) -> + @appendLine(data) + + appendLine: (data) -> + html = @createLine(data) + $("[data-chat-room='Best Room']").append(html) + + createLine: (data) -> + """ + <article class="chat-line"> + <span class="speaker">#{data["sent_by"]}</span> + <span class="body">#{data["body"]}</span> + </article> + """ +``` + +```ruby +# Somewhere in your app this is called, perhaps from a NewCommentJob +ActionCable.server.broadcast \ + "chat_#{room}", { sent_by: 'Paul', body: 'This is a cool chat app.' } +``` + + +### Rebroadcasting message + +A common use case is to rebroadcast a message sent by one client to any other connected clients. + +```ruby +# app/channels/chat_channel.rb +class ChatChannel < ApplicationCable::Channel + def subscribed + stream_from "chat_#{params[:room]}" + end + + def receive(data) + ActionCable.server.broadcast "chat_#{params[:room]}", data + end +end +``` + +```coffeescript +# Client-side, which assumes you've already requested the right to send web notifications +App.chatChannel = App.cable.subscriptions.create { channel: "ChatChannel", room: "Best Room" }, + received: (data) -> + # data => { sent_by: "Paul", body: "This is a cool chat app." } + +App.chatChannel.send({ sent_by: "Paul", body: "This is a cool chat app." }) +``` + +The rebroadcast will be received by all connected clients, _including_ the client that sent the message. Note that params are the same as they were when you subscribed to the channel. + + +### More complete examples + +See the [rails/actioncable-examples](https://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: 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 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 Action Cable config file in a Rails initializer with something like: + +```ruby +Rails.application.paths.add "config/cable", with: "somewhere/else/cable.yml" +``` + +### Allowed Request Origins + +Action Cable will only accept requests from specific origins. + +By default, only an origin matching the cable server itself will be permitted. +Additional origins can be specified using strings or regular expressions, provided in an array. + +```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". + +To disable protection and allow requests from any origin: + +```ruby +Rails.application.config.action_cable.disable_request_forgery_protection = true +``` + +To disable automatic access for same-origin requests, and strictly allow +only the configured origins: + +```ruby +Rails.application.config.action_cable.allow_same_origin_as_host = false +``` + +### 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. +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". + +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: + +```ruby +config.action_cable.url = "ws://example.com:28080" +``` + +Then add the following line to your layout before your JavaScript tag: + +```erb +<%= action_cable_meta_tag %> +``` + +And finally, create your consumer like so: + +```coffeescript +App.cable = ActionCable.createConsumer() +``` + +### Other Configurations + +The other common option to configure is the log tags applied to the per-connection logger. Here's an example that uses the user account id if available, else "no-account" while tagging: + +```ruby +config.action_cable.log_tags = [ + -> request { request.env['user_account_id'] || "no-account" }, + :action_cable, + -> request { request.uuid } +] +``` + +For a full list of all configuration options, see the `ActionCable::Server::Configuration` class. + +Also note that your server must provide at least the same number of database connections as you have workers. The default worker pool is set to 4, so that means you have to make at least that available. You can change that in `config/database.yml` through the `pool` attribute. + + +## 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 +application. The recommended basic setup is as follows: + +```ruby +# cable/config.ru +require_relative '../config/environment' +Rails.application.eager_load! + +run ActionCable.server +``` + +Then you start the server using a binstub in bin/cable ala: +```sh +#!/bin/bash +bundle exec puma -p 28080 cable/config.ru +``` + +The above will start a cable server on port 28080. + +### In app + +If you are using a server that supports the [Rack socket hijacking API](https://www.rubydoc.info/github/rack/rack/file/SPEC#label-Hijacking), Action Cable can run alongside your Rails application. For example, to listen for WebSocket requests on `/websocket`, specify that path to `config.action_cable.mount_path`: + +```ruby +# config/application.rb +class Application < Rails::Application + config.action_cable.mount_path = '/websocket' +end +``` + +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 + +Beware that currently, the cable server will _not_ auto-reload any changes in the framework. As we've discussed, long-running cable connections mean long-running objects. We don't yet have a way of reloading the classes of those objects in a safe manner. So when you change your channels, or the model your channels use, you must restart the cable server. + +We'll get all this abstracted properly when the framework is integrated into Rails. + +The WebSocket server doesn't have access to the session, but it has access to the cookies. This can be used when you need to handle authentication. You can see one way of doing that with Devise in this [article](https://greg.molnar.io/blog/actioncable-devise-authentication/). + +## Dependencies + +Action Cable provides a subscription adapter interface to process its pubsub internals. By default, asynchronous, inline, PostgreSQL, and 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 [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 +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. + +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](https://www.rubydoc.info/github/rack/rack/file/SPEC#label-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. + +## Frontend assets + +Action Cable's frontend assets are distributed through two channels: the +official gem and npm package, both titled `actioncable`. + +### Gem usage + +Through the `actioncable` gem, Action Cable's frontend assets are +available through the Rails Asset Pipeline. Create a `cable.js` or +`cable.coffee` file (this is automatically done for you with Rails +generators), and then simply require the assets: + +In JavaScript... + +```javascript +//= require action_cable +``` + +... and in CoffeeScript: + +```coffeescript +#= require action_cable +``` + +### npm usage + +In addition to being available through the `actioncable` gem, Action Cable's +frontend JS assets are also bundled in an officially supported npm module, +intended for usage in standalone frontend applications that communicate with a +Rails application. A common use case for this could be if you have a decoupled +frontend application written in React, Ember.js, etc. and want to add real-time +WebSocket functionality. + +### Installation + +``` +npm install actioncable --save +``` + +### Usage + +The `ActionCable` constant is available as a `require`-able module, so +you only have to require the package to gain access to the API that is +provided. + +In JavaScript... + +```javascript +ActionCable = require('actioncable') + +var cable = ActionCable.createConsumer('wss://RAILS-API-PATH.com/cable') + +cable.subscriptions.create('AppearanceChannel', { + // normal channel code goes here... +}); +``` + +and in CoffeeScript... + +```coffeescript +ActionCable = require('actioncable') + +cable = ActionCable.createConsumer('wss://RAILS-API-PATH.com/cable') + +cable.subscriptions.create 'AppearanceChannel', + # normal channel code goes here... +``` + +## Download and Installation + +The latest version of Action Cable can be installed with [RubyGems](#gem-usage), +or with [npm](#npm-usage). + +Source code can be downloaded as part of the Rails project on GitHub + +* https://github.com/rails/rails/tree/master/actioncable + +## License + +Action Cable is released under the MIT license: + +* https://opensource.org/licenses/MIT + + +## Support + +API documentation is at: + +* http://api.rubyonrails.org + +Bug reports for the Ruby on Rails project can be filed here: + +* https://github.com/rails/rails/issues + +Feature requests should be discussed on the rails-core mailing list here: + +* https://groups.google.com/forum/?fromgroups#!forum/rubyonrails-core diff --git a/actioncable/Rakefile b/actioncable/Rakefile new file mode 100644 index 0000000000..35de50f05a --- /dev/null +++ b/actioncable/Rakefile @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +require "rake/testtask" +require "pathname" +require "open3" +require "action_cable" + +task default: :test + +task :package + +Rake::TestTask.new do |t| + t.libs << "test" + t.test_files = Dir.glob("#{__dir__}/test/**/*_test.rb") + t.warning = true + t.verbose = true + t.ruby_opts = ["--dev"] if defined?(JRUBY_VERSION) +end + +namespace :test do + task :isolated do + Dir.glob("test/**/*_test.rb").all? do |file| + sh(Gem.ruby, "-w", "-Ilib:test", file) + end || raise("Failures") + end + + task :integration do + system("yarn test") || raise("Failures") + end +end + +namespace :assets do + desc "Generate ActionCable::INTERNAL JS module" + task :codegen do + require "json" + require "action_cable" + + File.open(File.join(__dir__, "app/javascript/action_cable/internal.js").to_s, "w+") do |file| + file.write("export default #{JSON.generate(ActionCable::INTERNAL)}") + end + end +end diff --git a/actioncable/actioncable.gemspec b/actioncable/actioncable.gemspec new file mode 100644 index 0000000000..29836f012f --- /dev/null +++ b/actioncable/actioncable.gemspec @@ -0,0 +1,35 @@ +# frozen_string_literal: true + +version = File.read(File.expand_path("../RAILS_VERSION", __dir__)).strip + +Gem::Specification.new do |s| + s.platform = Gem::Platform::RUBY + s.name = "actioncable" + s.version = version + s.summary = "WebSocket framework for Rails." + s.description = "Structure many real-time application concerns into channels over a single WebSocket connection." + + s.required_ruby_version = ">= 2.5.0" + + s.license = "MIT" + + s.author = ["Pratik Naik", "David Heinemeier Hansson"] + s.email = ["pratiknaik@gmail.com", "david@loudthinking.com"] + s.homepage = "http://rubyonrails.org" + + s.files = Dir["CHANGELOG.md", "MIT-LICENSE", "README.md", "lib/**/*", "app/assets/javascripts/action_cable.js"] + s.require_path = "lib" + + s.metadata = { + "source_code_uri" => "https://github.com/rails/rails/tree/v#{version}/actioncable", + "changelog_uri" => "https://github.com/rails/rails/blob/v#{version}/actioncable/CHANGELOG.md" + } + + # NOTE: Please read our dependency guidelines before updating versions: + # https://edgeguides.rubyonrails.org/security.html#dependency-management-and-cves + + s.add_dependency "actionpack", version + + s.add_dependency "nio4r", "~> 2.0" + s.add_dependency "websocket-driver", ">= 0.6.1" +end diff --git a/actioncable/app/assets/javascripts/action_cable.js b/actioncable/app/assets/javascripts/action_cable.js new file mode 100644 index 0000000000..65e32d6c3f --- /dev/null +++ b/actioncable/app/assets/javascripts/action_cable.js @@ -0,0 +1,491 @@ +(function(global, factory) { + typeof exports === "object" && typeof module !== "undefined" ? factory(exports) : typeof define === "function" && define.amd ? define([ "exports" ], factory) : factory(global.ActionCable = {}); +})(this, function(exports) { + "use strict"; + var adapters = { + logger: window.console, + WebSocket: window.WebSocket + }; + var logger = { + log: function log() { + if (this.enabled) { + var _adapters$logger; + for (var _len = arguments.length, messages = Array(_len), _key = 0; _key < _len; _key++) { + messages[_key] = arguments[_key]; + } + messages.push(Date.now()); + (_adapters$logger = adapters.logger).log.apply(_adapters$logger, [ "[ActionCable]" ].concat(messages)); + } + } + }; + var _typeof = typeof Symbol === "function" && typeof Symbol.iterator === "symbol" ? function(obj) { + return typeof obj; + } : function(obj) { + return obj && typeof Symbol === "function" && obj.constructor === Symbol && obj !== Symbol.prototype ? "symbol" : typeof obj; + }; + var classCallCheck = function(instance, Constructor) { + if (!(instance instanceof Constructor)) { + throw new TypeError("Cannot call a class as a function"); + } + }; + var now = function now() { + return new Date().getTime(); + }; + var secondsSince = function secondsSince(time) { + return (now() - time) / 1e3; + }; + var clamp = function clamp(number, min, max) { + return Math.max(min, Math.min(max, number)); + }; + var ConnectionMonitor = function() { + function ConnectionMonitor(connection) { + classCallCheck(this, ConnectionMonitor); + this.visibilityDidChange = this.visibilityDidChange.bind(this); + this.connection = connection; + this.reconnectAttempts = 0; + } + ConnectionMonitor.prototype.start = function start() { + if (!this.isRunning()) { + this.startedAt = now(); + delete this.stoppedAt; + this.startPolling(); + document.addEventListener("visibilitychange", this.visibilityDidChange); + logger.log("ConnectionMonitor started. pollInterval = " + this.getPollInterval() + " ms"); + } + }; + ConnectionMonitor.prototype.stop = function stop() { + if (this.isRunning()) { + this.stoppedAt = now(); + this.stopPolling(); + document.removeEventListener("visibilitychange", this.visibilityDidChange); + logger.log("ConnectionMonitor stopped"); + } + }; + ConnectionMonitor.prototype.isRunning = function isRunning() { + return this.startedAt && !this.stoppedAt; + }; + ConnectionMonitor.prototype.recordPing = function recordPing() { + this.pingedAt = now(); + }; + ConnectionMonitor.prototype.recordConnect = function recordConnect() { + this.reconnectAttempts = 0; + this.recordPing(); + delete this.disconnectedAt; + logger.log("ConnectionMonitor recorded connect"); + }; + ConnectionMonitor.prototype.recordDisconnect = function recordDisconnect() { + this.disconnectedAt = now(); + logger.log("ConnectionMonitor recorded disconnect"); + }; + ConnectionMonitor.prototype.startPolling = function startPolling() { + this.stopPolling(); + this.poll(); + }; + ConnectionMonitor.prototype.stopPolling = function stopPolling() { + clearTimeout(this.pollTimeout); + }; + ConnectionMonitor.prototype.poll = function poll() { + var _this = this; + this.pollTimeout = setTimeout(function() { + _this.reconnectIfStale(); + _this.poll(); + }, this.getPollInterval()); + }; + ConnectionMonitor.prototype.getPollInterval = function getPollInterval() { + var _constructor$pollInte = this.constructor.pollInterval, min = _constructor$pollInte.min, max = _constructor$pollInte.max, multiplier = _constructor$pollInte.multiplier; + var interval = multiplier * Math.log(this.reconnectAttempts + 1); + return Math.round(clamp(interval, min, max) * 1e3); + }; + ConnectionMonitor.prototype.reconnectIfStale = function reconnectIfStale() { + if (this.connectionIsStale()) { + logger.log("ConnectionMonitor detected stale connection. reconnectAttempts = " + this.reconnectAttempts + ", pollInterval = " + this.getPollInterval() + " ms, time disconnected = " + secondsSince(this.disconnectedAt) + " s, stale threshold = " + this.constructor.staleThreshold + " s"); + this.reconnectAttempts++; + if (this.disconnectedRecently()) { + logger.log("ConnectionMonitor skipping reopening recent disconnect"); + } else { + logger.log("ConnectionMonitor reopening"); + this.connection.reopen(); + } + } + }; + ConnectionMonitor.prototype.connectionIsStale = function connectionIsStale() { + return secondsSince(this.pingedAt ? this.pingedAt : this.startedAt) > this.constructor.staleThreshold; + }; + ConnectionMonitor.prototype.disconnectedRecently = function disconnectedRecently() { + return this.disconnectedAt && secondsSince(this.disconnectedAt) < this.constructor.staleThreshold; + }; + ConnectionMonitor.prototype.visibilityDidChange = function visibilityDidChange() { + var _this2 = this; + if (document.visibilityState === "visible") { + setTimeout(function() { + if (_this2.connectionIsStale() || !_this2.connection.isOpen()) { + logger.log("ConnectionMonitor reopening stale connection on visibilitychange. visbilityState = " + document.visibilityState); + _this2.connection.reopen(); + } + }, 200); + } + }; + return ConnectionMonitor; + }(); + ConnectionMonitor.pollInterval = { + min: 3, + max: 30, + multiplier: 5 + }; + ConnectionMonitor.staleThreshold = 6; + var INTERNAL = { + message_types: { + welcome: "welcome", + disconnect: "disconnect", + ping: "ping", + confirmation: "confirm_subscription", + rejection: "reject_subscription" + }, + disconnect_reasons: { + unauthorized: "unauthorized", + invalid_request: "invalid_request", + server_restart: "server_restart" + }, + default_mount_path: "/cable", + protocols: [ "actioncable-v1-json", "actioncable-unsupported" ] + }; + var message_types = INTERNAL.message_types, protocols = INTERNAL.protocols; + var supportedProtocols = protocols.slice(0, protocols.length - 1); + var indexOf = [].indexOf; + var Connection = function() { + function Connection(consumer) { + classCallCheck(this, Connection); + this.open = this.open.bind(this); + this.consumer = consumer; + this.subscriptions = this.consumer.subscriptions; + this.monitor = new ConnectionMonitor(this); + this.disconnected = true; + } + Connection.prototype.send = function send(data) { + if (this.isOpen()) { + this.webSocket.send(JSON.stringify(data)); + return true; + } else { + return false; + } + }; + Connection.prototype.open = function open() { + if (this.isActive()) { + logger.log("Attempted to open WebSocket, but existing socket is " + this.getState()); + return false; + } else { + logger.log("Opening WebSocket, current state is " + this.getState() + ", subprotocols: " + protocols); + if (this.webSocket) { + this.uninstallEventHandlers(); + } + this.webSocket = new adapters.WebSocket(this.consumer.url, protocols); + this.installEventHandlers(); + this.monitor.start(); + return true; + } + }; + Connection.prototype.close = function close() { + var _ref = arguments.length > 0 && arguments[0] !== undefined ? arguments[0] : { + allowReconnect: true + }, allowReconnect = _ref.allowReconnect; + if (!allowReconnect) { + this.monitor.stop(); + } + if (this.isActive()) { + return this.webSocket ? this.webSocket.close() : undefined; + } + }; + Connection.prototype.reopen = function reopen() { + logger.log("Reopening WebSocket, current state is " + this.getState()); + if (this.isActive()) { + try { + return this.close(); + } catch (error) { + logger.log("Failed to reopen WebSocket", error); + } finally { + logger.log("Reopening WebSocket in " + this.constructor.reopenDelay + "ms"); + setTimeout(this.open, this.constructor.reopenDelay); + } + } else { + return this.open(); + } + }; + Connection.prototype.getProtocol = function getProtocol() { + return this.webSocket ? this.webSocket.protocol : undefined; + }; + Connection.prototype.isOpen = function isOpen() { + return this.isState("open"); + }; + Connection.prototype.isActive = function isActive() { + return this.isState("open", "connecting"); + }; + Connection.prototype.isProtocolSupported = function isProtocolSupported() { + return indexOf.call(supportedProtocols, this.getProtocol()) >= 0; + }; + Connection.prototype.isState = function isState() { + for (var _len = arguments.length, states = Array(_len), _key = 0; _key < _len; _key++) { + states[_key] = arguments[_key]; + } + return indexOf.call(states, this.getState()) >= 0; + }; + Connection.prototype.getState = function getState() { + if (this.webSocket) { + for (var state in adapters.WebSocket) { + if (adapters.WebSocket[state] === this.webSocket.readyState) { + return state.toLowerCase(); + } + } + } + return null; + }; + Connection.prototype.installEventHandlers = function installEventHandlers() { + for (var eventName in this.events) { + var handler = this.events[eventName].bind(this); + this.webSocket["on" + eventName] = handler; + } + }; + Connection.prototype.uninstallEventHandlers = function uninstallEventHandlers() { + for (var eventName in this.events) { + this.webSocket["on" + eventName] = function() {}; + } + }; + return Connection; + }(); + Connection.reopenDelay = 500; + Connection.prototype.events = { + message: function message(event) { + if (!this.isProtocolSupported()) { + return; + } + var _JSON$parse = JSON.parse(event.data), identifier = _JSON$parse.identifier, message = _JSON$parse.message, reason = _JSON$parse.reason, reconnect = _JSON$parse.reconnect, type = _JSON$parse.type; + switch (type) { + case message_types.welcome: + this.monitor.recordConnect(); + return this.subscriptions.reload(); + + case message_types.disconnect: + logger.log("Disconnecting. Reason: " + reason); + return this.close({ + allowReconnect: reconnect + }); + + case message_types.ping: + return this.monitor.recordPing(); + + case message_types.confirmation: + return this.subscriptions.notify(identifier, "connected"); + + case message_types.rejection: + return this.subscriptions.reject(identifier); + + default: + return this.subscriptions.notify(identifier, "received", message); + } + }, + open: function open() { + logger.log("WebSocket onopen event, using '" + this.getProtocol() + "' subprotocol"); + this.disconnected = false; + if (!this.isProtocolSupported()) { + logger.log("Protocol is unsupported. Stopping monitor and disconnecting."); + return this.close({ + allowReconnect: false + }); + } + }, + close: function close(event) { + logger.log("WebSocket onclose event"); + if (this.disconnected) { + return; + } + this.disconnected = true; + this.monitor.recordDisconnect(); + return this.subscriptions.notifyAll("disconnected", { + willAttemptReconnect: this.monitor.isRunning() + }); + }, + error: function error() { + logger.log("WebSocket onerror event"); + } + }; + var extend = function extend(object, properties) { + if (properties != null) { + for (var key in properties) { + var value = properties[key]; + object[key] = value; + } + } + return object; + }; + var Subscription = function() { + function Subscription(consumer) { + var params = arguments.length > 1 && arguments[1] !== undefined ? arguments[1] : {}; + var mixin = arguments[2]; + classCallCheck(this, Subscription); + this.consumer = consumer; + this.identifier = JSON.stringify(params); + extend(this, mixin); + } + Subscription.prototype.perform = function perform(action) { + var data = arguments.length > 1 && arguments[1] !== undefined ? arguments[1] : {}; + data.action = action; + return this.send(data); + }; + Subscription.prototype.send = function send(data) { + return this.consumer.send({ + command: "message", + identifier: this.identifier, + data: JSON.stringify(data) + }); + }; + Subscription.prototype.unsubscribe = function unsubscribe() { + return this.consumer.subscriptions.remove(this); + }; + return Subscription; + }(); + var Subscriptions = function() { + function Subscriptions(consumer) { + classCallCheck(this, Subscriptions); + this.consumer = consumer; + this.subscriptions = []; + } + Subscriptions.prototype.create = function create(channelName, mixin) { + var channel = channelName; + var params = (typeof channel === "undefined" ? "undefined" : _typeof(channel)) === "object" ? channel : { + channel: channel + }; + var subscription = new Subscription(this.consumer, params, mixin); + return this.add(subscription); + }; + Subscriptions.prototype.add = function add(subscription) { + this.subscriptions.push(subscription); + this.consumer.ensureActiveConnection(); + this.notify(subscription, "initialized"); + this.sendCommand(subscription, "subscribe"); + return subscription; + }; + Subscriptions.prototype.remove = function remove(subscription) { + this.forget(subscription); + if (!this.findAll(subscription.identifier).length) { + this.sendCommand(subscription, "unsubscribe"); + } + return subscription; + }; + Subscriptions.prototype.reject = function reject(identifier) { + var _this = this; + return this.findAll(identifier).map(function(subscription) { + _this.forget(subscription); + _this.notify(subscription, "rejected"); + return subscription; + }); + }; + Subscriptions.prototype.forget = function forget(subscription) { + this.subscriptions = this.subscriptions.filter(function(s) { + return s !== subscription; + }); + return subscription; + }; + Subscriptions.prototype.findAll = function findAll(identifier) { + return this.subscriptions.filter(function(s) { + return s.identifier === identifier; + }); + }; + Subscriptions.prototype.reload = function reload() { + var _this2 = this; + return this.subscriptions.map(function(subscription) { + return _this2.sendCommand(subscription, "subscribe"); + }); + }; + Subscriptions.prototype.notifyAll = function notifyAll(callbackName) { + var _this3 = this; + for (var _len = arguments.length, args = Array(_len > 1 ? _len - 1 : 0), _key = 1; _key < _len; _key++) { + args[_key - 1] = arguments[_key]; + } + return this.subscriptions.map(function(subscription) { + return _this3.notify.apply(_this3, [ subscription, callbackName ].concat(args)); + }); + }; + Subscriptions.prototype.notify = function notify(subscription, callbackName) { + for (var _len2 = arguments.length, args = Array(_len2 > 2 ? _len2 - 2 : 0), _key2 = 2; _key2 < _len2; _key2++) { + args[_key2 - 2] = arguments[_key2]; + } + var subscriptions = void 0; + if (typeof subscription === "string") { + subscriptions = this.findAll(subscription); + } else { + subscriptions = [ subscription ]; + } + return subscriptions.map(function(subscription) { + return typeof subscription[callbackName] === "function" ? subscription[callbackName].apply(subscription, args) : undefined; + }); + }; + Subscriptions.prototype.sendCommand = function sendCommand(subscription, command) { + var identifier = subscription.identifier; + return this.consumer.send({ + command: command, + identifier: identifier + }); + }; + return Subscriptions; + }(); + var Consumer = function() { + function Consumer(url) { + classCallCheck(this, Consumer); + this.url = url; + this.subscriptions = new Subscriptions(this); + this.connection = new Connection(this); + } + Consumer.prototype.send = function send(data) { + return this.connection.send(data); + }; + Consumer.prototype.connect = function connect() { + return this.connection.open(); + }; + Consumer.prototype.disconnect = function disconnect() { + return this.connection.close({ + allowReconnect: false + }); + }; + Consumer.prototype.ensureActiveConnection = function ensureActiveConnection() { + if (!this.connection.isActive()) { + return this.connection.open(); + } + }; + return Consumer; + }(); + function createConsumer(url) { + if (url == null) { + var urlConfig = getConfig("url"); + url = urlConfig ? urlConfig : INTERNAL.default_mount_path; + } + return new Consumer(createWebSocketURL(url)); + } + function getConfig(name) { + var element = document.head.querySelector("meta[name='action-cable-" + name + "']"); + return element ? element.getAttribute("content") : undefined; + } + function createWebSocketURL(url) { + if (url && !/^wss?:/i.test(url)) { + var a = document.createElement("a"); + a.href = url; + a.href = a.href; + a.protocol = a.protocol.replace("http", "ws"); + return a.href; + } else { + return url; + } + } + exports.Connection = Connection; + exports.ConnectionMonitor = ConnectionMonitor; + exports.Consumer = Consumer; + exports.INTERNAL = INTERNAL; + exports.Subscription = Subscription; + exports.Subscriptions = Subscriptions; + exports.adapters = adapters; + exports.logger = logger; + exports.createConsumer = createConsumer; + exports.getConfig = getConfig; + exports.createWebSocketURL = createWebSocketURL; + Object.defineProperty(exports, "__esModule", { + value: true + }); +}); diff --git a/actioncable/app/javascript/action_cable/adapters.js b/actioncable/app/javascript/action_cable/adapters.js new file mode 100644 index 0000000000..9ba6d338ee --- /dev/null +++ b/actioncable/app/javascript/action_cable/adapters.js @@ -0,0 +1,4 @@ +export default { + logger: window.console, + WebSocket: window.WebSocket +} diff --git a/actioncable/app/javascript/action_cable/connection.js b/actioncable/app/javascript/action_cable/connection.js new file mode 100644 index 0000000000..b2910cb2a6 --- /dev/null +++ b/actioncable/app/javascript/action_cable/connection.js @@ -0,0 +1,161 @@ +import adapters from "./adapters" +import ConnectionMonitor from "./connection_monitor" +import INTERNAL from "./internal" +import logger from "./logger" + +// Encapsulate the cable connection held by the consumer. This is an internal class not intended for direct user manipulation. + +const {message_types, protocols} = INTERNAL +const supportedProtocols = protocols.slice(0, protocols.length - 1) + +const indexOf = [].indexOf + +class Connection { + constructor(consumer) { + this.open = this.open.bind(this) + this.consumer = consumer + this.subscriptions = this.consumer.subscriptions + this.monitor = new ConnectionMonitor(this) + this.disconnected = true + } + + send(data) { + if (this.isOpen()) { + this.webSocket.send(JSON.stringify(data)) + return true + } else { + return false + } + } + + open() { + if (this.isActive()) { + logger.log(`Attempted to open WebSocket, but existing socket is ${this.getState()}`) + return false + } else { + logger.log(`Opening WebSocket, current state is ${this.getState()}, subprotocols: ${protocols}`) + if (this.webSocket) { this.uninstallEventHandlers() } + this.webSocket = new adapters.WebSocket(this.consumer.url, protocols) + this.installEventHandlers() + this.monitor.start() + return true + } + } + + close({allowReconnect} = {allowReconnect: true}) { + if (!allowReconnect) { this.monitor.stop() } + if (this.isActive()) { return (this.webSocket ? this.webSocket.close() : undefined) } + } + + reopen() { + logger.log(`Reopening WebSocket, current state is ${this.getState()}`) + if (this.isActive()) { + try { + return this.close() + } catch (error) { + logger.log("Failed to reopen WebSocket", error) + } + finally { + logger.log(`Reopening WebSocket in ${this.constructor.reopenDelay}ms`) + setTimeout(this.open, this.constructor.reopenDelay) + } + } else { + return this.open() + } + } + + getProtocol() { + return (this.webSocket ? this.webSocket.protocol : undefined) + } + + isOpen() { + return this.isState("open") + } + + isActive() { + return this.isState("open", "connecting") + } + + // Private + + isProtocolSupported() { + return indexOf.call(supportedProtocols, this.getProtocol()) >= 0 + } + + isState(...states) { + return indexOf.call(states, this.getState()) >= 0 + } + + getState() { + if (this.webSocket) { + for (let state in adapters.WebSocket) { + if (adapters.WebSocket[state] === this.webSocket.readyState) { + return state.toLowerCase() + } + } + } + return null + } + + installEventHandlers() { + for (let eventName in this.events) { + const handler = this.events[eventName].bind(this) + this.webSocket[`on${eventName}`] = handler + } + } + + uninstallEventHandlers() { + for (let eventName in this.events) { + this.webSocket[`on${eventName}`] = function() {} + } + } + +} + +Connection.reopenDelay = 500 + +Connection.prototype.events = { + message(event) { + if (!this.isProtocolSupported()) { return } + const {identifier, message, reason, reconnect, type} = JSON.parse(event.data) + switch (type) { + case message_types.welcome: + this.monitor.recordConnect() + return this.subscriptions.reload() + case message_types.disconnect: + logger.log(`Disconnecting. Reason: ${reason}`) + return this.close({allowReconnect: reconnect}) + case message_types.ping: + return this.monitor.recordPing() + case message_types.confirmation: + return this.subscriptions.notify(identifier, "connected") + case message_types.rejection: + return this.subscriptions.reject(identifier) + default: + return this.subscriptions.notify(identifier, "received", message) + } + }, + + open() { + logger.log(`WebSocket onopen event, using '${this.getProtocol()}' subprotocol`) + this.disconnected = false + if (!this.isProtocolSupported()) { + logger.log("Protocol is unsupported. Stopping monitor and disconnecting.") + return this.close({allowReconnect: false}) + } + }, + + close(event) { + logger.log("WebSocket onclose event") + if (this.disconnected) { return } + this.disconnected = true + this.monitor.recordDisconnect() + return this.subscriptions.notifyAll("disconnected", {willAttemptReconnect: this.monitor.isRunning()}) + }, + + error() { + logger.log("WebSocket onerror event") + } +} + +export default Connection diff --git a/actioncable/app/javascript/action_cable/connection_monitor.js b/actioncable/app/javascript/action_cable/connection_monitor.js new file mode 100644 index 0000000000..f0e75ae137 --- /dev/null +++ b/actioncable/app/javascript/action_cable/connection_monitor.js @@ -0,0 +1,126 @@ +import logger from "./logger" + +// Responsible for ensuring the cable connection is in good health by validating the heartbeat pings sent from the server, and attempting +// revival reconnections if things go astray. Internal class, not intended for direct user manipulation. + +const now = () => new Date().getTime() + +const secondsSince = time => (now() - time) / 1000 + +const clamp = (number, min, max) => Math.max(min, Math.min(max, number)) + +class ConnectionMonitor { + constructor(connection) { + this.visibilityDidChange = this.visibilityDidChange.bind(this) + this.connection = connection + this.reconnectAttempts = 0 + } + + start() { + if (!this.isRunning()) { + this.startedAt = now() + delete this.stoppedAt + this.startPolling() + document.addEventListener("visibilitychange", this.visibilityDidChange) + logger.log(`ConnectionMonitor started. pollInterval = ${this.getPollInterval()} ms`) + } + } + + stop() { + if (this.isRunning()) { + this.stoppedAt = now() + this.stopPolling() + document.removeEventListener("visibilitychange", this.visibilityDidChange) + logger.log("ConnectionMonitor stopped") + } + } + + isRunning() { + return this.startedAt && !this.stoppedAt + } + + recordPing() { + this.pingedAt = now() + } + + recordConnect() { + this.reconnectAttempts = 0 + this.recordPing() + delete this.disconnectedAt + logger.log("ConnectionMonitor recorded connect") + } + + recordDisconnect() { + this.disconnectedAt = now() + logger.log("ConnectionMonitor recorded disconnect") + } + + // Private + + startPolling() { + this.stopPolling() + this.poll() + } + + stopPolling() { + clearTimeout(this.pollTimeout) + } + + poll() { + this.pollTimeout = setTimeout(() => { + this.reconnectIfStale() + this.poll() + } + , this.getPollInterval()) + } + + getPollInterval() { + const {min, max, multiplier} = this.constructor.pollInterval + const interval = multiplier * Math.log(this.reconnectAttempts + 1) + return Math.round(clamp(interval, min, max) * 1000) + } + + reconnectIfStale() { + if (this.connectionIsStale()) { + logger.log(`ConnectionMonitor detected stale connection. reconnectAttempts = ${this.reconnectAttempts}, pollInterval = ${this.getPollInterval()} ms, time disconnected = ${secondsSince(this.disconnectedAt)} s, stale threshold = ${this.constructor.staleThreshold} s`) + this.reconnectAttempts++ + if (this.disconnectedRecently()) { + logger.log("ConnectionMonitor skipping reopening recent disconnect") + } else { + logger.log("ConnectionMonitor reopening") + this.connection.reopen() + } + } + } + + connectionIsStale() { + return secondsSince(this.pingedAt ? this.pingedAt : this.startedAt) > this.constructor.staleThreshold + } + + disconnectedRecently() { + return this.disconnectedAt && (secondsSince(this.disconnectedAt) < this.constructor.staleThreshold) + } + + visibilityDidChange() { + if (document.visibilityState === "visible") { + setTimeout(() => { + if (this.connectionIsStale() || !this.connection.isOpen()) { + logger.log(`ConnectionMonitor reopening stale connection on visibilitychange. visbilityState = ${document.visibilityState}`) + this.connection.reopen() + } + } + , 200) + } + } + +} + +ConnectionMonitor.pollInterval = { + min: 3, + max: 30, + multiplier: 5 +} + +ConnectionMonitor.staleThreshold = 6 // Server::Connections::BEAT_INTERVAL * 2 (missed two pings) + +export default ConnectionMonitor diff --git a/actioncable/app/javascript/action_cable/consumer.js b/actioncable/app/javascript/action_cable/consumer.js new file mode 100644 index 0000000000..e8440f39f5 --- /dev/null +++ b/actioncable/app/javascript/action_cable/consumer.js @@ -0,0 +1,54 @@ +import Connection from "./connection" +import Subscriptions from "./subscriptions" + +// The ActionCable.Consumer establishes the connection to a server-side Ruby Connection object. Once established, +// the ActionCable.ConnectionMonitor will ensure that its properly maintained through heartbeats and checking for stale updates. +// The Consumer instance is also the gateway to establishing subscriptions to desired channels through the #createSubscription +// method. +// +// The following example shows how this can be setup: +// +// App = {} +// App.cable = ActionCable.createConsumer("ws://example.com/accounts/1") +// App.appearance = App.cable.subscriptions.create("AppearanceChannel") +// +// For more details on how you'd configure an actual channel subscription, see ActionCable.Subscription. +// +// When a consumer is created, it automatically connects with the server. +// +// To disconnect from the server, call +// +// App.cable.disconnect() +// +// and to restart the connection: +// +// App.cable.connect() +// +// Any channel subscriptions which existed prior to disconnecting will +// automatically resubscribe. + +export default class Consumer { + constructor(url) { + this.url = url + this.subscriptions = new Subscriptions(this) + this.connection = new Connection(this) + } + + send(data) { + return this.connection.send(data) + } + + connect() { + return this.connection.open() + } + + disconnect() { + return this.connection.close({allowReconnect: false}) + } + + ensureActiveConnection() { + if (!this.connection.isActive()) { + return this.connection.open() + } + } +} diff --git a/actioncable/app/javascript/action_cable/index.js b/actioncable/app/javascript/action_cable/index.js new file mode 100644 index 0000000000..9f41c14e94 --- /dev/null +++ b/actioncable/app/javascript/action_cable/index.js @@ -0,0 +1,45 @@ +import Connection from "./connection" +import ConnectionMonitor from "./connection_monitor" +import Consumer from "./consumer" +import INTERNAL from "./internal" +import Subscription from "./subscription" +import Subscriptions from "./subscriptions" +import adapters from "./adapters" +import logger from "./logger" + +export { + Connection, + ConnectionMonitor, + Consumer, + INTERNAL, + Subscription, + Subscriptions, + adapters, + logger, +} + +export function createConsumer(url) { + if (url == null) { + const urlConfig = getConfig("url") + url = (urlConfig ? urlConfig : INTERNAL.default_mount_path) + } + return new Consumer(createWebSocketURL(url)) +} + +export function getConfig(name) { + const element = document.head.querySelector(`meta[name='action-cable-${name}']`) + return (element ? element.getAttribute("content") : undefined) +} + +export function createWebSocketURL(url) { + if (url && !/^wss?:/i.test(url)) { + const a = document.createElement("a") + a.href = url + // Fix populating Location properties in IE. Otherwise, protocol will be blank. + a.href = a.href + a.protocol = a.protocol.replace("http", "ws") + return a.href + } else { + return url + } +} diff --git a/actioncable/app/javascript/action_cable/logger.js b/actioncable/app/javascript/action_cable/logger.js new file mode 100644 index 0000000000..ef4661ead1 --- /dev/null +++ b/actioncable/app/javascript/action_cable/logger.js @@ -0,0 +1,10 @@ +import adapters from "./adapters" + +export default { + log(...messages) { + if (this.enabled) { + messages.push(Date.now()) + adapters.logger.log("[ActionCable]", ...messages) + } + }, +} diff --git a/actioncable/app/javascript/action_cable/subscription.js b/actioncable/app/javascript/action_cable/subscription.js new file mode 100644 index 0000000000..7de08f93b3 --- /dev/null +++ b/actioncable/app/javascript/action_cable/subscription.js @@ -0,0 +1,89 @@ +// A new subscription is created through the ActionCable.Subscriptions instance available on the consumer. +// It provides a number of callbacks and a method for calling remote procedure calls on the corresponding +// Channel instance on the server side. +// +// An example demonstrates the basic functionality: +// +// App.appearance = App.cable.subscriptions.create("AppearanceChannel", { +// connected() { +// // Called once the subscription has been successfully completed +// }, +// +// disconnected({ willAttemptReconnect: boolean }) { +// // Called when the client has disconnected with the server. +// // The object will have an `willAttemptReconnect` property which +// // says whether the client has the intention of attempting +// // to reconnect. +// }, +// +// appear() { +// this.perform('appear', {appearing_on: this.appearingOn()}) +// }, +// +// away() { +// this.perform('away') +// }, +// +// appearingOn() { +// $('main').data('appearing-on') +// } +// }) +// +// The methods #appear and #away forward their intent to the remote AppearanceChannel instance on the server +// by calling the `perform` method with the first parameter being the action (which maps to AppearanceChannel#appear/away). +// The second parameter is a hash that'll get JSON encoded and made available on the server in the data parameter. +// +// This is how the server component would look: +// +// class AppearanceChannel < ApplicationActionCable::Channel +// def subscribed +// current_user.appear +// end +// +// def unsubscribed +// current_user.disappear +// end +// +// def appear(data) +// current_user.appear on: data['appearing_on'] +// end +// +// def away +// current_user.away +// end +// end +// +// The "AppearanceChannel" name is automatically mapped between the client-side subscription creation and the server-side Ruby class name. +// The AppearanceChannel#appear/away public methods are exposed automatically to client-side invocation through the perform method. + +const extend = function(object, properties) { + if (properties != null) { + for (let key in properties) { + const value = properties[key] + object[key] = value + } + } + return object +} + +export default class Subscription { + constructor(consumer, params = {}, mixin) { + this.consumer = consumer + this.identifier = JSON.stringify(params) + extend(this, mixin) + } + + // Perform a channel action with the optional data passed as an attribute + perform(action, data = {}) { + data.action = action + return this.send(data) + } + + send(data) { + return this.consumer.send({command: "message", identifier: this.identifier, data: JSON.stringify(data)}) + } + + unsubscribe() { + return this.consumer.subscriptions.remove(this) + } +} diff --git a/actioncable/app/javascript/action_cable/subscriptions.js b/actioncable/app/javascript/action_cable/subscriptions.js new file mode 100644 index 0000000000..867cafb407 --- /dev/null +++ b/actioncable/app/javascript/action_cable/subscriptions.js @@ -0,0 +1,86 @@ +import Subscription from "./subscription" + +// Collection class for creating (and internally managing) channel subscriptions. The only method intended to be triggered by the user +// us ActionCable.Subscriptions#create, and it should be called through the consumer like so: +// +// App = {} +// App.cable = ActionCable.createConsumer("ws://example.com/accounts/1") +// App.appearance = App.cable.subscriptions.create("AppearanceChannel") +// +// For more details on how you'd configure an actual channel subscription, see ActionCable.Subscription. + +export default class Subscriptions { + constructor(consumer) { + this.consumer = consumer + this.subscriptions = [] + } + + create(channelName, mixin) { + const channel = channelName + const params = typeof channel === "object" ? channel : {channel} + const subscription = new Subscription(this.consumer, params, mixin) + return this.add(subscription) + } + + // Private + + add(subscription) { + this.subscriptions.push(subscription) + this.consumer.ensureActiveConnection() + this.notify(subscription, "initialized") + this.sendCommand(subscription, "subscribe") + return subscription + } + + remove(subscription) { + this.forget(subscription) + if (!this.findAll(subscription.identifier).length) { + this.sendCommand(subscription, "unsubscribe") + } + return subscription + } + + reject(identifier) { + return this.findAll(identifier).map((subscription) => { + this.forget(subscription) + this.notify(subscription, "rejected") + return subscription + }) + } + + forget(subscription) { + this.subscriptions = (this.subscriptions.filter((s) => s !== subscription)) + return subscription + } + + findAll(identifier) { + return this.subscriptions.filter((s) => s.identifier === identifier) + } + + reload() { + return this.subscriptions.map((subscription) => + this.sendCommand(subscription, "subscribe")) + } + + notifyAll(callbackName, ...args) { + return this.subscriptions.map((subscription) => + this.notify(subscription, callbackName, ...args)) + } + + notify(subscription, callbackName, ...args) { + let subscriptions + if (typeof subscription === "string") { + subscriptions = this.findAll(subscription) + } else { + subscriptions = [subscription] + } + + return subscriptions.map((subscription) => + (typeof subscription[callbackName] === "function" ? subscription[callbackName](...args) : undefined)) + } + + sendCommand(subscription, command) { + const {identifier} = subscription + return this.consumer.send({command, identifier}) + } +} diff --git a/actioncable/bin/test b/actioncable/bin/test new file mode 100755 index 0000000000..c53377cc97 --- /dev/null +++ b/actioncable/bin/test @@ -0,0 +1,5 @@ +#!/usr/bin/env ruby +# frozen_string_literal: true + +COMPONENT_ROOT = File.expand_path("..", __dir__) +require_relative "../../tools/test" diff --git a/actioncable/karma.conf.js b/actioncable/karma.conf.js new file mode 100644 index 0000000000..845b38d74f --- /dev/null +++ b/actioncable/karma.conf.js @@ -0,0 +1,64 @@ +const config = { + browsers: ["ChromeHeadless"], + frameworks: ["qunit"], + files: [ + "test/javascript/compiled/test.js", + ], + + client: { + clearContext: false, + qunit: { + showUI: true + } + }, + + singleRun: true, + autoWatch: false, + + captureTimeout: 180000, + browserDisconnectTimeout: 180000, + browserDisconnectTolerance: 3, + browserNoActivityTimeout: 300000, +} + +if (process.env.CI) { + config.customLaunchers = { + sl_chrome: sauce("chrome", 70), + sl_ff: sauce("firefox", 63), + sl_safari: sauce("safari", 12.0, "macOS 10.13"), + sl_edge: sauce("microsoftedge", 17.17134, "Windows 10"), + sl_ie_11: sauce("internet explorer", 11, "Windows 8.1"), + } + + config.browsers = Object.keys(config.customLaunchers) + config.reporters = ["dots", "saucelabs"] + + config.sauceLabs = { + testName: "ActionCable JS Client", + retryLimit: 3, + build: buildId(), + } + + function sauce(browserName, version, platform) { + const options = { + base: "SauceLabs", + browserName: browserName.toString(), + version: version.toString(), + } + if (platform) { + options.platform = platform.toString() + } + return options + } + + function buildId() { + const { TRAVIS_BUILD_NUMBER, TRAVIS_BUILD_ID } = process.env + return TRAVIS_BUILD_NUMBER && TRAVIS_BUILD_ID + ? `TRAVIS #${TRAVIS_BUILD_NUMBER} (${TRAVIS_BUILD_ID})` + : "" + } +} + +module.exports = function(karmaConfig) { + karmaConfig.set(config) +} diff --git a/actioncable/lib/action_cable.rb b/actioncable/lib/action_cable.rb new file mode 100644 index 0000000000..ad5fd43155 --- /dev/null +++ b/actioncable/lib/action_cable.rb @@ -0,0 +1,62 @@ +# frozen_string_literal: true + +#-- +# Copyright (c) 2015-2019 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 = { + message_types: { + welcome: "welcome", + disconnect: "disconnect", + ping: "ping", + confirmation: "confirm_subscription", + rejection: "reject_subscription" + }, + disconnect_reasons: { + unauthorized: "unauthorized", + invalid_request: "invalid_request", + server_restart: "server_restart" + }, + default_mount_path: "/cable", + protocols: ["actioncable-v1-json", "actioncable-unsupported"].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 + autoload :TestHelper + autoload :TestCase +end diff --git a/actioncable/lib/action_cable/channel.rb b/actioncable/lib/action_cable/channel.rb new file mode 100644 index 0000000000..d5118b9dc9 --- /dev/null +++ b/actioncable/lib/action_cable/channel.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +module ActionCable + module Channel + extend ActiveSupport::Autoload + + eager_autoload do + autoload :Base + autoload :Broadcasting + autoload :Callbacks + autoload :Naming + autoload :PeriodicTimers + autoload :Streams + autoload :TestCase + 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..ad0d3685cd --- /dev/null +++ b/actioncable/lib/action_cable/channel/base.rb @@ -0,0 +1,309 @@ +# frozen_string_literal: true + +require "set" +require "active_support/rescuable" + +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 RESTful + # constraint form for their actions. Instead, Action Cable operates through a + # remote-procedure call model. You can declare any public method on the + # channel (optionally taking a <tt>data</tt> argument), and this method is + # automatically exposed as callable to the client. + # + # Example: + # + # 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, the subscribed and unsubscribed methods are not callable methods, as they + # were already declared in ActionCable::Channel::Base, but <tt>#appear</tt> + # and <tt>#away</tt> are. <tt>#generate_connection_token</tt> is also not + # callable, since it's a private method. You'll see that appear accepts a data + # parameter, which it then uses as part of its model call. <tt>#away</tt> + # does not, since it's simply a trigger action. + # + # Also note that in this example, <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 + include ActiveSupport::Rescuable + + 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 + + private + # 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! # :doc: + @action_methods = nil + end + + # Refresh the cached action_methods when a new action_method is added. + def method_added(name) # :doc: + 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. + # + # The counter starts at 1 because it's awaiting a call to #subscribe_to_channel + @defer_subscription_confirmation_counter = Concurrent::AtomicFixnum.new(1) + + @reject_subscription = nil + @subscription_confirmation_sent = nil + + delegate_connection_identifiers + 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) + payload = { channel_class: self.class.name, action: action, data: data } + ActiveSupport::Notifications.instrument("perform_action.action_cable", payload) do + dispatch_action(action, data) + end + else + logger.error "Unable to process #{action_signature(action, data)}" + end + end + + # This method is called after subscription has been added to the connection + # and confirms or rejects the subscription. + def subscribe_to_channel + run_callbacks :subscribe do + subscribed + end + + reject_subscription if subscription_rejected? + ensure_confirmation_sent + end + + # Called by the cable connection when it's cut, so the channel has a chance to cleanup with callbacks. + # This method is not intended to be called directly by the user. Instead, overwrite the #unsubscribed callback. + def unsubscribe_from_channel # :nodoc: + run_callbacks :unsubscribe do + unsubscribed + end + end + + private + # 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 # :doc: + # Override in subclasses + end + + # Called once a consumer has cut its cable connection. Can be used for cleaning up connections or marking + # users as offline or the like. + def unsubscribed # :doc: + # 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) # :doc: + status = "#{self.class.name} transmitting #{data.inspect.truncate(300)}" + status += " (via #{via})" if via + logger.debug(status) + + payload = { channel_class: self.class.name, data: data, via: via } + ActiveSupport::Notifications.instrument("transmit.action_cable", payload) do + connection.transmit identifier: @identifier, message: data + end + end + + def ensure_confirmation_sent # :doc: + return if subscription_rejected? + @defer_subscription_confirmation_counter.decrement + transmit_subscription_confirmation unless defer_subscription_confirmation? + end + + def defer_subscription_confirmation! # :doc: + @defer_subscription_confirmation_counter.increment + end + + def defer_subscription_confirmation? # :doc: + @defer_subscription_confirmation_counter.value > 0 + end + + def subscription_confirmation_sent? # :doc: + @subscription_confirmation_sent + end + + def reject # :doc: + @reject_subscription = true + end + + def subscription_rejected? # :doc: + @reject_subscription + end + + def delegate_connection_identifiers + connection.identifiers.each do |identifier| + define_singleton_method(identifier) do + connection.send(identifier) + end + 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) unless subscription_rejected? + 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 + rescue Exception => exception + rescue_with_handler(exception) || raise + 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" + + ActiveSupport::Notifications.instrument("transmit_subscription_confirmation.action_cable", channel_class: self.class.name) do + connection.transmit identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:confirmation] + @subscription_confirmation_sent = true + end + 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" + + ActiveSupport::Notifications.instrument("transmit_subscription_rejection.action_cable", channel_class: self.class.name) do + connection.transmit identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:rejection] + end + 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..9a96720f4a --- /dev/null +++ b/actioncable/lib/action_cable/channel/broadcasting.rb @@ -0,0 +1,31 @@ +# frozen_string_literal: true + +require "active_support/core_ext/object/to_param" + +module ActionCable + module Channel + module Broadcasting + extend ActiveSupport::Concern + + delegate :broadcasting_for, to: :class + + module ClassMethods + # 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..e4cb19b26a --- /dev/null +++ b/actioncable/lib/action_cable/channel/callbacks.rb @@ -0,0 +1,37 @@ +# frozen_string_literal: true + +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 + + module ClassMethods + 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..9c324a2a53 --- /dev/null +++ b/actioncable/lib/action_cable/channel/naming.rb @@ -0,0 +1,25 @@ +# frozen_string_literal: true + +module ActionCable + module Channel + module Naming + extend ActiveSupport::Concern + + module ClassMethods + # 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' + # FooChats::BarAppearancesChannel.channel_name # => 'foo_chats:bar_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..830b3efa3c --- /dev/null +++ b/actioncable/lib/action_cable/channel/periodic_timers.rb @@ -0,0 +1,78 @@ +# frozen_string_literal: true + +module ActionCable + module Channel + module PeriodicTimers + extend ActiveSupport::Concern + + included do + class_attribute :periodic_timers, instance_reader: false, default: [] + + after_subscribe :start_periodic_timers + after_unsubscribe :stop_periodic_timers + end + + module ClassMethods + # Periodically performs a task on the channel, like updating an online + # user counter, polling a backend for new status messages, sending + # regular "heartbeat" messages, or doing some internal work and giving + # progress updates. + # + # Pass a method name or lambda argument or provide a block to call. + # Specify the calling period in seconds using the <tt>every:</tt> + # keyword argument. + # + # periodically :transmit_progress, every: 5.seconds + # + # periodically every: 3.minutes do + # transmit action: :update_count, count: current_count + # end + # + def periodically(callback_or_method_name = nil, every:, &block) + callback = + if block_given? + raise ArgumentError, "Pass a block or provide a callback arg, not both" if callback_or_method_name + block + else + case callback_or_method_name + when Proc + callback_or_method_name + when Symbol + -> { __send__ callback_or_method_name } + else + raise ArgumentError, "Expected a Symbol method name or a Proc, got #{callback_or_method_name.inspect}" + end + end + + unless every.kind_of?(Numeric) && every > 0 + raise ArgumentError, "Expected every: to be a positive number of seconds, got #{every.inspect}" + end + + 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 << start_periodic_timer(callback, every: options.fetch(:every)) + end + end + + def start_periodic_timer(callback, every:) + connection.server.event_loop.timer every do + connection.worker_pool.async_exec self, connection: connection, &callback + end + end + + def stop_periodic_timers + active_periodic_timers.each { |timer| timer.shutdown } + active_periodic_timers.clear + 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..81c2c38064 --- /dev/null +++ b/actioncable/lib/action_cable/channel/streams.rb @@ -0,0 +1,176 @@ +# frozen_string_literal: true + +module ActionCable + module Channel + # 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, even 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 + # 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 + # + # Based on the above example, the subscribers of this channel will get whatever data is put into the, + # let's say, <tt>comments_for_45</tt> broadcasting as soon as it's put there. + # + # An example broadcasting for this channel looks like so: + # + # ActionCable.server.broadcast "comments_for_45", author: 'DHH', content: 'Rails is just swell' + # + # If you have a stream that is related to a model, then the broadcasting used can be generated from the model and channel. + # The following example would subscribe to a broadcasting like <tt>comments:Z2lkOi8vVGVzdEFwcC9Qb3N0LzE</tt>. + # + # 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 also supply a callback that lets you alter what is sent out. + # The below example shows how you can use this to provide performance introspection in the process: + # + # class ChatChannel < ApplicationCable::Channel + # def subscribed + # @room = Chat::Room[params[:room_number]] + # + # stream_for @room, coder: ActiveSupport::JSON do |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. + # Pass <tt>coder: ActiveSupport::JSON</tt> to decode messages as JSON before passing to the callback. + # Defaults to <tt>coder: nil</tt> which does no decoding, passes raw messages. + def stream_from(broadcasting, callback = nil, coder: nil, &block) + broadcasting = String(broadcasting) + + # Don't send the confirmation until pubsub#subscribe is successful + defer_subscription_confirmation! + + # Build a stream handler by wrapping the user-provided callback with + # a decoder or defaulting to a JSON-decoding retransmitter. + handler = worker_pool_stream_handler(broadcasting, callback || block, coder: coder) + streams << [ broadcasting, handler ] + + connection.server.event_loop.post do + pubsub.subscribe(broadcasting, handler, lambda do + ensure_confirmation_sent + 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. + # + # Pass <tt>coder: ActiveSupport::JSON</tt> to decode messages as JSON before passing to the callback. + # Defaults to <tt>coder: nil</tt> which does no decoding, passes raw messages. + def stream_for(model, callback = nil, coder: nil, &block) + stream_from(broadcasting_for([ channel_name, model ]), callback || block, coder: coder) + end + + # 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 + + # Always wrap the outermost handler to invoke the user handler on the + # worker pool rather than blocking the event loop. + def worker_pool_stream_handler(broadcasting, user_handler, coder: nil) + handler = stream_handler(broadcasting, user_handler, coder: coder) + + -> message do + connection.worker_pool.async_invoke handler, :call, message, connection: connection + end + end + + # May be overridden to add instrumentation, logging, specialized error + # handling, or other forms of handler decoration. + # + # TODO: Tests demonstrating this. + def stream_handler(broadcasting, user_handler, coder: nil) + if user_handler + stream_decoder user_handler, coder: coder + else + default_stream_handler broadcasting, coder: coder + end + end + + # May be overridden to change the default stream handling behavior + # which decodes JSON and transmits to the client. + # + # TODO: Tests demonstrating this. + # + # TODO: Room for optimization. Update transmit API to be coder-aware + # so we can no-op when pubsub and connection are both JSON-encoded. + # Then we can skip decode+encode if we're just proxying messages. + def default_stream_handler(broadcasting, coder:) + coder ||= ActiveSupport::JSON + stream_transmitter stream_decoder(coder: coder), broadcasting: broadcasting + end + + def stream_decoder(handler = identity_handler, coder:) + if coder + -> message { handler.(coder.decode(message)) } + else + handler + end + end + + def stream_transmitter(handler = identity_handler, broadcasting:) + via = "streamed from #{broadcasting}" + + -> (message) do + transmit handler.(message), via: via + end + end + + def identity_handler + -> message { message } + end + end + end +end diff --git a/actioncable/lib/action_cable/channel/test_case.rb b/actioncable/lib/action_cable/channel/test_case.rb new file mode 100644 index 0000000000..c4cf0ac0e7 --- /dev/null +++ b/actioncable/lib/action_cable/channel/test_case.rb @@ -0,0 +1,312 @@ +# frozen_string_literal: true + +require "active_support" +require "active_support/test_case" +require "active_support/core_ext/hash/indifferent_access" +require "json" + +module ActionCable + module Channel + class NonInferrableChannelError < ::StandardError + def initialize(name) + super "Unable to determine the channel to test from #{name}. " + + "You'll need to specify it using `tests YourChannel` in your " + + "test case definition." + end + end + + # Stub `stream_from` to track streams for the channel. + # Add public aliases for `subscription_confirmation_sent?` and + # `subscription_rejected?`. + module ChannelStub + def confirmed? + subscription_confirmation_sent? + end + + def rejected? + subscription_rejected? + end + + def stream_from(broadcasting, *) + streams << broadcasting + end + + def stop_all_streams + @_streams = [] + end + + def streams + @_streams ||= [] + end + + # Make periodic timers no-op + def start_periodic_timers; end + alias stop_periodic_timers start_periodic_timers + end + + class ConnectionStub + attr_reader :transmissions, :identifiers, :subscriptions, :logger + + def initialize(identifiers = {}) + @transmissions = [] + + identifiers.each do |identifier, val| + define_singleton_method(identifier) { val } + end + + @subscriptions = ActionCable::Connection::Subscriptions.new(self) + @identifiers = identifiers.keys + @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new) + end + + def transmit(cable_message) + transmissions << cable_message.with_indifferent_access + end + end + + # Superclass for Action Cable channel functional tests. + # + # == Basic example + # + # Functional tests are written as follows: + # 1. First, one uses the +subscribe+ method to simulate subscription creation. + # 2. Then, one asserts whether the current state is as expected. "State" can be anything: + # transmitted messages, subscribed streams, etc. + # + # For example: + # + # class ChatChannelTest < ActionCable::Channel::TestCase + # def test_subscribed_with_room_number + # # Simulate a subscription creation + # subscribe room_number: 1 + # + # # Asserts that the subscription was successfully created + # assert subscription.confirmed? + # + # # Asserts that the channel subscribes connection to a stream + # assert_has_stream "chat_1" + # + # # Asserts that the channel subscribes connection to a specific + # # stream created for a model + # assert_has_stream_for Room.find(1) + # end + # + # def test_does_not_stream_with_incorrect_room_number + # subscribe room_number: -1 + # + # # Asserts that not streams was started + # assert_no_streams + # end + # + # def test_does_not_subscribe_without_room_number + # subscribe + # + # # Asserts that the subscription was rejected + # assert subscription.rejected? + # end + # end + # + # You can also perform actions: + # def test_perform_speak + # subscribe room_number: 1 + # + # perform :speak, message: "Hello, Rails!" + # + # assert_equal "Hello, Rails!", transmissions.last["text"] + # end + # + # == Special methods + # + # ActionCable::Channel::TestCase will also automatically provide the following instance + # methods for use in the tests: + # + # <b>connection</b>:: + # An ActionCable::Channel::ConnectionStub, representing the current HTTP connection. + # <b>subscription</b>:: + # An instance of the current channel, created when you call `subscribe`. + # <b>transmissions</b>:: + # A list of all messages that have been transmitted into the channel. + # + # + # == Channel is automatically inferred + # + # ActionCable::Channel::TestCase will automatically infer the channel under test + # from the test class name. If the channel cannot be inferred from the test + # class name, you can explicitly set it with +tests+. + # + # class SpecialEdgeCaseChannelTest < ActionCable::Channel::TestCase + # tests SpecialChannel + # end + # + # == Specifying connection identifiers + # + # You need to set up your connection manually to provide values for the identifiers. + # To do this just use: + # + # stub_connection(user: users[:john]) + # + # == Testing broadcasting + # + # ActionCable::Channel::TestCase enhances ActionCable::TestHelper assertions (e.g. + # +assert_broadcasts+) to handle broadcasting to models: + # + # + # # in your channel + # def speak(data) + # broadcast_to room, text: data["message"] + # end + # + # def test_speak + # subscribe room_id: rooms[:chat].id + # + # assert_broadcasts_on(rooms[:chat], text: "Hello, Rails!") do + # perform :speak, message: "Hello, Rails!" + # end + # end + class TestCase < ActiveSupport::TestCase + module Behavior + extend ActiveSupport::Concern + + include ActiveSupport::Testing::ConstantLookup + include ActionCable::TestHelper + + CHANNEL_IDENTIFIER = "test_stub" + + included do + class_attribute :_channel_class + + attr_reader :connection, :subscription + + ActiveSupport.run_load_hooks(:action_cable_channel_test_case, self) + end + + module ClassMethods + def tests(channel) + case channel + when String, Symbol + self._channel_class = channel.to_s.camelize.constantize + when Module + self._channel_class = channel + else + raise NonInferrableChannelError.new(channel) + end + end + + def channel_class + if channel = self._channel_class + channel + else + tests determine_default_channel(name) + end + end + + def determine_default_channel(name) + channel = determine_constant_from_test_name(name) do |constant| + Class === constant && constant < ActionCable::Channel::Base + end + raise NonInferrableChannelError.new(name) if channel.nil? + channel + end + end + + # Setup test connection with the specified identifiers: + # + # class ApplicationCable < ActionCable::Connection::Base + # identified_by :user, :token + # end + # + # stub_connection(user: users[:john], token: 'my-secret-token') + def stub_connection(identifiers = {}) + @connection = ConnectionStub.new(identifiers) + end + + # Subscribe to the channel under test. Optionally pass subscription parameters as a Hash. + def subscribe(params = {}) + @connection ||= stub_connection + @subscription = self.class.channel_class.new(connection, CHANNEL_IDENTIFIER, params.with_indifferent_access) + @subscription.singleton_class.include(ChannelStub) + @subscription.subscribe_to_channel + @subscription + end + + # Unsubscribe the subscription under test. + def unsubscribe + check_subscribed! + subscription.unsubscribe_from_channel + end + + # Perform action on a channel. + # + # NOTE: Must be subscribed. + def perform(action, data = {}) + check_subscribed! + subscription.perform_action(data.stringify_keys.merge("action" => action.to_s)) + end + + # Returns messages transmitted into channel + def transmissions + # Return only directly sent message (via #transmit) + connection.transmissions.map { |data| data["message"] }.compact + end + + # Enhance TestHelper assertions to handle non-String + # broadcastings + def assert_broadcasts(stream_or_object, *args) + super(broadcasting_for(stream_or_object), *args) + end + + def assert_broadcast_on(stream_or_object, *args) + super(broadcasting_for(stream_or_object), *args) + end + + # Asserts that no streams have been started. + # + # def test_assert_no_started_stream + # subscribe + # assert_no_streams + # end + # + def assert_no_streams + assert subscription.streams.empty?, "No streams started was expected, but #{subscription.streams.count} found" + end + + # Asserts that the specified stream has been started. + # + # def test_assert_started_stream + # subscribe + # assert_has_stream 'messages' + # end + # + def assert_has_stream(stream) + assert subscription.streams.include?(stream), "Stream #{stream} has not been started" + end + + # Asserts that the specified stream for a model has started. + # + # def test_assert_started_stream_for + # subscribe id: 42 + # assert_has_stream_for User.find(42) + # end + # + def assert_has_stream_for(object) + assert_has_stream(broadcasting_for(object)) + end + + private + def check_subscribed! + raise "Must be subscribed!" if subscription.nil? || subscription.rejected? + end + + def broadcasting_for(stream_or_object) + return stream_or_object if stream_or_object.is_a?(String) + + self.class.channel_class.broadcasting_for( + [self.class.channel_class.channel_name, stream_or_object] + ) + end + end + + include Behavior + 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..804b89a707 --- /dev/null +++ b/actioncable/lib/action_cable/connection.rb @@ -0,0 +1,21 @@ +# frozen_string_literal: true + +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..aef3386f71 --- /dev/null +++ b/actioncable/lib/action_cable/connection/authorization.rb @@ -0,0 +1,15 @@ +# frozen_string_literal: true + +module ActionCable + module Connection + module Authorization + class UnauthorizedError < StandardError; end + + # Closes the WebSocket connection if it is open and returns a 404 "File not Found" response. + def reject_unauthorized_connection + logger.error "An unauthorized connection attempt was rejected" + raise UnauthorizedError + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb new file mode 100644 index 0000000000..0044afad98 --- /dev/null +++ b/actioncable/lib/action_cable/connection/base.rb @@ -0,0 +1,262 @@ +# frozen_string_literal: true + +require "action_dispatch" + +module ActionCable + module Connection + # For every WebSocket connection the Action Cable server accepts, a Connection object will be instantiated. This instance becomes the parent + # of all of the channel subscriptions that are created from there on. Incoming messages are then routed to these channel subscriptions + # based on an identifier sent by the Action Cable consumer. The Connection itself does not deal with any specific application logic beyond + # authentication and authorization. + # + # 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 + # + # private + # def find_verified_user + # User.find_by_identity(cookies.encrypted[:identity_id]) || + # reject_unauthorized_connection + # end + # end + # end + # + # 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. + # + # Finally, we add a tag to the connection-specific logger with the 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, :worker_pool, :protocol + delegate :event_loop, :pubsub, to: :server + + def initialize(server, env, coder: ActiveSupport::JSON) + @server, @env, @coder = server, env, coder + + @worker_pool = server.worker_pool + @logger = new_tagged_logger + + @websocket = ActionCable::Connection::WebSocket.new(env, self, 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 -- instead rely upon on the #connect (and #disconnect) callbacks. + def process #:nodoc: + logger.info started_request_message + + if websocket.possible? && allow_request_origin? + respond_to_successful_request + else + respond_to_invalid_request + end + end + + # Decodes WebSocket messages and dispatches them to subscribed channels. + # WebSocket message transfer encoding is always JSON. + def receive(websocket_message) #:nodoc: + send_async :dispatch_websocket_message, websocket_message + end + + def dispatch_websocket_message(websocket_message) #:nodoc: + if websocket.alive? + subscriptions.execute_command decode(websocket_message) + else + logger.error "Ignoring message processed after the WebSocket was closed: #{websocket_message.inspect})" + end + end + + def transmit(cable_message) # :nodoc: + websocket.transmit encode(cable_message) + end + + # Close the WebSocket connection. + def close(reason: nil, reconnect: true) + transmit( + type: ActionCable::INTERNAL[:message_types][:disconnect], + reason: reason, + reconnect: reconnect + ) + 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 <tt>identifier</tt>, <tt>started_at</tt>, <tt>subscriptions</tt>, and <tt>request_id</tt>. + # This can be returned by a health check against the connection. + def statistics + { + identifier: connection_identifier, + started_at: @started_at, + subscriptions: subscriptions.identifiers, + request_id: @env["action_dispatch.request_id"] + } + end + + def beat + transmit type: ActionCable::INTERNAL[:message_types][: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: + # log errors to make diagnosing socket errors easier + logger.error "WebSocket error occurred: #{message}" + end + + def on_close(reason, code) # :nodoc: + send_async :handle_close + end + + private + attr_reader :websocket + attr_reader :message_buffer + + # The request that initiated the WebSocket connection is available here. This gives access to the environment, cookies, etc. + def request # :doc: + @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 # :doc: + request.cookie_jar + end + + def encode(cable_message) + @coder.encode cable_message + end + + def decode(websocket_message) + @coder.decode websocket_message + end + + def handle_open + @protocol = websocket.protocol + connect if respond_to?(:connect) + subscribe_to_internal_channel + send_welcome_message + + message_buffer.process! + server.add_connection(self) + rescue ActionCable::Connection::Authorization::UnauthorizedError + close(reason: ActionCable::INTERNAL[:disconnect_reasons][:unauthorized], reconnect: false) if websocket.alive? + 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 send_welcome_message + # Send welcome message to the internal connection monitor channel. + # This ensures the connection monitor state is reset after a successful + # websocket connection. + transmit type: ActionCable::INTERNAL[:message_types][:welcome] + end + + def allow_request_origin? + return true if server.config.disable_request_forgery_protection + + proto = Rack::Request.new(env).ssl? ? "https" : "http" + if server.config.allow_same_origin_as_host && env["HTTP_ORIGIN"] == "#{proto}://#{env['HTTP_HOST']}" + true + elsif 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 + logger.info successful_request_message + websocket.rack_response + end + + def respond_to_invalid_request + close(reason: ActionCable::INTERNAL[:disconnect_reasons][:invalid_request]) if websocket.alive? + + logger.error invalid_request_message + logger.info finished_request_message + [ 404, { "Content-Type" => "text/plain" }, [ "Page not found" ] ] + end + + # Tags are declared in the server but computed in the connection. This allows us per-connection tailored tags. + def new_tagged_logger + TaggedLoggerProxy.new server.logger, + 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]" : "[non-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]" : "[non-WebSocket]", + request.ip, + Time.now.to_s ] + end + + def invalid_request_message + "Failed to upgrade to WebSocket (REQUEST_METHOD: %s, HTTP_CONNECTION: %s, HTTP_UPGRADE: %s)" % [ + env["REQUEST_METHOD"], env["HTTP_CONNECTION"], env["HTTP_UPGRADE"] + ] + end + + def successful_request_message + "Successfully upgraded to WebSocket (REQUEST_METHOD: %s, HTTP_CONNECTION: %s, HTTP_UPGRADE: %s)" % [ + env["REQUEST_METHOD"], env["HTTP_CONNECTION"], env["HTTP_UPGRADE"] + ] + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/client_socket.rb b/actioncable/lib/action_cable/connection/client_socket.rb new file mode 100644 index 0000000000..4b1964c4ae --- /dev/null +++ b/actioncable/lib/action_cable/connection/client_socket.rb @@ -0,0 +1,157 @@ +# frozen_string_literal: true + +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" + + false + end + + CONNECTING = 0 + OPEN = 1 + CLOSING = 2 + CLOSED = 3 + + attr_reader :env, :url + + def initialize(env, event_target, event_loop, protocols) + @env = env + @event_target = event_target + @event_loop = 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, protocols: protocols) + + @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(@event_loop, self) + end + + def start_driver + return if @driver.nil? || @driver_started + @stream.hijack_rack_socket + + if callback = @env["async.callback"] + callback.call([101, {}, @stream]) + end + + @driver_started = true + @driver.start + end + + def rack_response + start_driver + [ -1, {}, [] ] + end + + def write(data) + @stream.write(data) + rescue => e + emit_error e.message + 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 || (code >= 3000 && code <= 4999) + raise ArgumentError, "Failed to execute 'close' on WebSocket: " \ + "The code must be either 1000, or between 3000 and 4999. " \ + "#{code} is neither." + 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 + + def protocol + @driver.protocol + end + + private + def open + return unless @ready_state == CONNECTING + @ready_state = OPEN + + @event_target.on_open + end + + def receive_message(data) + return unless @ready_state == OPEN + + @event_target.on_message(data) + end + + def emit_error(message) + return if @ready_state >= CLOSING + + @event_target.on_error(message) + end + + def begin_close(reason, code) + return if @ready_state == CLOSED + @ready_state = CLOSING + @close_params = [reason, code] + + @stream.shutdown if @stream + finalize_close + end + + def finalize_close + return if @ready_state == CLOSED + @ready_state = CLOSED + + @event_target.on_close(*@close_params) + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/identification.rb b/actioncable/lib/action_cable/connection/identification.rb new file mode 100644 index 0000000000..cc544685dd --- /dev/null +++ b/actioncable/lib/action_cable/connection/identification.rb @@ -0,0 +1,47 @@ +# frozen_string_literal: true + +require "set" + +module ActionCable + module Connection + module Identification + extend ActiveSupport::Concern + + included do + class_attribute :identifiers, default: Set.new + end + + module ClassMethods + # 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..f03904137b --- /dev/null +++ b/actioncable/lib/action_cable/connection/internal_channel.rb @@ -0,0 +1,45 @@ +# frozen_string_literal: true + +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 decode(message) } + @_internal_subscriptions ||= [] + @_internal_subscriptions << [ internal_channel, callback ] + + server.event_loop.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| server.event_loop.post { pubsub.unsubscribe(channel, callback) } } + end + end + + def process_internal_message(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..965841b67e --- /dev/null +++ b/actioncable/lib/action_cable/connection/message_buffer.rb @@ -0,0 +1,54 @@ +# frozen_string_literal: true + +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. + class MessageBuffer # :nodoc: + 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 + + private + attr_reader :connection + attr_reader :buffered_messages + + def valid?(message) + message.is_a?(String) + end + + def receive(message) + connection.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..e658948a55 --- /dev/null +++ b/actioncable/lib/action_cable/connection/stream.rb @@ -0,0 +1,117 @@ +# frozen_string_literal: true + +require "thread" + +module ActionCable + module Connection + #-- + # This class is heavily based on faye-websocket-ruby + # + # Copyright (c) 2010-2015 James Coglan + class Stream # :nodoc: + def initialize(event_loop, socket) + @event_loop = event_loop + @socket_object = socket + @stream_send = socket.env["stream.send"] + + @rack_hijack_io = nil + @write_lock = Mutex.new + + @write_head = nil + @write_buffer = Queue.new + 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) + if @stream_send + return @stream_send.call(data) + end + + if @write_lock.try_lock + begin + if @write_head.nil? && @write_buffer.empty? + written = @rack_hijack_io.write_nonblock(data, exception: false) + + case written + when :wait_writable + # proceed below + when data.bytesize + return data.bytesize + else + @write_head = data.byteslice(written, data.bytesize) + @event_loop.writes_pending @rack_hijack_io + + return data.bytesize + end + end + ensure + @write_lock.unlock + end + end + + @write_buffer << data + @event_loop.writes_pending @rack_hijack_io + + data.bytesize + rescue EOFError, Errno::ECONNRESET + @socket_object.client_gone + end + + def flush_write_buffer + @write_lock.synchronize do + loop do + if @write_head.nil? + return true if @write_buffer.empty? + @write_head = @write_buffer.pop + end + + written = @rack_hijack_io.write_nonblock(@write_head, exception: false) + case written + when :wait_writable + return false + when @write_head.bytesize + @write_head = nil + else + @write_head = @write_head.byteslice(written, @write_head.bytesize) + return false + end + end + end + end + + def receive(data) + @socket_object.parse(data) + end + + def hijack_rack_socket + return unless @socket_object.env["rack.hijack"] + + # This should return the underlying io according to the SPEC: + @rack_hijack_io = @socket_object.env["rack.hijack"].call + # Retain existing behaviour if required: + @rack_hijack_io ||= @socket_object.env["rack.hijack_io"] + + @event_loop.attach(@rack_hijack_io, self) + end + + private + def clean_rack_hijack + return unless @rack_hijack_io + @event_loop.detach(@rack_hijack_io, self) + @rack_hijack_io = nil + end + end + end +end diff --git a/actioncable/lib/action_cable/connection/stream_event_loop.rb b/actioncable/lib/action_cable/connection/stream_event_loop.rb new file mode 100644 index 0000000000..d95afc50ba --- /dev/null +++ b/actioncable/lib/action_cable/connection/stream_event_loop.rb @@ -0,0 +1,136 @@ +# frozen_string_literal: true + +require "nio" +require "thread" + +module ActionCable + module Connection + class StreamEventLoop + def initialize + @nio = @executor = @thread = nil + @map = {} + @stopping = false + @todo = Queue.new + + @spawn_mutex = Mutex.new + end + + def timer(interval, &block) + Concurrent::TimerTask.new(execution_interval: interval, &block).tap(&:execute) + end + + def post(task = nil, &block) + task ||= block + + spawn + @executor << task + end + + def attach(io, stream) + @todo << lambda do + @map[io] = @nio.register(io, :r) + @map[io].value = stream + end + wakeup + end + + def detach(io, stream) + @todo << lambda do + @nio.deregister io + @map.delete io + io.close + end + wakeup + end + + def writes_pending(io) + @todo << lambda do + if monitor = @map[io] + monitor.interests = :rw + end + end + wakeup + end + + def stop + @stopping = true + wakeup if @nio + end + + private + def spawn + return if @thread && @thread.status + + @spawn_mutex.synchronize do + return if @thread && @thread.status + + @nio ||= NIO::Selector.new + + @executor ||= Concurrent::ThreadPoolExecutor.new( + min_threads: 1, + max_threads: 10, + max_queue: 0, + ) + + @thread = Thread.new { run } + + return true + 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 = monitor.value + + begin + if monitor.writable? + if stream.flush_write_buffer + monitor.interests = :r + end + next unless monitor.readable? + end + + incoming = io.read_nonblock(4096, exception: false) + case incoming + when :wait_readable + next + when nil + stream.close + else + stream.receive incoming + end + rescue + # We expect one of EOFError or Errno::ECONNRESET in + # normal operation (when the client goes away). But if + # 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..1ad8d05107 --- /dev/null +++ b/actioncable/lib/action_cable/connection/subscriptions.rb @@ -0,0 +1,79 @@ +# frozen_string_literal: true + +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. + class Subscriptions # :nodoc: + 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 + + return if subscriptions.key?(id_key) + + subscription_klass = id_options[:channel].safe_constantize + + if subscription_klass && ActionCable::Channel::Base >= subscription_klass + subscription = subscription_klass.new(connection, id_key, id_options) + subscriptions[id_key] = subscription + subscription.subscribe_to_channel + else + logger.error "Subscription class not found: #{id_options[:channel].inspect}" + end + end + + def remove(data) + logger.info "Unsubscribing from channel: #{data['identifier']}" + remove_subscription find(data) + 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| remove_subscription(channel) } + end + + private + attr_reader :connection, :subscriptions + 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..85831806a9 --- /dev/null +++ b/actioncable/lib/action_cable/connection/tagged_logger_proxy.rb @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +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 + + private + def log(type, message) # :doc: + 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..31f29fdd2f --- /dev/null +++ b/actioncable/lib/action_cable/connection/web_socket.rb @@ -0,0 +1,41 @@ +# frozen_string_literal: true + +require "websocket/driver" + +module ActionCable + module Connection + # Wrap the real socket to minimize the externally-presented API + class WebSocket # :nodoc: + def initialize(env, event_target, event_loop, protocols: ActionCable::INTERNAL[:protocols]) + @websocket = ::WebSocket::Driver.websocket?(env) ? ClientSocket.new(env, event_target, event_loop, protocols) : nil + end + + def possible? + websocket + end + + def alive? + websocket && websocket.alive? + end + + def transmit(data) + websocket.transmit data + end + + def close + websocket.close + end + + def protocol + websocket.protocol + end + + def rack_response + websocket.rack_response + end + + private + 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..53cbb597cd --- /dev/null +++ b/actioncable/lib/action_cable/engine.rb @@ -0,0 +1,79 @@ +# frozen_string_literal: true + +require "rails" +require "action_cable" +require "action_cable/helpers/action_cable_helper" +require "active_support/core_ext/hash/indifferent_access" + +module ActionCable + class Engine < Rails::Engine # :nodoc: + config.action_cable = ActiveSupport::OrderedOptions.new + config.action_cable.mount_path = ActionCable::INTERNAL[:default_mount_path] + + 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 ||= /https?:\/\/localhost:\d+/ 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 + + previous_connection_class = connection_class + self.connection_class = -> { "ApplicationCable::Connection".safe_constantize || previous_connection_class.call } + + 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 new file mode 100644 index 0000000000..cd1d9bccef --- /dev/null +++ b/actioncable/lib/action_cable/gem_version.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +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 = 6 + MINOR = 0 + TINY = 0 + PRE = "alpha" + + 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..df16c02e83 --- /dev/null +++ b/actioncable/lib/action_cable/helpers/action_cable_helper.rb @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +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' => 'reload' %> + # </head> + # + # 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: + # + # #= require cable + # @App = {} + # App.cable = Cable.createConsumer() + # + # 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" /> + # + def action_cable_meta_tag + 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 +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..283400d9e7 --- /dev/null +++ b/actioncable/lib/action_cable/remote_connections.rb @@ -0,0 +1,71 @@ +# frozen_string_literal: true + +require "active_support/core_ext/module/redefine_method" + +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 of 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 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. + redefine_method :identifiers do + server.connection_identifiers + end + + protected + attr_reader :server + + private + 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..8d485a44f6 --- /dev/null +++ b/actioncable/lib/action_cable/server.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +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..2b9e1cba3b --- /dev/null +++ b/actioncable/lib/action_cable/server/base.rb @@ -0,0 +1,91 @@ +# frozen_string_literal: true + +require "monitor" + +module ActionCable + module Server + # 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 more information. + class Base + include ActionCable::Server::Broadcasting + include ActionCable::Server::Connections + + cattr_accessor :config, instance_accessor: true, default: ActionCable::Server::Configuration.new + + def self.logger; config.logger; end + delegate :logger, to: :config + + attr_reader :mutex + + def initialize + @mutex = Monitor.new + @remote_connections = @event_loop = @worker_pool = @pubsub = nil + end + + # Called by Rack to setup the server. + def call(env) + setup_heartbeat_timer + config.connection_class.call.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 + + def restart + connections.each do |connection| + connection.close(reason: ActionCable::INTERNAL[:disconnect_reasons][:server_restart]) + end + + @mutex.synchronize do + # Shutdown the worker pool + @worker_pool.halt if @worker_pool + @worker_pool = nil + + # Shutdown the pub/sub adapter + @pubsub.shutdown if @pubsub + @pubsub = nil + end + end + + # Gateway to RemoteConnections. See that class for details. + def remote_connections + @remote_connections || @mutex.synchronize { @remote_connections ||= RemoteConnections.new(self) } + end + + def event_loop + @event_loop || @mutex.synchronize { @event_loop ||= ActionCable::Connection::StreamEventLoop.new } + end + + # The worker pool is where we run connection callbacks and channel actions. We do as little as possible on the server's main thread. + # The worker pool is an executor service that's backed by a pool of threads working from a task queue. The thread pool size maxes out + # at 4 worker threads by default. Tune the size yourself with <tt>config.action_cable.worker_pool_size</tt>. + # + # Using Active Record, Redis, etc within your channel actions means you'll get a separate connection from each thread in the worker pool. + # Plan your deployment accordingly: 5 servers each running 5 Puma workers each running an 8-thread worker pool means at least 200 database + # connections. + # + # Also, ensure that your database connection pool size is as least as large as your worker pool size. Otherwise, workers may oversubscribe + # the database connection pool and block while they wait for other workers to release their connections. Use a smaller worker pool or a larger + # database connection pool instead. + def worker_pool + @worker_pool || @mutex.synchronize { @worker_pool ||= ActionCable::Server::Worker.new(max_size: config.worker_pool_size) } + end + + # Adapter used for all streams/broadcasting. + def pubsub + @pubsub || @mutex.synchronize { @pubsub ||= config.pubsub_adapter.new(self) } + end + + # All of the identifiers applied to the connection class associated with this server. + def connection_identifiers + config.connection_class.call.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..bc54d784b3 --- /dev/null +++ b/actioncable/lib/action_cable/server/broadcasting.rb @@ -0,0 +1,54 @@ +# frozen_string_literal: true + +module ActionCable + module Server + # Broadcasting is how other parts of your application can send messages to a channel's subscribers. As explained in Channel, most of the time, these + # broadcastings are streamed directly to the clients subscribed to the named broadcasting. Let's explain with a full-stack example: + # + # class WebNotificationsChannel < ApplicationCable::Channel + # def subscribed + # stream_from "web_notifications_#{current_user.id}" + # end + # end + # + # # 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>. This will later be JSON encoded. + def broadcast(broadcasting, message, coder: ActiveSupport::JSON) + broadcaster_for(broadcasting, coder: coder).broadcast(message) + end + + # 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, coder: ActiveSupport::JSON) + Broadcaster.new(self, String(broadcasting), coder: coder) + end + + private + class Broadcaster + attr_reader :server, :broadcasting, :coder + + def initialize(server, broadcasting, coder:) + @server, @broadcasting, @coder = server, broadcasting, coder + end + + def broadcast(message) + server.logger.debug "[ActionCable] Broadcasting to #{broadcasting}: #{message.inspect}" + + payload = { broadcasting: broadcasting, message: message, coder: coder } + ActiveSupport::Notifications.instrument("broadcast.action_cable", payload) do + encoded = coder ? coder.encode(message) : message + server.pubsub.broadcast broadcasting, encoded + end + end + end + end + 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..26209537df --- /dev/null +++ b/actioncable/lib/action_cable/server/configuration.rb @@ -0,0 +1,56 @@ +# frozen_string_literal: true + +module ActionCable + module Server + # 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 :disable_request_forgery_protection, :allowed_request_origins, :allow_same_origin_as_host + attr_accessor :cable, :url, :mount_path + + def initialize + @log_tags = [] + + @connection_class = -> { ActionCable::Connection::Base } + @worker_pool_size = 4 + + @disable_request_forgery_protection = false + @allow_same_origin_as_host = true + 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" }) + + # Require the adapter itself and give useful feedback about + # 1. Missing adapter gems and + # 2. Adapter gems' missing dependencies. + path_to_adapter = "action_cable/subscription_adapter/#{adapter}" + begin + require path_to_adapter + rescue LoadError => e + # We couldn't require the adapter itself. Raise an exception that + # points out config typos and missing gems. + if e.path == path_to_adapter + # We can assume that a non-builtin adapter was specified, so it's + # either misspelled or missing from Gemfile. + raise e.class, "Could not load the '#{adapter}' Action Cable pubsub adapter. Ensure that the adapter is spelled correctly in config/cable.yml and that you've added the necessary adapter gem to your Gemfile.", e.backtrace + + # Bubbled up from the adapter require. Prefix the exception message + # with some guidance about how to address it and reraise. + else + raise e.class, "Error loading the '#{adapter}' Action Cable pubsub adapter. Missing a gem it depends on? #{e.message}", e.backtrace + end + 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..39557d63a7 --- /dev/null +++ b/actioncable/lib/action_cable/server/connections.rb @@ -0,0 +1,36 @@ +# frozen_string_literal: true + +module ActionCable + module Server + # 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 + @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 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 ||= event_loop.timer(BEAT_INTERVAL) do + event_loop.post { connections.map(&:beat) } + end + 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..187c8f7939 --- /dev/null +++ b/actioncable/lib/action_cable/server/worker.rb @@ -0,0 +1,75 @@ +# frozen_string_literal: true + +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. + class Worker # :nodoc: + include ActiveSupport::Callbacks + + thread_mattr_accessor :connection + define_callbacks :work + include ActiveRecordConnectionManagement + + attr_reader :executor + + def initialize(max_size: 5) + @executor = Concurrent::ThreadPoolExecutor.new( + min_threads: 1, + max_threads: max_size, + max_queue: 0, + ) + end + + # Stop processing work: any work that has not already started + # running will be discarded from the queue + def halt + @executor.shutdown + end + + def stopping? + @executor.shuttingdown? + end + + def work(connection) + self.connection = connection + + run_callbacks :work do + yield + end + ensure + self.connection = nil + end + + def async_exec(receiver, *args, connection:, &block) + async_invoke receiver, :instance_exec, *args, connection: connection, &block + end + + def async_invoke(receiver, method, *args, connection: receiver, &block) + @executor.post do + invoke(receiver, method, *args, connection: connection, &block) + end + end + + def invoke(receiver, method, *args, connection:, &block) + work(connection) do + receiver.send method, *args, &block + 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) + 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..2e378d4bf3 --- /dev/null +++ b/actioncable/lib/action_cable/server/worker/active_record_connection_management.rb @@ -0,0 +1,21 @@ +# frozen_string_literal: true + +module ActionCable + module Server + class Worker + 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 } + end + end + end + end +end diff --git a/actioncable/lib/action_cable/subscription_adapter.rb b/actioncable/lib/action_cable/subscription_adapter.rb new file mode 100644 index 0000000000..6a9d5c2080 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter.rb @@ -0,0 +1,12 @@ +# frozen_string_literal: true + +module ActionCable + module SubscriptionAdapter + extend ActiveSupport::Autoload + + autoload :Base + autoload :Test + autoload :SubscriberMap + autoload :ChannelPrefix + 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..c9930299c7 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/async.rb @@ -0,0 +1,29 @@ +# frozen_string_literal: true + +require "action_cable/subscription_adapter/inline" + +module ActionCable + module SubscriptionAdapter + class Async < Inline # :nodoc: + private + def new_subscriber_map + AsyncSubscriberMap.new(server.event_loop) + end + + class AsyncSubscriberMap < SubscriberMap + def initialize(event_loop) + @event_loop = event_loop + super() + end + + def add_subscriber(*) + @event_loop.post { super } + end + + def invoke_callback(*) + @event_loop.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..34077707fd --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/base.rb @@ -0,0 +1,30 @@ +# frozen_string_literal: true + +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/channel_prefix.rb b/actioncable/lib/action_cable/subscription_adapter/channel_prefix.rb new file mode 100644 index 0000000000..df0aa040f5 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/channel_prefix.rb @@ -0,0 +1,28 @@ +# frozen_string_literal: true + +module ActionCable + module SubscriptionAdapter + module ChannelPrefix # :nodoc: + def broadcast(channel, payload) + channel = channel_with_prefix(channel) + super + end + + def subscribe(channel, callback, success_callback = nil) + channel = channel_with_prefix(channel) + super + end + + def unsubscribe(channel, callback) + channel = channel_with_prefix(channel) + super + end + + private + # Returns the channel name, including channel_prefix specified in cable.yml + def channel_with_prefix(channel) + [@server.config.cable[:channel_prefix], channel].compact.join(":") + 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..d2c85c1c8d --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/inline.rb @@ -0,0 +1,37 @@ +# frozen_string_literal: true + +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..50ec438c3a --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb @@ -0,0 +1,130 @@ +# frozen_string_literal: true + +gem "pg", ">= 0.18", "< 2.0" +require "pg" +require "thread" +require "digest/sha1" + +module ActionCable + module SubscriptionAdapter + class PostgreSQL < Base # :nodoc: + def initialize(*) + super + @listener = nil + end + + def broadcast(channel, payload) + with_broadcast_connection do |pg_conn| + pg_conn.exec("NOTIFY #{pg_conn.escape_identifier(channel_identifier(channel))}, '#{pg_conn.escape_string(payload)}'") + end + end + + def subscribe(channel, callback, success_callback = nil) + listener.add_subscriber(channel_identifier(channel), callback, success_callback) + end + + def unsubscribe(channel, callback) + listener.remove_subscriber(channel_identifier(channel), callback) + end + + def shutdown + listener.shutdown + end + + def with_subscriptions_connection(&block) # :nodoc: + ar_conn = ActiveRecord::Base.connection_pool.checkout.tap do |conn| + # Action Cable is taking ownership over this database connection, and + # will perform the necessary cleanup tasks + ActiveRecord::Base.connection_pool.remove(conn) + end + pg_conn = ar_conn.raw_connection + + verify!(pg_conn) + yield pg_conn + ensure + ar_conn.disconnect! + end + + def with_broadcast_connection(&block) # :nodoc: + ActiveRecord::Base.connection_pool.with_connection do |ar_conn| + pg_conn = ar_conn.raw_connection + verify!(pg_conn) + yield pg_conn + end + end + + private + def channel_identifier(channel) + channel.size > 63 ? Digest::SHA1.hexdigest(channel) : channel + end + + def listener + @listener || @server.mutex.synchronize { @listener ||= Listener.new(self, @server.event_loop) } + end + + def verify!(pg_conn) + unless pg_conn.is_a?(PG::Connection) + raise "The Active Record database must be PostgreSQL in order to use the PostgreSQL Action Cable storage adapter" + end + end + + class Listener < SubscriberMap + def initialize(adapter, event_loop) + super() + + @adapter = adapter + @event_loop = event_loop + @queue = Queue.new + + @thread = Thread.new do + Thread.current.abort_on_exception = true + listen + end + end + + def listen + @adapter.with_subscriptions_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}") + @event_loop.post(&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(*) + @event_loop.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..ad8fa52760 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb @@ -0,0 +1,179 @@ +# frozen_string_literal: true + +require "thread" + +gem "redis", ">= 3", "< 5" +require "redis" + +module ActionCable + module SubscriptionAdapter + class Redis < Base # :nodoc: + prepend ChannelPrefix + + # Overwrite this factory method for Redis connections if you want to use a different Redis library than the redis gem. + # This is needed, for example, when using Makara proxies for distributed Redis. + cattr_accessor :redis_connector, default: ->(config) do + config[:id] ||= "ActionCable-PID-#{$$}" + ::Redis.new(config.slice(:url, :host, :port, :db, :password, :id)) + end + + def initialize(*) + super + @listener = nil + @redis_connection_for_broadcasts = nil + end + + def broadcast(channel, payload) + redis_connection_for_broadcasts.publish(channel, payload) + end + + def subscribe(channel, callback, success_callback = nil) + listener.add_subscriber(channel, callback, success_callback) + end + + def unsubscribe(channel, callback) + listener.remove_subscriber(channel, callback) + end + + def shutdown + @listener.shutdown if @listener + end + + def redis_connection_for_subscriptions + redis_connection + end + + private + def listener + @listener || @server.mutex.synchronize { @listener ||= Listener.new(self, @server.event_loop) } + end + + def redis_connection_for_broadcasts + @redis_connection_for_broadcasts || @server.mutex.synchronize do + @redis_connection_for_broadcasts ||= redis_connection + end + end + + def redis_connection + self.class.redis_connector.call(@server.config.cable) + end + + class Listener < SubscriberMap + def initialize(adapter, event_loop) + super() + + @adapter = adapter + @event_loop = event_loop + + @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.respond_to?(:_client) ? conn._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 + @event_loop.post(&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(*) + @event_loop.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..01cdc2dfa1 --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/subscriber_map.rb @@ -0,0 +1,59 @@ +# frozen_string_literal: true + +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 do + return if !@subscribers.key?(channel) + @subscribers[channel].dup + end + + 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/subscription_adapter/test.rb b/actioncable/lib/action_cable/subscription_adapter/test.rb new file mode 100644 index 0000000000..ce604cc88e --- /dev/null +++ b/actioncable/lib/action_cable/subscription_adapter/test.rb @@ -0,0 +1,40 @@ +# frozen_string_literal: true + +require_relative "async" + +module ActionCable + module SubscriptionAdapter + # == Test adapter for Action Cable + # + # The test adapter should be used only in testing. Along with + # <tt>ActionCable::TestHelper</tt> it makes a great tool to test your Rails application. + # + # To use the test adapter set +adapter+ value to +test+ in your +config/cable.yml+ file. + # + # NOTE: Test adapter extends the <tt>ActionCable::SubscriptionsAdapter::Async</tt> adapter, + # so it could be used in system tests too. + class Test < Async + def broadcast(channel, payload) + broadcasts(channel) << payload + super + end + + def broadcasts(channel) + channels_data[channel] ||= [] + end + + def clear_messages(channel) + channels_data[channel] = [] + end + + def clear + @channels_data = nil + end + + private + def channels_data + @channels_data ||= {} + end + end + end +end diff --git a/actioncable/lib/action_cable/test_case.rb b/actioncable/lib/action_cable/test_case.rb new file mode 100644 index 0000000000..d153259bf6 --- /dev/null +++ b/actioncable/lib/action_cable/test_case.rb @@ -0,0 +1,11 @@ +# frozen_string_literal: true + +require "active_support/test_case" + +module ActionCable + class TestCase < ActiveSupport::TestCase + include ActionCable::TestHelper + + ActiveSupport.run_load_hooks(:action_cable_test_case, self) + end +end diff --git a/actioncable/lib/action_cable/test_helper.rb b/actioncable/lib/action_cable/test_helper.rb new file mode 100644 index 0000000000..7bc877663c --- /dev/null +++ b/actioncable/lib/action_cable/test_helper.rb @@ -0,0 +1,133 @@ +# frozen_string_literal: true + +module ActionCable + # Provides helper methods for testing Action Cable broadcasting + module TestHelper + def before_setup # :nodoc: + server = ActionCable.server + test_adapter = ActionCable::SubscriptionAdapter::Test.new(server) + + @old_pubsub_adapter = server.pubsub + + server.instance_variable_set(:@pubsub, test_adapter) + super + end + + def after_teardown # :nodoc: + super + ActionCable.server.instance_variable_set(:@pubsub, @old_pubsub_adapter) + end + + # Asserts that the number of broadcasted messages to the stream matches the given number. + # + # def test_broadcasts + # assert_broadcasts 'messages', 0 + # ActionCable.server.broadcast 'messages', { text: 'hello' } + # assert_broadcasts 'messages', 1 + # ActionCable.server.broadcast 'messages', { text: 'world' } + # assert_broadcasts 'messages', 2 + # end + # + # If a block is passed, that block should cause the specified number of + # messages to be broadcasted. + # + # def test_broadcasts_again + # assert_broadcasts('messages', 1) do + # ActionCable.server.broadcast 'messages', { text: 'hello' } + # end + # + # assert_broadcasts('messages', 2) do + # ActionCable.server.broadcast 'messages', { text: 'hi' } + # ActionCable.server.broadcast 'messages', { text: 'how are you?' } + # end + # end + # + def assert_broadcasts(stream, number) + if block_given? + original_count = broadcasts_size(stream) + yield + new_count = broadcasts_size(stream) + actual_count = new_count - original_count + else + actual_count = broadcasts_size(stream) + end + + assert_equal number, actual_count, "#{number} broadcasts to #{stream} expected, but #{actual_count} were sent" + end + + # Asserts that no messages have been sent to the stream. + # + # def test_no_broadcasts + # assert_no_broadcasts 'messages' + # ActionCable.server.broadcast 'messages', { text: 'hi' } + # assert_broadcasts 'messages', 1 + # end + # + # If a block is passed, that block should not cause any message to be sent. + # + # def test_broadcasts_again + # assert_no_broadcasts 'messages' do + # # No job messages should be sent from this block + # end + # end + # + # Note: This assertion is simply a shortcut for: + # + # assert_broadcasts 'messages', 0, &block + # + def assert_no_broadcasts(stream, &block) + assert_broadcasts stream, 0, &block + end + + # Asserts that the specified message has been sent to the stream. + # + # def test_assert_transmited_message + # ActionCable.server.broadcast 'messages', text: 'hello' + # assert_broadcast_on('messages', text: 'hello') + # end + # + # If a block is passed, that block should cause a message with the specified data to be sent. + # + # def test_assert_broadcast_on_again + # assert_broadcast_on('messages', text: 'hello') do + # ActionCable.server.broadcast 'messages', text: 'hello' + # end + # end + # + def assert_broadcast_on(stream, data) + # Encode to JSON and back–we want to use this value to compare + # with decoded JSON. + # Comparing JSON strings doesn't work due to the order if the keys. + serialized_msg = + ActiveSupport::JSON.decode(ActiveSupport::JSON.encode(data)) + + new_messages = broadcasts(stream) + if block_given? + old_messages = new_messages + clear_messages(stream) + + yield + new_messages = broadcasts(stream) + clear_messages(stream) + + # Restore all sent messages + (old_messages + new_messages).each { |m| pubsub_adapter.broadcast(stream, m) } + end + + message = new_messages.find { |msg| ActiveSupport::JSON.decode(msg) == serialized_msg } + + assert message, "No messages sent with #{data} to #{stream}" + end + + def pubsub_adapter # :nodoc: + ActionCable.server.pubsub + end + + delegate :broadcasts, :clear_messages, to: :pubsub_adapter + + private + def broadcasts_size(channel) + broadcasts(channel).size + 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..86115c6065 --- /dev/null +++ b/actioncable/lib/action_cable/version.rb @@ -0,0 +1,10 @@ +# frozen_string_literal: true + +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..ea9662436c --- /dev/null +++ b/actioncable/lib/rails/generators/channel/USAGE @@ -0,0 +1,12 @@ +Description: +============ + Stubs out a new cable channel for the server (in Ruby) and client (in JavaScript). + Pass the channel name, either CamelCased or under_scored, and an optional list of channel actions as arguments. + +Example: +======== + rails generate channel Chat speak + + creates a Chat channel class and JavaScript asset: + Channel: app/channels/chat_channel.rb + Assets: app/javascript/channels/chat_channel.js 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..ef51981e89 --- /dev/null +++ b/actioncable/lib/rails/generators/channel/channel_generator.rb @@ -0,0 +1,50 @@ +# frozen_string_literal: true + +module Rails + module Generators + class ChannelGenerator < NamedBase + source_root File.expand_path("templates", __dir__) + + 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] + if behavior == :invoke + template "javascript/index.js", "app/javascript/channels/index.js" + template "javascript/consumer.js", "app/javascript/channels/consumer.js" + end + + js_template "javascript/channel", File.join("app/javascript/channels", class_path, "#{file_name}_channel") + end + + generate_application_cable_files + end + + private + def file_name + @_file_name ||= super.sub(/_channel\z/i, "") + end + + # FIXME: Change these files to symlinks once RubyGems 2.5.0 is required. + def generate_application_cable_files + return if 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.tt b/actioncable/lib/rails/generators/channel/templates/application_cable/channel.rb.tt new file mode 100644 index 0000000000..d672697283 --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/application_cable/channel.rb.tt @@ -0,0 +1,4 @@ +module ApplicationCable + class Channel < ActionCable::Channel::Base + end +end diff --git a/actioncable/lib/rails/generators/channel/templates/application_cable/connection.rb.tt b/actioncable/lib/rails/generators/channel/templates/application_cable/connection.rb.tt new file mode 100644 index 0000000000..0ff5442f47 --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/application_cable/connection.rb.tt @@ -0,0 +1,4 @@ +module ApplicationCable + class Connection < ActionCable::Connection::Base + end +end diff --git a/actioncable/lib/rails/generators/channel/templates/channel.rb.tt b/actioncable/lib/rails/generators/channel/templates/channel.rb.tt new file mode 100644 index 0000000000..4bcfb2be4d --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/channel.rb.tt @@ -0,0 +1,16 @@ +<% 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 -%> diff --git a/actioncable/lib/rails/generators/channel/templates/javascript/channel.js.tt b/actioncable/lib/rails/generators/channel/templates/javascript/channel.js.tt new file mode 100644 index 0000000000..33baaa5a22 --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/javascript/channel.js.tt @@ -0,0 +1,20 @@ +import consumer from "./consumer" + +consumer.subscriptions.create("<%= class_name %>Channel", { + connected: function() { + // Called when the subscription is ready for use on the server + }, + + disconnected: function() { + // Called when the subscription has been terminated by the server + }, + + received: function(data) { + // Called when there's incoming data on the websocket for this channel + }<%= actions.any? ? ",\n" : '' %> +<% actions.each do |action| -%> + <%=action %>: function() { + return this.perform('<%= action %>'); + }<%= action == actions[-1] ? '' : ",\n" %> +<% end -%> +}); diff --git a/actioncable/lib/rails/generators/channel/templates/javascript/consumer.js.tt b/actioncable/lib/rails/generators/channel/templates/javascript/consumer.js.tt new file mode 100644 index 0000000000..76ca3d0f2f --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/javascript/consumer.js.tt @@ -0,0 +1,6 @@ +// Action Cable provides the framework to deal with WebSockets in Rails. +// You can generate new channels where WebSocket features live using the `rails generate channel` command. + +import ActionCable from "actioncable" + +export default ActionCable.createConsumer() diff --git a/actioncable/lib/rails/generators/channel/templates/javascript/index.js.tt b/actioncable/lib/rails/generators/channel/templates/javascript/index.js.tt new file mode 100644 index 0000000000..0cfcf74919 --- /dev/null +++ b/actioncable/lib/rails/generators/channel/templates/javascript/index.js.tt @@ -0,0 +1,5 @@ +// Load all the channels within this directory and all subdirectories. +// Channel files must be named *_channel.js. + +const channels = require.context('.', true, /_channel\.js$/) +channels.keys().forEach(channels) diff --git a/actioncable/package.json b/actioncable/package.json new file mode 100644 index 0000000000..db78c1a09a --- /dev/null +++ b/actioncable/package.json @@ -0,0 +1,51 @@ +{ + "name": "actioncable", + "version": "6.0.0-alpha", + "description": "WebSocket framework for Ruby on Rails.", + "main": "app/assets/javascripts/action_cable.js", + "files": [ + "app/assets/javascripts/*.js", + "src/*.js" + ], + "repository": { + "type": "git", + "url": "rails/rails" + }, + "keywords": [ + "websockets", + "actioncable", + "rails" + ], + "author": "David Heinemeier Hansson <david@loudthinking.com>", + "license": "MIT", + "bugs": { + "url": "https://github.com/rails/rails/issues" + }, + "homepage": "http://rubyonrails.org/", + "devDependencies": { + "babel-core": "^6.25.0", + "babel-plugin-external-helpers": "^6.22.0", + "babel-preset-env": "^1.6.0", + "eslint": "^4.3.0", + "eslint-plugin-import": "^2.7.0", + "karma": "^3.1.1", + "karma-chrome-launcher": "^2.2.0", + "karma-qunit": "^2.1.0", + "karma-sauce-launcher": "^1.2.0", + "mock-socket": "^2.0.0", + "qunit": "^2.8.0", + "rollup": "^0.58.2", + "rollup-plugin-babel": "^3.0.4", + "rollup-plugin-commonjs": "^9.1.0", + "rollup-plugin-node-resolve": "^3.3.0", + "rollup-plugin-uglify": "^3.0.0" + }, + "scripts": { + "prebuild": "yarn lint && bundle exec rake assets:codegen", + "build": "rollup --config rollup.config.js", + "lint": "eslint app/javascript", + "prepublishOnly": "rm -rf src && cp -R app/javascript/action_cable src", + "pretest": "bundle exec rake assets:codegen && rollup --config rollup.config.test.js", + "test": "karma start" + } +} diff --git a/actioncable/rollup.config.js b/actioncable/rollup.config.js new file mode 100644 index 0000000000..64727e0887 --- /dev/null +++ b/actioncable/rollup.config.js @@ -0,0 +1,24 @@ +import babel from "rollup-plugin-babel" +import uglify from "rollup-plugin-uglify" + +const uglifyOptions = { + mangle: false, + compress: false, + output: { + beautify: true, + indent_level: 2 + } +} + +export default { + input: "app/javascript/action_cable/index.js", + output: { + file: "app/assets/javascripts/action_cable.js", + format: "umd", + name: "ActionCable" + }, + plugins: [ + babel(), + uglify(uglifyOptions) + ] +} diff --git a/actioncable/rollup.config.test.js b/actioncable/rollup.config.test.js new file mode 100644 index 0000000000..f92ff36240 --- /dev/null +++ b/actioncable/rollup.config.test.js @@ -0,0 +1,18 @@ +import babel from "rollup-plugin-babel" +import commonjs from "rollup-plugin-commonjs" +import resolve from "rollup-plugin-node-resolve" + +export default { + input: "test/javascript/src/test.js", + + output: { + file: "test/javascript/compiled/test.js", + format: "iife" + }, + + plugins: [ + resolve(), + commonjs(), + babel() + ] +} diff --git a/actioncable/test/channel/base_test.rb b/actioncable/test/channel/base_test.rb new file mode 100644 index 0000000000..39b5879607 --- /dev/null +++ b/actioncable/test/channel/base_test.rb @@ -0,0 +1,282 @@ +# frozen_string_literal: true + +require "test_helper" +require "minitest/mock" +require "stubs/test_connection" +require "stubs/room" + +class ActionCable::Channel::BaseTest < ActionCable::TestCase + class ActionCable::Channel::Base + def kick + @last_action = [ :kick ] + end + + def topic + end + end + + class BasicChannel < ActionCable::Channel::Base + def chatters + @last_action = [ :chatters ] + end + end + + class ChatChannel < BasicChannel + attr_reader :room, :last_action + after_subscribe :toggle_subscribed + after_unsubscribe :toggle_subscribed + + class SomeCustomError < StandardError; end + rescue_from SomeCustomError, with: :error_handler + + def initialize(*) + @subscribed = false + super + end + + def subscribed + @room = Room.new params[:id] + @actions = [] + end + + def unsubscribed + @room = nil + end + + def toggle_subscribed + @subscribed = !@subscribed + end + + def leave + @last_action = [ :leave ] + end + + def speak(data) + @last_action = [ :speak, data ] + end + + def topic(data) + @last_action = [ :topic, data ] + end + + def subscribed? + @subscribed + end + + def get_latest + transmit data: "latest" + end + + def receive + @last_action = [ :receive ] + end + + def error_action + raise SomeCustomError + end + + private + def rm_rf + @last_action = [ :rm_rf ] + end + + def error_handler + @last_action = [ :error_action ] + end + end + + setup do + @user = User.new "lifo" + @connection = TestConnection.new(@user) + @channel = ChatChannel.new @connection, "{id: 1}", id: 1 + end + + test "should subscribe to a channel" do + @channel.subscribe_to_channel + assert_equal 1, @channel.room.id + end + + test "on subscribe callbacks" do + @channel.subscribe_to_channel + assert @channel.subscribed + end + + test "channel params" do + assert_equal({ id: 1 }, @channel.params) + end + + test "unsubscribing from a channel" do + @channel.subscribe_to_channel + + assert @channel.room + assert_predicate @channel, :subscribed? + + @channel.unsubscribe_from_channel + + assert_not @channel.room + assert_not_predicate @channel, :subscribed? + end + + test "connection identifiers" do + assert_equal @user.name, @channel.current_user.name + end + + test "callable action without any argument" do + @channel.perform_action "action" => :leave + assert_equal [ :leave ], @channel.last_action + end + + test "callable action with arguments" do + data = { "action" => :speak, "content" => "Hello World" } + + @channel.perform_action data + assert_equal [ :speak, data ], @channel.last_action + end + + test "should not dispatch a private method" do + @channel.perform_action "action" => :rm_rf + assert_nil @channel.last_action + end + + test "should not dispatch a public method defined on Base" do + @channel.perform_action "action" => :kick + assert_nil @channel.last_action + end + + test "should dispatch a public method defined on Base and redefined on channel" do + data = { "action" => :topic, "content" => "This is Sparta!" } + + @channel.perform_action data + assert_equal [ :topic, data ], @channel.last_action + end + + test "should dispatch calling a public method defined in an ancestor" do + @channel.perform_action "action" => :chatters + assert_equal [ :chatters ], @channel.last_action + end + + test "should dispatch receive action when perform_action is called with empty action" do + data = { "content" => "hello" } + @channel.perform_action data + assert_equal [ :receive ], @channel.last_action + end + + test "transmitting data" do + @channel.perform_action "action" => :get_latest + + expected = { "identifier" => "{id: 1}", "message" => { "data" => "latest" } } + assert_equal expected, @connection.last_transmission + end + + test "do not send subscription confirmation on initialize" do + assert_nil @connection.last_transmission + end + + test "subscription confirmation on subscribe_to_channel" do + expected = { "identifier" => "{id: 1}", "type" => "confirm_subscription" } + @channel.subscribe_to_channel + assert_equal expected, @connection.last_transmission + end + + test "actions available on Channel" do + available_actions = %w(room last_action subscribed unsubscribed toggle_subscribed leave speak subscribed? get_latest receive chatters topic error_action).to_set + assert_equal available_actions, ChatChannel.action_methods + end + + test "invalid action on Channel" do + assert_logged("Unable to process ActionCable::Channel::BaseTest::ChatChannel#invalid_action") do + @channel.perform_action "action" => :invalid_action + end + end + + test "notification for perform_action" do + events = [] + ActiveSupport::Notifications.subscribe "perform_action.action_cable" do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + data = { "action" => :speak, "content" => "hello" } + @channel.perform_action data + + assert_equal 1, events.length + assert_equal "perform_action.action_cable", events[0].name + assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class] + assert_equal :speak, events[0].payload[:action] + assert_equal data, events[0].payload[:data] + ensure + ActiveSupport::Notifications.unsubscribe "perform_action.action_cable" + end + + test "notification for transmit" do + events = [] + ActiveSupport::Notifications.subscribe "transmit.action_cable" do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + @channel.perform_action "action" => :get_latest + expected_data = { data: "latest" } + + assert_equal 1, events.length + assert_equal "transmit.action_cable", events[0].name + assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class] + assert_equal expected_data, events[0].payload[:data] + assert_nil events[0].payload[:via] + ensure + ActiveSupport::Notifications.unsubscribe "transmit.action_cable" + end + + test "notification for transmit_subscription_confirmation" do + @channel.subscribe_to_channel + + events = [] + ActiveSupport::Notifications.subscribe "transmit_subscription_confirmation.action_cable" do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + @channel.stub(:subscription_confirmation_sent?, false) do + @channel.send(:transmit_subscription_confirmation) + + assert_equal 1, events.length + assert_equal "transmit_subscription_confirmation.action_cable", events[0].name + assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class] + end + ensure + ActiveSupport::Notifications.unsubscribe "transmit_subscription_confirmation.action_cable" + end + + test "notification for transmit_subscription_rejection" do + events = [] + ActiveSupport::Notifications.subscribe "transmit_subscription_rejection.action_cable" do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + @channel.send(:transmit_subscription_rejection) + + assert_equal 1, events.length + assert_equal "transmit_subscription_rejection.action_cable", events[0].name + assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class] + ensure + ActiveSupport::Notifications.unsubscribe "transmit_subscription_rejection.action_cable" + end + + test "behaves like rescuable" do + @channel.perform_action "action" => :error_action + assert_equal [ :error_action ], @channel.last_action + end + + private + def assert_logged(message) + old_logger = @connection.logger + log = StringIO.new + @connection.instance_variable_set(:@logger, Logger.new(log)) + + begin + yield + + log.rewind + assert_match message, log.read + ensure + @connection.instance_variable_set(:@logger, old_logger) + end + end +end diff --git a/actioncable/test/channel/broadcasting_test.rb b/actioncable/test/channel/broadcasting_test.rb new file mode 100644 index 0000000000..2cbfabc1d0 --- /dev/null +++ b/actioncable/test/channel/broadcasting_test.rb @@ -0,0 +1,39 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_connection" +require "stubs/room" + +class ActionCable::Channel::BroadcastingTest < ActionCable::TestCase + class ChatChannel < ActionCable::Channel::Base + end + + setup do + @connection = TestConnection.new + end + + test "broadcasts_to" do + assert_called_with( + ActionCable.server, + :broadcast, + [ + "action_cable:channel:broadcasting_test:chat:Room#1-Campfire", + "Hello World" + ] + ) do + ChatChannel.broadcast_to(Room.new(1), "Hello World") + end + end + + test "broadcasting_for with an object" do + assert_equal "Room#1-Campfire", ChatChannel.broadcasting_for(Room.new(1)) + end + + test "broadcasting_for with an array" do + assert_equal "Room#1-Campfire:Room#2-Campfire", ChatChannel.broadcasting_for([ Room.new(1), Room.new(2) ]) + end + + test "broadcasting_for with a string" do + assert_equal "hello", ChatChannel.broadcasting_for("hello") + end +end diff --git a/actioncable/test/channel/naming_test.rb b/actioncable/test/channel/naming_test.rb new file mode 100644 index 0000000000..45652d9cc9 --- /dev/null +++ b/actioncable/test/channel/naming_test.rb @@ -0,0 +1,12 @@ +# frozen_string_literal: true + +require "test_helper" + +class ActionCable::Channel::NamingTest < ActionCable::TestCase + class ChatChannel < ActionCable::Channel::Base + end + + test "channel_name" do + assert_equal "action_cable:channel:naming_test:chat", ChatChannel.channel_name + end +end diff --git a/actioncable/test/channel/periodic_timers_test.rb b/actioncable/test/channel/periodic_timers_test.rb new file mode 100644 index 0000000000..0c979f4c7c --- /dev/null +++ b/actioncable/test/channel/periodic_timers_test.rb @@ -0,0 +1,85 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_connection" +require "stubs/room" +require "active_support/time" + +class ActionCable::Channel::PeriodicTimersTest < ActionCable::TestCase + class ChatChannel < ActionCable::Channel::Base + # Method name arg + periodically :send_updates, every: 1 + + # Proc arg + periodically -> { ping }, every: 2 + + # Block arg + periodically every: 3 do + ping + end + + private + def ping + end + end + + setup do + @connection = TestConnection.new + end + + test "periodic timers definition" do + timers = ChatChannel.periodic_timers + + assert_equal 3, timers.size + + timers.each_with_index do |timer, i| + assert_kind_of Proc, timer[0] + assert_equal i + 1, timer[1][:every] + end + end + + test "disallow negative and zero periods" do + [ 0, 0.0, 0.seconds, -1, -1.seconds, "foo", :foo, Object.new ].each do |invalid| + e = assert_raise ArgumentError do + ChatChannel.periodically :send_updates, every: invalid + end + assert_match(/Expected every:/, e.message) + end + end + + test "disallow block and arg together" do + e = assert_raise ArgumentError do + ChatChannel.periodically(:send_updates, every: 1) { ping } + end + assert_match(/not both/, e.message) + end + + test "disallow unknown args" do + [ "send_updates", Object.new, nil ].each do |invalid| + e = assert_raise ArgumentError do + ChatChannel.periodically invalid, every: 1 + end + assert_match(/Expected a Symbol/, e.message) + end + end + + test "timer start and stop" do + mock = Minitest::Mock.new + 3.times { mock.expect(:shutdown, nil) } + + assert_called( + @connection.server.event_loop, + :timer, + times: 3, + returns: mock + ) do + channel = ChatChannel.new @connection, "{id: 1}", id: 1 + + channel.subscribe_to_channel + channel.unsubscribe_from_channel + assert_equal [], channel.send(:active_periodic_timers) + end + + assert mock.verify + end +end diff --git a/actioncable/test/channel/rejection_test.rb b/actioncable/test/channel/rejection_test.rb new file mode 100644 index 0000000000..683eafcac0 --- /dev/null +++ b/actioncable/test/channel/rejection_test.rb @@ -0,0 +1,56 @@ +# frozen_string_literal: true + +require "test_helper" +require "minitest/mock" +require "stubs/test_connection" +require "stubs/room" + +class ActionCable::Channel::RejectionTest < ActionCable::TestCase + class SecretChannel < ActionCable::Channel::Base + def subscribed + reject if params[:id] > 0 + end + + def secret_action + end + end + + setup do + @user = User.new "lifo" + @connection = TestConnection.new(@user) + end + + test "subscription rejection" do + subscriptions = Minitest::Mock.new + subscriptions.expect(:remove_subscription, SecretChannel, [SecretChannel]) + + @connection.stub(:subscriptions, subscriptions) do + @channel = SecretChannel.new @connection, "{id: 1}", id: 1 + @channel.subscribe_to_channel + + expected = { "identifier" => "{id: 1}", "type" => "reject_subscription" } + assert_equal expected, @connection.last_transmission + end + + assert subscriptions.verify + end + + test "does not execute action if subscription is rejected" do + subscriptions = Minitest::Mock.new + subscriptions.expect(:remove_subscription, SecretChannel, [SecretChannel]) + + @connection.stub(:subscriptions, subscriptions) do + @channel = SecretChannel.new @connection, "{id: 1}", id: 1 + @channel.subscribe_to_channel + + expected = { "identifier" => "{id: 1}", "type" => "reject_subscription" } + assert_equal expected, @connection.last_transmission + assert_equal 1, @connection.transmissions.size + + @channel.perform_action("action" => :secret_action) + assert_equal 1, @connection.transmissions.size + end + + assert subscriptions.verify + end +end diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb new file mode 100644 index 0000000000..9ad2213d47 --- /dev/null +++ b/actioncable/test/channel/stream_test.rb @@ -0,0 +1,230 @@ +# frozen_string_literal: true + +require "test_helper" +require "minitest/mock" +require "stubs/test_connection" +require "stubs/room" + +module ActionCable::StreamTests + class Connection < ActionCable::Connection::Base + attr_reader :websocket + + def send_async(method, *args) + send method, *args + end + end + + class ChatChannel < ActionCable::Channel::Base + def subscribed + if params[:id] + @room = Room.new params[:id] + stream_from "test_room_#{@room.id}", coder: pick_coder(params[:coder]) + end + end + + def send_confirmation + transmit_subscription_confirmation + end + + private + def pick_coder(coder) + case coder + when nil, "json" + ActiveSupport::JSON + when "custom" + DummyEncoder + when "none" + nil + end + end + end + + module DummyEncoder + extend self + def encode(*) '{ "foo": "encoded" }' end + def decode(*) { foo: "decoded" } end + end + + class SymbolChannel < ActionCable::Channel::Base + def subscribed + stream_from :channel + end + end + + class StreamTest < ActionCable::TestCase + test "streaming start and stop" do + run_in_eventmachine do + connection = TestConnection.new + pubsub = Minitest::Mock.new connection.pubsub + + pubsub.expect(:subscribe, nil, ["test_room_1", Proc, Proc]) + pubsub.expect(:unsubscribe, nil, ["test_room_1", Proc]) + + connection.stub(:pubsub, pubsub) do + channel = ChatChannel.new connection, "{id: 1}", id: 1 + channel.subscribe_to_channel + + wait_for_async + channel.unsubscribe_from_channel + end + + assert pubsub.verify + end + end + + test "stream from non-string channel" do + run_in_eventmachine do + connection = TestConnection.new + pubsub = Minitest::Mock.new connection.pubsub + + pubsub.expect(:subscribe, nil, ["channel", Proc, Proc]) + pubsub.expect(:unsubscribe, nil, ["channel", Proc]) + + connection.stub(:pubsub, pubsub) do + channel = SymbolChannel.new connection, "" + channel.subscribe_to_channel + + wait_for_async + + channel.unsubscribe_from_channel + end + + assert pubsub.verify + end + end + + test "stream_for" do + run_in_eventmachine do + connection = TestConnection.new + + channel = ChatChannel.new connection, "" + channel.subscribe_to_channel + channel.stream_for Room.new(1) + wait_for_async + + pubsub_call = channel.pubsub.class.class_variable_get "@@subscribe_called" + + assert_equal "action_cable:stream_tests:chat:Room#1-Campfire", pubsub_call[:channel] + assert_instance_of Proc, pubsub_call[:callback] + assert_instance_of Proc, pubsub_call[:success_callback] + end + end + + test "stream_from subscription confirmation" do + run_in_eventmachine do + connection = TestConnection.new + + channel = ChatChannel.new connection, "{id: 1}", id: 1 + channel.subscribe_to_channel + + assert_nil connection.last_transmission + + wait_for_async + + confirmation = { "identifier" => "{id: 1}", "type" => "confirm_subscription" } + connection.transmit(confirmation) + + assert_equal confirmation, connection.last_transmission, "Did not receive subscription confirmation within 0.1s" + end + end + + test "subscription confirmation should only be sent out once" do + run_in_eventmachine do + connection = TestConnection.new + + channel = ChatChannel.new connection, "test_channel" + channel.send_confirmation + channel.send_confirmation + + wait_for_async + + expected = { "identifier" => "test_channel", "type" => "confirm_subscription" } + assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation" + + assert_equal 1, connection.transmissions.size + end + end + end + + require "action_cable/subscription_adapter/async" + + class UserCallbackChannel < ActionCable::Channel::Base + def subscribed + stream_from :channel do + Thread.current[:ran_callback] = true + end + end + end + + class MultiChatChannel < ActionCable::Channel::Base + def subscribed + stream_from "main_room" + stream_from "test_all_rooms" + end + end + + class StreamFromTest < ActionCable::TestCase + setup do + @server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Async) + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + end + + test "custom encoder" do + run_in_eventmachine do + connection = open_connection + subscribe_to connection, identifiers: { id: 1 } + + assert_called(connection.websocket, :transmit) do + @server.broadcast "test_room_1", { foo: "bar" }, { coder: DummyEncoder } + wait_for_async + wait_for_executor connection.server.worker_pool.executor + end + end + end + + test "user supplied callbacks are run through the worker pool" do + run_in_eventmachine do + connection = open_connection + receive(connection, command: "subscribe", channel: UserCallbackChannel.name, identifiers: { id: 1 }) + + @server.broadcast "channel", {} + wait_for_async + assert_not Thread.current[:ran_callback], "User callback was not run through the worker pool" + end + end + + test "subscription confirmation should only be sent out once with multiple stream_from" do + run_in_eventmachine do + connection = open_connection + expected = { "identifier" => { "channel" => MultiChatChannel.name }.to_json, "type" => "confirm_subscription" } + assert_called_with(connection.websocket, :transmit, [expected.to_json]) do + receive(connection, command: "subscribe", channel: MultiChatChannel.name, identifiers: {}) + wait_for_async + end + end + end + + private + def subscribe_to(connection, identifiers:) + receive connection, command: "subscribe", identifiers: identifiers + end + + def open_connection + env = Rack::MockRequest.env_for "/test", "HTTP_HOST" => "localhost", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket", "HTTP_ORIGIN" => "http://rubyonrails.com" + + Connection.new(@server, env).tap do |connection| + connection.process + assert_predicate connection.websocket, :possible? + + wait_for_async + assert_predicate connection.websocket, :alive? + end + end + + def receive(connection, command:, identifiers:, channel: "ActionCable::StreamTests::ChatChannel") + identifier = JSON.generate(identifiers.merge(channel: channel)) + connection.dispatch_websocket_message JSON.generate(command: command, identifier: identifier) + wait_for_async + end + end +end diff --git a/actioncable/test/channel/test_case_test.rb b/actioncable/test/channel/test_case_test.rb new file mode 100644 index 0000000000..9c360d5dc3 --- /dev/null +++ b/actioncable/test/channel/test_case_test.rb @@ -0,0 +1,214 @@ +# frozen_string_literal: true + +require "test_helper" + +class TestTestChannel < ActionCable::Channel::Base +end + +class NonInferrableExplicitClassChannelTest < ActionCable::Channel::TestCase + tests TestTestChannel + + def test_set_channel_class_manual + assert_equal TestTestChannel, self.class.channel_class + end +end + +class NonInferrableSymbolNameChannelTest < ActionCable::Channel::TestCase + tests :test_test_channel + + def test_set_channel_class_manual_using_symbol + assert_equal TestTestChannel, self.class.channel_class + end +end + +class NonInferrableStringNameChannelTest < ActionCable::Channel::TestCase + tests "test_test_channel" + + def test_set_channel_class_manual_using_string + assert_equal TestTestChannel, self.class.channel_class + end +end + +class SubscriptionsTestChannel < ActionCable::Channel::Base +end + +class SubscriptionsTestChannelTest < ActionCable::Channel::TestCase + def setup + stub_connection + end + + def test_no_subscribe + assert_nil subscription + end + + def test_subscribe + subscribe + + assert subscription.confirmed? + assert_not subscription.rejected? + assert_equal 1, connection.transmissions.size + assert_equal ActionCable::INTERNAL[:message_types][:confirmation], + connection.transmissions.last["type"] + end +end + +class StubConnectionTest < ActionCable::Channel::TestCase + tests SubscriptionsTestChannel + + def test_connection_identifiers + stub_connection username: "John", admin: true + + subscribe + + assert_equal "John", subscription.username + assert subscription.admin + end +end + +class RejectionTestChannel < ActionCable::Channel::Base + def subscribed + reject + end +end + +class RejectionTestChannelTest < ActionCable::Channel::TestCase + def test_rejection + subscribe + + assert_not subscription.confirmed? + assert subscription.rejected? + assert_equal 1, connection.transmissions.size + assert_equal ActionCable::INTERNAL[:message_types][:rejection], + connection.transmissions.last["type"] + end +end + +class StreamsTestChannel < ActionCable::Channel::Base + def subscribed + stream_from "test_#{params[:id] || 0}" + end +end + +class StreamsTestChannelTest < ActionCable::Channel::TestCase + def test_stream_without_params + subscribe + + assert_has_stream "test_0" + end + + def test_stream_with_params + subscribe id: 42 + + assert_has_stream "test_42" + end +end + +class StreamsForTestChannel < ActionCable::Channel::Base + def subscribed + stream_for User.new(params[:id]) + end +end + +class StreamsForTestChannelTest < ActionCable::Channel::TestCase + def test_stream_with_params + subscribe id: 42 + + assert_has_stream_for User.new(42) + end +end + +class NoStreamsTestChannel < ActionCable::Channel::Base + def subscribed; end # no-op +end + +class NoStreamsTestChannelTest < ActionCable::Channel::TestCase + def test_stream_with_params + subscribe + + assert_no_streams + end +end + +class PerformTestChannel < ActionCable::Channel::Base + def echo(data) + data.delete("action") + transmit data + end + + def ping + transmit type: "pong" + end +end + +class PerformTestChannelTest < ActionCable::Channel::TestCase + def setup + stub_connection user_id: 2016 + subscribe id: 5 + end + + def test_perform_with_params + perform :echo, text: "You are man!" + + assert_equal({ "text" => "You are man!" }, transmissions.last) + end + + def test_perform_and_transmit + perform :ping + + assert_equal "pong", transmissions.last["type"] + end +end + +class PerformUnsubscribedTestChannelTest < ActionCable::Channel::TestCase + tests PerformTestChannel + + def test_perform_when_unsubscribed + assert_raises do + perform :echo + end + end +end + +class BroadcastsTestChannel < ActionCable::Channel::Base + def broadcast(data) + ActionCable.server.broadcast( + "broadcast_#{params[:id]}", + text: data["message"], user_id: user_id + ) + end + + def broadcast_to_user(data) + user = User.new user_id + + self.class.broadcast_to user, text: data["message"] + end +end + +class BroadcastsTestChannelTest < ActionCable::Channel::TestCase + def setup + stub_connection user_id: 2017 + subscribe id: 5 + end + + def test_broadcast_matchers_included + assert_broadcast_on("broadcast_5", user_id: 2017, text: "SOS") do + perform :broadcast, message: "SOS" + end + end + + def test_broadcast_to_object + user = User.new(2017) + + assert_broadcasts(user, 1) do + perform :broadcast_to_user, text: "SOS" + end + end + + def test_broadcast_to_object_with_data + user = User.new(2017) + + assert_broadcast_on(user, text: "SOS") do + perform :broadcast_to_user, message: "SOS" + end + end +end diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb new file mode 100644 index 0000000000..e5f43488c4 --- /dev/null +++ b/actioncable/test/client_test.rb @@ -0,0 +1,314 @@ +# frozen_string_literal: true + +require "test_helper" +require "concurrent" + +require "websocket-client-simple" +require "json" + +require "active_support/hash_with_indifferent_access" + +#### +# 😷 Warning suppression 😷 +WebSocket::Frame::Handler::Handler03.prepend Module.new { + def initialize(*) + @application_data_buffer = nil + super + end +} + +WebSocket::Frame::Data.prepend Module.new { + def initialize(*) + @masking_key = nil + super + end +} +# +#### + +class ClientTest < ActionCable::TestCase + WAIT_WHEN_EXPECTING_EVENT = 2 + WAIT_WHEN_NOT_EXPECTING_EVENT = 0.5 + + class EchoChannel < ActionCable::Channel::Base + def subscribed + stream_from "global" + end + + def unsubscribed + "Goodbye from EchoChannel!" + end + + def ding(data) + transmit(dong: data["message"]) + end + + def delay(data) + sleep 1 + transmit(dong: data["message"]) + end + + def bulk(data) + ActionCable.server.broadcast "global", wide: data["message"] + end + end + + def setup + ActionCable.instance_variable_set(:@server, nil) + server = ActionCable.server + server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } + + server.config.cable = ActiveSupport::HashWithIndifferentAccess.new(adapter: "async") + + # and now the "real" setup for our test: + server.config.disable_request_forgery_protection = true + end + + def with_puma_server(rack_app = ActionCable.server, port = 3099) + server = ::Puma::Server.new(rack_app, ::Puma::Events.strings) + server.add_tcp_listener "127.0.0.1", port + server.min_threads = 1 + server.max_threads = 4 + + thread = server.run + + begin + yield port + + ensure + server.stop + + begin + thread.join + + rescue IOError + # Work around https://bugs.ruby-lang.org/issues/13405 + # + # Puma's sometimes raising while shutting down, when it closes + # its internal pipe. We can safely ignore that, but we do need + # to do the step skipped by the exception: + server.binder.close + + rescue RuntimeError => ex + # Work around https://bugs.ruby-lang.org/issues/13239 + raise unless ex.message =~ /can't modify frozen IOError/ + + # Handle this as if it were the IOError: do the same as above. + server.binder.close + end + end + end + + class SyncClient + attr_reader :pings + + def initialize(port) + messages = @messages = Queue.new + closed = @closed = Concurrent::Event.new + has_messages = @has_messages = Concurrent::Semaphore.new(0) + pings = @pings = Concurrent::AtomicFixnum.new(0) + + open = Concurrent::Promise.new + + @ws = WebSocket::Client::Simple.connect("ws://127.0.0.1:#{port}/") do |ws| + ws.on(:error) do |event| + event = RuntimeError.new(event.message) unless event.is_a?(Exception) + + if open.pending? + open.fail(event) + else + messages << event + has_messages.release + end + end + + ws.on(:open) do |event| + open.set(true) + end + + ws.on(:message) do |event| + if event.type == :close + closed.set + else + message = JSON.parse(event.data) + if message["type"] == "ping" + pings.increment + else + messages << message + has_messages.release + end + end + end + + ws.on(:close) do |event| + closed.set + end + end + + open.wait!(WAIT_WHEN_EXPECTING_EVENT) + end + + def read_message + @has_messages.try_acquire(1, WAIT_WHEN_EXPECTING_EVENT) + + msg = @messages.pop(true) + raise msg if msg.is_a?(Exception) + + msg + end + + def read_messages(expected_size = 0) + list = [] + loop do + 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 + end + list + end + + def send_message(message) + @ws.send(JSON.generate(message)) + end + + def close + sleep WAIT_WHEN_NOT_EXPECTING_EVENT + + unless @messages.empty? + raise "#{@messages.size} messages unprocessed" + 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 websocket_client(port) + SyncClient.new(port) + end + + def concurrently(enum) + enum.map { |*x| Concurrent::Future.execute { yield(*x) } }.map(&:value!) + end + + def test_single_client + with_puma_server do |port| + c = websocket_client(port) + assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack + c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") + assert_equal({ "identifier" => "{\"channel\":\"ClientTest::EchoChannel\"}", "type" => "confirm_subscription" }, c.read_message) + c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "ding", message: "hello") + assert_equal({ "identifier" => "{\"channel\":\"ClientTest::EchoChannel\"}", "message" => { "dong" => "hello" } }, c.read_message) + c.close + end + end + + def test_interacting_clients + with_puma_server do |port| + clients = concurrently(10.times) { websocket_client(port) } + + barrier_1 = Concurrent::CyclicBarrier.new(clients.size) + barrier_2 = Concurrent::CyclicBarrier.new(clients.size) + + concurrently(clients) do |c| + assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack + c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") + assert_equal({ "identifier" => '{"channel":"ClientTest::EchoChannel"}', "type" => "confirm_subscription" }, c.read_message) + c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "ding", message: "hello") + assert_equal({ "identifier" => '{"channel":"ClientTest::EchoChannel"}', "message" => { "dong" => "hello" } }, c.read_message) + barrier_1.wait WAIT_WHEN_EXPECTING_EVENT + c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "bulk", message: "hello") + barrier_2.wait WAIT_WHEN_EXPECTING_EVENT + assert_equal clients.size, c.read_messages(clients.size).size + end + + concurrently(clients, &:close) + end + end + + def test_many_clients + with_puma_server do |port| + clients = concurrently(100.times) { websocket_client(port) } + + concurrently(clients) do |c| + assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack + c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") + assert_equal({ "identifier" => '{"channel":"ClientTest::EchoChannel"}', "type" => "confirm_subscription" }, c.read_message) + c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "ding", message: "hello") + assert_equal({ "identifier" => '{"channel":"ClientTest::EchoChannel"}', "message" => { "dong" => "hello" } }, c.read_message) + end + + concurrently(clients, &:close) + end + end + + def test_disappearing_client + with_puma_server do |port| + c = websocket_client(port) + assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack + c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") + assert_equal({ "identifier" => "{\"channel\":\"ClientTest::EchoChannel\"}", "type" => "confirm_subscription" }, c.read_message) + c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "delay", message: "hello") + c.close # disappear before write + + c = websocket_client(port) + assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack + c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") + assert_equal({ "identifier" => "{\"channel\":\"ClientTest::EchoChannel\"}", "type" => "confirm_subscription" }, c.read_message) + c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "ding", message: "hello") + assert_equal({ "identifier" => '{"channel":"ClientTest::EchoChannel"}', "message" => { "dong" => "hello" } }, c.read_message) + c.close # disappear before read + end + end + + def test_unsubscribe_client + with_puma_server do |port| + app = ActionCable.server + identifier = JSON.generate(channel: "ClientTest::EchoChannel") + + c = websocket_client(port) + assert_equal({ "type" => "welcome" }, c.read_message) + c.send_message command: "subscribe", identifier: identifier + assert_equal({ "identifier" => "{\"channel\":\"ClientTest::EchoChannel\"}", "type" => "confirm_subscription" }, c.read_message) + assert_equal(1, app.connections.count) + assert(app.remote_connections.where(identifier: identifier)) + + subscriptions = app.connections.first.subscriptions.send(:subscriptions) + assert_not_equal 0, subscriptions.size, "Missing EchoChannel subscription" + channel = subscriptions.first[1] + assert_called(channel, :unsubscribed) do + c.close + sleep 0.1 # Data takes a moment to process + end + + # 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 = websocket_client(port) + assert_equal({ "type" => "welcome" }, c.read_message) + c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") + assert_equal({ "identifier" => "{\"channel\":\"ClientTest::EchoChannel\"}", "type" => "confirm_subscription" }, c.read_message) + + ActionCable.server.restart + c.wait_for_close + assert_predicate c, :closed? + end + end +end diff --git a/actioncable/test/connection/authorization_test.rb b/actioncable/test/connection/authorization_test.rb new file mode 100644 index 0000000000..ac5c128135 --- /dev/null +++ b/actioncable/test/connection/authorization_test.rb @@ -0,0 +1,36 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" + +class ActionCable::Connection::AuthorizationTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + attr_reader :websocket + + def connect + reject_unauthorized_connection + end + + def send_async(method, *args) + send method, *args + end + end + + test "unauthorized connection" do + run_in_eventmachine do + server = TestServer.new + server.config.allowed_request_origins = %w( http://rubyonrails.com ) + + env = Rack::MockRequest.env_for "/test", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket", + "HTTP_HOST" => "localhost", "HTTP_ORIGIN" => "http://rubyonrails.com" + + connection = Connection.new(server, env) + + assert_called_with(connection.websocket, :transmit, [{ type: "disconnect", reason: "unauthorized", reconnect: false }.to_json]) do + assert_called(connection.websocket, :close) do + connection.process + end + end + end + end +end diff --git a/actioncable/test/connection/base_test.rb b/actioncable/test/connection/base_test.rb new file mode 100644 index 0000000000..299879ad4c --- /dev/null +++ b/actioncable/test/connection/base_test.rb @@ -0,0 +1,143 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" +require "active_support/core_ext/object/json" + +class ActionCable::Connection::BaseTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + attr_reader :websocket, :subscriptions, :message_buffer, :connected + + def connect + @connected = true + end + + def disconnect + @connected = false + end + + def send_async(method, *args) + send method, *args + end + end + + setup do + @server = TestServer.new + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + end + + test "making a connection with invalid headers" do + run_in_eventmachine do + connection = ActionCable::Connection::Base.new(@server, Rack::MockRequest.env_for("/test")) + response = connection.process + assert_equal 404, response[0] + end + end + + test "websocket connection" do + run_in_eventmachine do + connection = open_connection + connection.process + + assert_predicate connection.websocket, :possible? + + wait_for_async + assert_predicate connection.websocket, :alive? + end + end + + test "rack response" do + run_in_eventmachine do + connection = open_connection + response = connection.process + + assert_equal [ -1, {}, [] ], response + end + end + + test "on connection open" do + run_in_eventmachine do + connection = open_connection + + assert_called_with(connection.websocket, :transmit, [{ type: "welcome" }.to_json]) do + assert_called(connection.message_buffer, :process!) do + connection.process + wait_for_async + end + end + + assert_equal [ connection ], @server.connections + assert connection.connected + end + end + + test "on connection close" do + run_in_eventmachine do + connection = open_connection + connection.process + + # Setup the connection + connection.send :handle_open + assert connection.connected + + assert_called(connection.subscriptions, :unsubscribe_from_all) do + connection.send :handle_close + end + + assert_not connection.connected + assert_equal [], @server.connections + end + end + + test "connection statistics" do + run_in_eventmachine do + connection = open_connection + connection.process + + statistics = connection.statistics + + assert_predicate statistics[:identifier], :blank? + assert_kind_of Time, statistics[:started_at] + assert_equal [], statistics[:subscriptions] + end + end + + test "explicitly closing a connection" do + run_in_eventmachine do + connection = open_connection + connection.process + + assert_called(connection.websocket, :close) do + connection.close(reason: "testing") + end + 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_HOST" => "localhost", "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", + "HTTP_HOST" => "localhost", "HTTP_ORIGIN" => "http://rubyonrails.com" + + Connection.new(@server, env) + end +end diff --git a/actioncable/test/connection/client_socket_test.rb b/actioncable/test/connection/client_socket_test.rb new file mode 100644 index 0000000000..1ab3c3b71d --- /dev/null +++ b/actioncable/test/connection/client_socket_test.rb @@ -0,0 +1,93 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" + +class ActionCable::Connection::ClientSocketTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + attr_reader :connected, :websocket, :errors + + def initialize(*) + super + @errors = [] + end + + def connect + @connected = true + end + + def disconnect + @connected = false + end + + def send_async(method, *args) + send method, *args + end + + def on_error(message) + @errors << message + end + end + + setup do + @server = TestServer.new + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + end + + test "delegate socket errors to on_error handler" do + run_in_eventmachine do + connection = open_connection + + # Internal hax = :( + client = connection.websocket.send(:websocket) + client.instance_variable_get("@stream").stub(:write, proc { raise "foo" }) do + assert_not_called(client, :client_gone) do + client.write("boo") + end + end + assert_equal %w[ foo ], connection.errors + end + end + + test "closes hijacked i/o socket at shutdown" do + run_in_eventmachine do + connection = open_connection + + client = connection.websocket.send(:websocket) + event = Concurrent::Event.new + client.instance_variable_get("@stream") + .instance_variable_get("@rack_hijack_io") + .define_singleton_method(:close) { event.set } + connection.close(reason: "testing") + event.wait + end + end + + private + def open_connection + env = Rack::MockRequest.env_for "/test", + "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket", + "HTTP_HOST" => "localhost", "HTTP_ORIGIN" => "http://rubyonrails.com" + io, client_io = \ + begin + Socket.pair(Socket::AF_UNIX, Socket::SOCK_STREAM, 0) + rescue + StringIO.new + end + env["rack.hijack"] = -> { env["rack.hijack_io"] = io } + + Connection.new(@server, env).tap do |connection| + connection.process + if client_io + # Make sure server returns handshake response + Timeout.timeout(1) do + loop do + break if client_io.readline == "\r\n" + end + end + end + connection.send :handle_open + assert connection.connected + end + end +end diff --git a/actioncable/test/connection/cross_site_forgery_test.rb b/actioncable/test/connection/cross_site_forgery_test.rb new file mode 100644 index 0000000000..3e21138ffc --- /dev/null +++ b/actioncable/test/connection/cross_site_forgery_test.rb @@ -0,0 +1,92 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" + +class ActionCable::Connection::CrossSiteForgeryTest < ActionCable::TestCase + HOST = "rubyonrails.com" + + class Connection < ActionCable::Connection::Base + def send_async(method, *args) + send method, *args + end + end + + setup do + @server = TestServer.new + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + @server.config.allow_same_origin_as_host = false + end + + teardown do + @server.config.disable_request_forgery_protection = false + @server.config.allowed_request_origins = [] + @server.config.allow_same_origin_as_host = true + end + + test "disable forgery protection" do + @server.config.disable_request_forgery_protection = true + assert_origin_allowed "http://rubyonrails.com" + assert_origin_allowed "http://hax.com" + end + + test "explicitly specified a single allowed origin" do + @server.config.allowed_request_origins = "http://hax.com" + assert_origin_not_allowed "http://rubyonrails.com" + assert_origin_allowed "http://hax.com" + end + + test "explicitly specified multiple allowed origins" do + @server.config.allowed_request_origins = %w( http://rubyonrails.com http://www.rubyonrails.com ) + assert_origin_allowed "http://rubyonrails.com" + assert_origin_allowed "http://www.rubyonrails.com" + assert_origin_not_allowed "http://hax.com" + end + + test "explicitly specified a single regexp allowed origin" do + @server.config.allowed_request_origins = /.*ha.*/ + assert_origin_not_allowed "http://rubyonrails.com" + assert_origin_allowed "http://hax.com" + end + + test "explicitly specified multiple regexp allowed origins" do + @server.config.allowed_request_origins = [/http:\/\/ruby.*/, /.*rai.s.*com/, "string" ] + assert_origin_allowed "http://rubyonrails.com" + assert_origin_allowed "http://www.rubyonrails.com" + assert_origin_not_allowed "http://hax.com" + assert_origin_not_allowed "http://rails.co.uk" + end + + test "allow same origin as host" do + @server.config.allow_same_origin_as_host = true + assert_origin_allowed "http://#{HOST}" + assert_origin_not_allowed "http://hax.com" + assert_origin_not_allowed "http://rails.co.uk" + end + + private + def assert_origin_allowed(origin) + response = connect_with_origin origin + assert_equal(-1, response[0]) + end + + def assert_origin_not_allowed(origin) + response = connect_with_origin origin + assert_equal 404, response[0] + end + + def connect_with_origin(origin) + response = nil + + run_in_eventmachine do + response = Connection.new(@server, env_for_origin(origin)).process + end + + response + end + + def env_for_origin(origin) + Rack::MockRequest.env_for "/test", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket", "SERVER_NAME" => HOST, + "HTTP_HOST" => HOST, "HTTP_ORIGIN" => origin + end +end diff --git a/actioncable/test/connection/identifier_test.rb b/actioncable/test/connection/identifier_test.rb new file mode 100644 index 0000000000..707f4bab72 --- /dev/null +++ b/actioncable/test/connection/identifier_test.rb @@ -0,0 +1,77 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" +require "stubs/user" + +class ActionCable::Connection::IdentifierTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + identified_by :current_user + attr_reader :websocket + + public :process_internal_message + + def connect + self.current_user = User.new "lifo" + end + end + + test "connection identifier" do + run_in_eventmachine do + open_connection + assert_equal "User#lifo", @connection.connection_identifier + end + end + + test "should subscribe to internal channel on open and unsubscribe on close" do + run_in_eventmachine do + server = TestServer.new + + open_connection(server) + close_connection + wait_for_async + + %w[subscribe unsubscribe].each do |method| + pubsub_call = server.pubsub.class.class_variable_get "@@#{method}_called" + + assert_equal "action_cable/User#lifo", pubsub_call[:channel] + assert_instance_of Proc, pubsub_call[:callback] + end + end + end + + test "processing disconnect message" do + run_in_eventmachine do + open_connection + + assert_called(@connection.websocket, :close) do + @connection.process_internal_message "type" => "disconnect" + end + end + end + + test "processing invalid message" do + run_in_eventmachine do + open_connection + + assert_not_called(@connection.websocket, :close) do + @connection.process_internal_message "type" => "unknown" + end + end + end + + private + def open_connection(server = nil) + server ||= TestServer.new + + env = Rack::MockRequest.env_for "/test", "HTTP_HOST" => "localhost", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket" + @connection = Connection.new(server, env) + + @connection.process + @connection.send :handle_open + end + + def close_connection + @connection.send :handle_close + end +end diff --git a/actioncable/test/connection/multiple_identifiers_test.rb b/actioncable/test/connection/multiple_identifiers_test.rb new file mode 100644 index 0000000000..51716410b2 --- /dev/null +++ b/actioncable/test/connection/multiple_identifiers_test.rb @@ -0,0 +1,34 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" +require "stubs/user" + +class ActionCable::Connection::MultipleIdentifiersTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + identified_by :current_user, :current_room + + def connect + self.current_user = User.new "lifo" + self.current_room = Room.new "my", "room" + end + end + + test "multiple connection identifiers" do + run_in_eventmachine do + open_connection + + assert_equal "Room#my-room:User#lifo", @connection.connection_identifier + end + end + + private + def open_connection + server = TestServer.new + env = Rack::MockRequest.env_for "/test", "HTTP_HOST" => "localhost", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket" + @connection = Connection.new(server, env) + + @connection.process + @connection.send :handle_open + end +end diff --git a/actioncable/test/connection/stream_test.rb b/actioncable/test/connection/stream_test.rb new file mode 100644 index 0000000000..0f4576db40 --- /dev/null +++ b/actioncable/test/connection/stream_test.rb @@ -0,0 +1,69 @@ +# frozen_string_literal: true + +require "test_helper" +require "minitest/mock" +require "stubs/test_server" + +class ActionCable::Connection::StreamTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + attr_reader :connected, :websocket, :errors + + def initialize(*) + super + @errors = [] + end + + def connect + @connected = true + end + + def disconnect + @connected = false + end + + def send_async(method, *args) + send method, *args + end + + def on_error(message) + @errors << message + end + end + + setup do + @server = TestServer.new + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + end + + [ EOFError, Errno::ECONNRESET ].each do |closed_exception| + test "closes socket on #{closed_exception}" do + run_in_eventmachine do + connection = open_connection + + # Internal hax = :( + client = connection.websocket.send(:websocket) + rack_hijack_io = client.instance_variable_get("@stream").instance_variable_get("@rack_hijack_io") + rack_hijack_io.stub(:write, proc { raise(closed_exception, "foo") }) do + assert_called(client, :client_gone) do + client.write("boo") + end + end + assert_equal [], connection.errors + end + end + end + + private + def open_connection + env = Rack::MockRequest.env_for "/test", + "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket", + "HTTP_HOST" => "localhost", "HTTP_ORIGIN" => "http://rubyonrails.com" + env["rack.hijack"] = -> { env["rack.hijack_io"] = StringIO.new } + + Connection.new(@server, env).tap do |connection| + connection.process + connection.send :handle_open + assert connection.connected + end + end +end diff --git a/actioncable/test/connection/string_identifier_test.rb b/actioncable/test/connection/string_identifier_test.rb new file mode 100644 index 0000000000..f7019b926a --- /dev/null +++ b/actioncable/test/connection/string_identifier_test.rb @@ -0,0 +1,36 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" + +class ActionCable::Connection::StringIdentifierTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + identified_by :current_token + + def connect + self.current_token = "random-string" + end + + def send_async(method, *args) + send method, *args + end + end + + test "connection identifier" do + run_in_eventmachine do + open_connection + + assert_equal "random-string", @connection.connection_identifier + end + end + + private + def open_connection + server = TestServer.new + env = Rack::MockRequest.env_for "/test", "HTTP_HOST" => "localhost", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket" + @connection = Connection.new(server, env) + + @connection.process + @connection.send :on_open + end +end diff --git a/actioncable/test/connection/subscriptions_test.rb b/actioncable/test/connection/subscriptions_test.rb new file mode 100644 index 0000000000..902085c5d6 --- /dev/null +++ b/actioncable/test/connection/subscriptions_test.rb @@ -0,0 +1,119 @@ +# frozen_string_literal: true + +require "test_helper" + +class ActionCable::Connection::SubscriptionsTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + attr_reader :websocket + + def send_async(method, *args) + send method, *args + end + end + + class ChatChannel < ActionCable::Channel::Base + attr_reader :room, :lines + + def subscribed + @room = Room.new params[:id] + @lines = [] + end + + def speak(data) + @lines << data + end + end + + setup do + @server = TestServer.new + + @chat_identifier = ActiveSupport::JSON.encode(id: 1, channel: "ActionCable::Connection::SubscriptionsTest::ChatChannel") + end + + test "subscribe command" do + run_in_eventmachine do + setup_connection + channel = subscribe_to_chat_channel + + assert_kind_of ChatChannel, channel + assert_equal 1, channel.room.id + end + end + + test "subscribe command without an identifier" do + run_in_eventmachine do + setup_connection + + @subscriptions.execute_command "command" => "subscribe" + assert_empty @subscriptions.identifiers + end + end + + test "unsubscribe command" do + run_in_eventmachine do + setup_connection + subscribe_to_chat_channel + + channel = subscribe_to_chat_channel + + assert_called(channel, :unsubscribe_from_channel) do + @subscriptions.execute_command "command" => "unsubscribe", "identifier" => @chat_identifier + end + + assert_empty @subscriptions.identifiers + end + end + + test "unsubscribe command without an identifier" do + run_in_eventmachine do + setup_connection + + @subscriptions.execute_command "command" => "unsubscribe" + assert_empty @subscriptions.identifiers + end + end + + test "message command" do + run_in_eventmachine do + setup_connection + channel = subscribe_to_chat_channel + + data = { "content" => "Hello World!", "action" => "speak" } + @subscriptions.execute_command "command" => "message", "identifier" => @chat_identifier, "data" => ActiveSupport::JSON.encode(data) + + assert_equal [ data ], channel.lines + end + end + + test "unsubscribe from all" do + run_in_eventmachine do + setup_connection + + channel1 = subscribe_to_chat_channel + + channel2_id = ActiveSupport::JSON.encode(id: 2, channel: "ActionCable::Connection::SubscriptionsTest::ChatChannel") + channel2 = subscribe_to_chat_channel(channel2_id) + + assert_called(channel1, :unsubscribe_from_channel) do + assert_called(channel2, :unsubscribe_from_channel) do + @subscriptions.unsubscribe_from_all + end + end + end + end + + private + def subscribe_to_chat_channel(identifier = @chat_identifier) + @subscriptions.execute_command "command" => "subscribe", "identifier" => identifier + assert_equal identifier, @subscriptions.identifiers.last + + @subscriptions.send :find, "identifier" => identifier + end + + def setup_connection + env = Rack::MockRequest.env_for "/test", "HTTP_HOST" => "localhost", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket" + @connection = Connection.new(@server, env) + + @subscriptions = ActionCable::Connection::Subscriptions.new(@connection) + end +end diff --git a/actioncable/test/javascript/src/test.js b/actioncable/test/javascript/src/test.js new file mode 100644 index 0000000000..eea1c0a408 --- /dev/null +++ b/actioncable/test/javascript/src/test.js @@ -0,0 +1,6 @@ +import "./test_helpers/index" +import "./unit/action_cable_test" +import "./unit/connection_test" +import "./unit/consumer_test" +import "./unit/subscription_test" +import "./unit/subscriptions_test" diff --git a/actioncable/test/javascript/src/test_helpers/consumer_test_helper.js b/actioncable/test/javascript/src/test_helpers/consumer_test_helper.js new file mode 100644 index 0000000000..d1dabc9fc4 --- /dev/null +++ b/actioncable/test/javascript/src/test_helpers/consumer_test_helper.js @@ -0,0 +1,58 @@ +import { WebSocket as MockWebSocket, Server as MockServer } from "mock-socket" +import * as ActionCable from "../../../../app/javascript/action_cable/index" +import {defer, testURL} from "./index" + +export default function(name, options, callback) { + if (options == null) { options = {} } + if (callback == null) { + callback = options + options = {} + } + + if (options.url == null) { options.url = testURL } + + return QUnit.test(name, function(assert) { + const doneAsync = assert.async() + + ActionCable.adapters.WebSocket = MockWebSocket + const server = new MockServer(options.url) + const consumer = ActionCable.createConsumer(options.url) + + server.on("connection", function() { + const clients = server.clients() + assert.equal(clients.length, 1) + assert.equal(clients[0].readyState, WebSocket.OPEN) + }) + + server.broadcastTo = function(subscription, data, callback) { + if (data == null) { data = {} } + data.identifier = subscription.identifier + + if (data.message_type) { + data.type = ActionCable.INTERNAL.message_types[data.message_type] + delete data.message_type + } + + server.send(JSON.stringify(data)) + defer(callback) + } + + const done = function() { + consumer.disconnect() + server.close() + doneAsync() + } + + const testData = {assert, consumer, server, done} + + if (options.connect === false) { + callback(testData) + } else { + server.on("connection", function() { + testData.client = server.clients()[0] + callback(testData) + }) + consumer.connect() + } + }) +} diff --git a/actioncable/test/javascript/src/test_helpers/index.js b/actioncable/test/javascript/src/test_helpers/index.js new file mode 100644 index 0000000000..0cd4e260b3 --- /dev/null +++ b/actioncable/test/javascript/src/test_helpers/index.js @@ -0,0 +1,10 @@ +import * as ActionCable from "../../../../app/javascript/action_cable/index" + +export const testURL = "ws://cable.example.com/" + +export function defer(callback) { + setTimeout(callback, 1) +} + +const originalWebSocket = ActionCable.adapters.WebSocket +QUnit.testDone(() => ActionCable.adapters.WebSocket = originalWebSocket) diff --git a/actioncable/test/javascript/src/unit/action_cable_test.js b/actioncable/test/javascript/src/unit/action_cable_test.js new file mode 100644 index 0000000000..daad900aca --- /dev/null +++ b/actioncable/test/javascript/src/unit/action_cable_test.js @@ -0,0 +1,45 @@ +import * as ActionCable from "../../../../app/javascript/action_cable/index" +import {testURL} from "../test_helpers/index" + +const {module, test} = QUnit + +module("ActionCable", () => { + module("Adapters", () => { + module("WebSocket", () => { + test("default is window.WebSocket", assert => { + assert.equal(ActionCable.adapters.WebSocket, window.WebSocket) + }) + }) + + module("logger", () => { + test("default is window.console", assert => { + assert.equal(ActionCable.adapters.logger, window.console) + }) + }) + }) + + module("#createConsumer", () => { + test("uses specified URL", assert => { + const consumer = ActionCable.createConsumer(testURL) + assert.equal(consumer.url, testURL) + }) + + test("uses default URL", assert => { + const pattern = new RegExp(`${ActionCable.INTERNAL.default_mount_path}$`) + const consumer = ActionCable.createConsumer() + assert.ok(pattern.test(consumer.url), `Expected ${consumer.url} to match ${pattern}`) + }) + + test("uses URL from meta tag", assert => { + const element = document.createElement("meta") + element.setAttribute("name", "action-cable-url") + element.setAttribute("content", testURL) + + document.head.appendChild(element) + const consumer = ActionCable.createConsumer() + document.head.removeChild(element) + + assert.equal(consumer.url, testURL) + }) + }) +}) diff --git a/actioncable/test/javascript/src/unit/connection_test.js b/actioncable/test/javascript/src/unit/connection_test.js new file mode 100644 index 0000000000..9b1a975bfb --- /dev/null +++ b/actioncable/test/javascript/src/unit/connection_test.js @@ -0,0 +1,28 @@ +import * as ActionCable from "../../../../app/javascript/action_cable/index" + +const {module, test} = QUnit + +module("ActionCable.Connection", () => { + module("#getState", () => { + test("uses the configured WebSocket adapter", assert => { + ActionCable.adapters.WebSocket = { foo: 1, BAR: "42" } + const connection = new ActionCable.Connection({}) + connection.webSocket = {} + connection.webSocket.readyState = 1 + assert.equal(connection.getState(), "foo") + connection.webSocket.readyState = "42" + assert.equal(connection.getState(), "bar") + }) + }) + + module("#open", () => { + test("uses the configured WebSocket adapter", assert => { + const FakeWebSocket = function() {} + ActionCable.adapters.WebSocket = FakeWebSocket + const connection = new ActionCable.Connection({}) + connection.monitor = { start() {} } + connection.open() + assert.equal(connection.webSocket instanceof FakeWebSocket, true) + }) + }) +}) diff --git a/actioncable/test/javascript/src/unit/consumer_test.js b/actioncable/test/javascript/src/unit/consumer_test.js new file mode 100644 index 0000000000..acc618bf0c --- /dev/null +++ b/actioncable/test/javascript/src/unit/consumer_test.js @@ -0,0 +1,19 @@ +import consumerTest from "../test_helpers/consumer_test_helper" + +const {module} = QUnit + +module("ActionCable.Consumer", () => { + consumerTest("#connect", {connect: false}, ({consumer, server, assert, done}) => { + server.on("connection", () => { + assert.equal(consumer.connect(), false) + done() + }) + + consumer.connect() + }) + + consumerTest("#disconnect", ({consumer, client, done}) => { + client.addEventListener("close", done) + consumer.disconnect() + }) +}) diff --git a/actioncable/test/javascript/src/unit/subscription_test.js b/actioncable/test/javascript/src/unit/subscription_test.js new file mode 100644 index 0000000000..bf32e5f27d --- /dev/null +++ b/actioncable/test/javascript/src/unit/subscription_test.js @@ -0,0 +1,54 @@ +import consumerTest from "../test_helpers/consumer_test_helper" + +const {module} = QUnit + +module("ActionCable.Subscription", () => { + consumerTest("#initialized callback", ({server, consumer, assert, done}) => + consumer.subscriptions.create("chat", { + initialized() { + assert.ok(true) + done() + } + }) + ) + + consumerTest("#connected callback", ({server, consumer, assert, done}) => { + const subscription = consumer.subscriptions.create("chat", { + connected() { + assert.ok(true) + done() + } + }) + + server.broadcastTo(subscription, {message_type: "confirmation"}) + }) + + consumerTest("#disconnected callback", ({server, consumer, assert, done}) => { + const subscription = consumer.subscriptions.create("chat", { + disconnected() { + assert.ok(true) + done() + } + }) + + server.broadcastTo(subscription, {message_type: "confirmation"}, () => server.close()) + }) + + consumerTest("#perform", ({consumer, server, assert, done}) => { + const subscription = consumer.subscriptions.create("chat", { + connected() { + this.perform({publish: "hi"}) + } + }) + + server.on("message", (message) => { + const data = JSON.parse(message) + assert.equal(data.identifier, subscription.identifier) + assert.equal(data.command, "message") + assert.deepEqual(data.data, JSON.stringify({action: { publish: "hi" }})) + done() + }) + + server.broadcastTo(subscription, {message_type: "confirmation"}) + }) +}) diff --git a/actioncable/test/javascript/src/unit/subscriptions_test.js b/actioncable/test/javascript/src/unit/subscriptions_test.js new file mode 100644 index 0000000000..33af5d4d82 --- /dev/null +++ b/actioncable/test/javascript/src/unit/subscriptions_test.js @@ -0,0 +1,31 @@ +import consumerTest from "../test_helpers/consumer_test_helper" + +const {module} = QUnit + +module("ActionCable.Subscriptions", () => { + consumerTest("create subscription with channel string", ({consumer, server, assert, done}) => { + const channel = "chat" + + server.on("message", (message) => { + const data = JSON.parse(message) + assert.equal(data.command, "subscribe") + assert.equal(data.identifier, JSON.stringify({channel})) + done() + }) + + consumer.subscriptions.create(channel) + }) + + consumerTest("create subscription with channel object", ({consumer, server, assert, done}) => { + const channel = {channel: "chat", room: "action"} + + server.on("message", (message) => { + const data = JSON.parse(message) + assert.equal(data.command, "subscribe") + assert.equal(data.identifier, JSON.stringify(channel)) + done() + }) + + consumer.subscriptions.create(channel) + }) +}) diff --git a/actioncable/test/server/base_test.rb b/actioncable/test/server/base_test.rb new file mode 100644 index 0000000000..d46debea45 --- /dev/null +++ b/actioncable/test/server/base_test.rb @@ -0,0 +1,38 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" +require "active_support/core_ext/hash/indifferent_access" + +class BaseTest < ActionCable::TestCase + def setup + @server = ActionCable::Server::Base.new + @server.config.cable = { adapter: "async" }.with_indifferent_access + end + + class FakeConnection + def close + end + end + + test "#restart closes all open connections" do + conn = FakeConnection.new + @server.add_connection(conn) + + assert_called(conn, :close) do + @server.restart + end + end + + test "#restart shuts down worker pool" do + assert_called(@server.worker_pool, :halt) do + @server.restart + end + end + + test "#restart shuts down pub/sub adapter" do + assert_called(@server.pubsub, :shutdown) do + @server.restart + end + end +end diff --git a/actioncable/test/server/broadcasting_test.rb b/actioncable/test/server/broadcasting_test.rb new file mode 100644 index 0000000000..860e79b821 --- /dev/null +++ b/actioncable/test/server/broadcasting_test.rb @@ -0,0 +1,58 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" + +class BroadcastingTest < ActionCable::TestCase + 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 + + test "broadcast generates notification" do + server = TestServer.new + + events = [] + ActiveSupport::Notifications.subscribe "broadcast.action_cable" do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + broadcasting = "test_queue" + message = { body: "test message" } + server.broadcast(broadcasting, message) + + assert_equal 1, events.length + assert_equal "broadcast.action_cable", events[0].name + assert_equal broadcasting, events[0].payload[:broadcasting] + assert_equal message, events[0].payload[:message] + assert_equal ActiveSupport::JSON, events[0].payload[:coder] + ensure + ActiveSupport::Notifications.unsubscribe "broadcast.action_cable" + end + + test "broadcaster from broadcaster_for generates notification" do + server = TestServer.new + + events = [] + ActiveSupport::Notifications.subscribe "broadcast.action_cable" do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + broadcasting = "test_queue" + message = { body: "test message" } + + broadcaster = server.broadcaster_for(broadcasting) + broadcaster.broadcast(message) + + assert_equal 1, events.length + assert_equal "broadcast.action_cable", events[0].name + assert_equal broadcasting, events[0].payload[:broadcasting] + assert_equal message, events[0].payload[:message] + assert_equal ActiveSupport::JSON, events[0].payload[:coder] + ensure + ActiveSupport::Notifications.unsubscribe "broadcast.action_cable" + end +end diff --git a/actioncable/test/stubs/global_id.rb b/actioncable/test/stubs/global_id.rb new file mode 100644 index 0000000000..15fab6b8a7 --- /dev/null +++ b/actioncable/test/stubs/global_id.rb @@ -0,0 +1,10 @@ +# frozen_string_literal: true + +class GlobalID + attr_reader :uri + delegate :to_param, :to_s, to: :uri + + def initialize(gid, options = {}) + @uri = gid + end +end diff --git a/actioncable/test/stubs/room.rb b/actioncable/test/stubs/room.rb new file mode 100644 index 0000000000..df7236f408 --- /dev/null +++ b/actioncable/test/stubs/room.rb @@ -0,0 +1,18 @@ +# frozen_string_literal: true + +class Room + attr_reader :id, :name + + def initialize(id, name = "Campfire") + @id = id + @name = name + end + + def to_global_id + GlobalID.new("Room##{id}-#{name}") + end + + def to_gid_param + to_global_id.to_param + end +end diff --git a/actioncable/test/stubs/test_adapter.rb b/actioncable/test/stubs/test_adapter.rb new file mode 100644 index 0000000000..3b25c9168f --- /dev/null +++ b/actioncable/test/stubs/test_adapter.rb @@ -0,0 +1,14 @@ +# frozen_string_literal: true + +class SuccessAdapter < ActionCable::SubscriptionAdapter::Base + def broadcast(channel, payload) + end + + def subscribe(channel, callback, success_callback = nil) + @@subscribe_called = { channel: channel, callback: callback, success_callback: success_callback } + end + + def unsubscribe(channel, callback) + @@unsubscribe_called = { channel: channel, callback: callback } + end +end diff --git a/actioncable/test/stubs/test_connection.rb b/actioncable/test/stubs/test_connection.rb new file mode 100644 index 0000000000..155c68e38c --- /dev/null +++ b/actioncable/test/stubs/test_connection.rb @@ -0,0 +1,35 @@ +# frozen_string_literal: true + +require "stubs/user" + +class TestConnection + attr_reader :identifiers, :logger, :current_user, :server, :subscriptions, :transmissions + + delegate :pubsub, to: :server + + def initialize(user = User.new("lifo"), coder: ActiveSupport::JSON, subscription_adapter: SuccessAdapter) + @coder = coder + @identifiers = [ :current_user ] + + @current_user = user + @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new) + @server = TestServer.new(subscription_adapter: subscription_adapter) + @transmissions = [] + end + + def transmit(cable_message) + @transmissions << encode(cable_message) + end + + def last_transmission + decode @transmissions.last if @transmissions.any? + end + + def decode(websocket_message) + @coder.decode websocket_message + end + + def encode(cable_message) + @coder.encode cable_message + end +end diff --git a/actioncable/test/stubs/test_server.rb b/actioncable/test/stubs/test_server.rb new file mode 100644 index 0000000000..0bc4625e28 --- /dev/null +++ b/actioncable/test/stubs/test_server.rb @@ -0,0 +1,32 @@ +# frozen_string_literal: true + +require "ostruct" + +class TestServer + include ActionCable::Server::Connections + include ActionCable::Server::Broadcasting + + attr_reader :logger, :config, :mutex + + def initialize(subscription_adapter: SuccessAdapter) + @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new) + + @config = OpenStruct.new(log_tags: [], subscription_adapter: subscription_adapter) + + @mutex = Monitor.new + end + + def pubsub + @pubsub ||= @config.subscription_adapter.new(self) + end + + def event_loop + @event_loop ||= ActionCable::Connection::StreamEventLoop.new.tap do |loop| + loop.instance_variable_set(:@executor, Concurrent.global_io_executor) + end + end + + def worker_pool + @worker_pool ||= ActionCable::Server::Worker.new(max_size: 5) + end +end diff --git a/actioncable/test/stubs/user.rb b/actioncable/test/stubs/user.rb new file mode 100644 index 0000000000..7894d1d9ae --- /dev/null +++ b/actioncable/test/stubs/user.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +class User + attr_reader :name + + def initialize(name) + @name = name + end + + def to_global_id + GlobalID.new("User##{name}") + end + + def to_gid_param + to_global_id.to_param + end +end diff --git a/actioncable/test/subscription_adapter/async_test.rb b/actioncable/test/subscription_adapter/async_test.rb new file mode 100644 index 0000000000..6e038259b5 --- /dev/null +++ b/actioncable/test/subscription_adapter/async_test.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +require "test_helper" +require_relative "common" + +class AsyncAdapterTest < ActionCable::TestCase + include CommonSubscriptionAdapterTest + + def setup + super + + @tx_adapter.shutdown + @tx_adapter = @rx_adapter + end + + def cable_config + { adapter: "async" } + end +end diff --git a/actioncable/test/subscription_adapter/base_test.rb b/actioncable/test/subscription_adapter/base_test.rb new file mode 100644 index 0000000000..999dc0cba1 --- /dev/null +++ b/actioncable/test/subscription_adapter/base_test.rb @@ -0,0 +1,65 @@ +# frozen_string_literal: true + +require "test_helper" +require "stubs/test_server" + +class ActionCable::SubscriptionAdapter::BaseTest < ActionCable::TestCase + ## TEST THAT ERRORS ARE RETURNED FOR INHERITORS THAT DON'T OVERRIDE METHODS + + class BrokenAdapter < ActionCable::SubscriptionAdapter::Base + end + + setup do + @server = TestServer.new + @server.config.subscription_adapter = BrokenAdapter + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + end + + test "#broadcast returns NotImplementedError by default" do + assert_raises NotImplementedError do + BrokenAdapter.new(@server).broadcast("channel", "payload") + end + end + + test "#subscribe returns NotImplementedError by default" do + callback = lambda { puts "callback" } + success_callback = lambda { puts "success" } + + assert_raises NotImplementedError do + BrokenAdapter.new(@server).subscribe("channel", callback, success_callback) + end + end + + test "#unsubscribe returns NotImplementedError by default" do + callback = lambda { puts "callback" } + + assert_raises NotImplementedError do + BrokenAdapter.new(@server).unsubscribe("channel", callback) + end + end + + # TEST METHODS THAT ARE REQUIRED OF THE ADAPTER'S BACKEND STORAGE OBJECT + + test "#broadcast is implemented" do + assert_nothing_raised do + SuccessAdapter.new(@server).broadcast("channel", "payload") + end + end + + test "#subscribe is implemented" do + callback = lambda { puts "callback" } + success_callback = lambda { puts "success" } + + assert_nothing_raised do + SuccessAdapter.new(@server).subscribe("channel", callback, success_callback) + end + end + + test "#unsubscribe is implemented" do + callback = lambda { puts "callback" } + + assert_nothing_raised do + SuccessAdapter.new(@server).unsubscribe("channel", callback) + end + end +end diff --git a/actioncable/test/subscription_adapter/channel_prefix.rb b/actioncable/test/subscription_adapter/channel_prefix.rb new file mode 100644 index 0000000000..3071facd9d --- /dev/null +++ b/actioncable/test/subscription_adapter/channel_prefix.rb @@ -0,0 +1,38 @@ +# frozen_string_literal: true + +require "test_helper" + +class ActionCable::Server::WithIndependentConfig < ActionCable::Server::Base + # ActionCable::Server::Base defines config as a class variable. + # Need config to be an instance variable here as we're testing 2 separate configs + def config + @config ||= ActionCable::Server::Configuration.new + end +end + +module ChannelPrefixTest + def test_channel_prefix + server2 = ActionCable::Server::WithIndependentConfig.new + server2.config.cable = alt_cable_config + server2.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } + + adapter_klass = server2.config.pubsub_adapter + + rx_adapter2 = adapter_klass.new(server2) + tx_adapter2 = adapter_klass.new(server2) + + subscribe_as_queue("channel") do |queue| + subscribe_as_queue("channel", rx_adapter2) do |queue2| + @tx_adapter.broadcast("channel", "hello world") + tx_adapter2.broadcast("channel", "hello world 2") + + assert_equal "hello world", queue.pop + assert_equal "hello world 2", queue2.pop + end + end + end + + def alt_cable_config + cable_config.merge(channel_prefix: "foo") + end +end diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb new file mode 100644 index 0000000000..b3e9ae9d5c --- /dev/null +++ b/actioncable/test/subscription_adapter/common.rb @@ -0,0 +1,131 @@ +# frozen_string_literal: true + +require "test_helper" +require "concurrent" + +require "active_support/core_ext/hash/indifferent_access" +require "pathname" + +module CommonSubscriptionAdapterTest + WAIT_WHEN_EXPECTING_EVENT = 3 + WAIT_WHEN_NOT_EXPECTING_EVENT = 0.2 + + def setup + server = ActionCable::Server::Base.new + server.config.cable = cable_config.with_indifferent_access + server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } + + adapter_klass = server.config.pubsub_adapter + + @rx_adapter = adapter_klass.new(server) + @tx_adapter = adapter_klass.new(server) + end + + def teardown + [@rx_adapter, @tx_adapter].uniq.compact.each(&:shutdown) + end + + def subscribe_as_queue(channel, adapter = @rx_adapter) + queue = Queue.new + + callback = -> data { queue << data } + subscribed = Concurrent::Event.new + adapter.subscribe(channel, callback, Proc.new { subscribed.set }) + subscribed.wait(WAIT_WHEN_EXPECTING_EVENT) + assert_predicate subscribed, :set? + + yield queue + + sleep WAIT_WHEN_NOT_EXPECTING_EVENT + assert_empty queue + ensure + adapter.unsubscribe(channel, callback) if subscribed.set? + end + + def test_subscribe_and_unsubscribe + subscribe_as_queue("channel") do |queue| + end + end + + def test_basic_broadcast + subscribe_as_queue("channel") do |queue| + @tx_adapter.broadcast("channel", "hello world") + + assert_equal "hello world", queue.pop + end + end + + def test_broadcast_after_unsubscribe + keep_queue = nil + subscribe_as_queue("channel") do |queue| + keep_queue = queue + + @tx_adapter.broadcast("channel", "hello world") + + assert_equal "hello world", queue.pop + end + + @tx_adapter.broadcast("channel", "hello void") + + sleep WAIT_WHEN_NOT_EXPECTING_EVENT + assert_empty keep_queue + end + + def test_multiple_broadcast + subscribe_as_queue("channel") do |queue| + @tx_adapter.broadcast("channel", "bananas") + @tx_adapter.broadcast("channel", "apples") + + received = [] + 2.times { received << queue.pop } + assert_equal ["apples", "bananas"], received.sort + end + end + + def test_identical_subscriptions + subscribe_as_queue("channel") do |queue| + subscribe_as_queue("channel") do |queue_2| + @tx_adapter.broadcast("channel", "hello") + + assert_equal "hello", queue_2.pop + end + + assert_equal "hello", queue.pop + end + end + + def test_simultaneous_subscriptions + subscribe_as_queue("channel") do |queue| + subscribe_as_queue("other channel") do |queue_2| + @tx_adapter.broadcast("channel", "apples") + @tx_adapter.broadcast("other channel", "oranges") + + assert_equal "apples", queue.pop + assert_equal "oranges", queue_2.pop + end + end + end + + def test_channel_filtered_broadcast + subscribe_as_queue("channel") do |queue| + @tx_adapter.broadcast("other channel", "one") + @tx_adapter.broadcast("channel", "two") + + assert_equal "two", queue.pop + end + end + + def test_long_identifiers + channel_1 = "a" * 100 + "1" + channel_2 = "a" * 100 + "2" + subscribe_as_queue(channel_1) do |queue| + subscribe_as_queue(channel_2) do |queue_2| + @tx_adapter.broadcast(channel_1, "apples") + @tx_adapter.broadcast(channel_2, "oranges") + + assert_equal "apples", queue.pop + assert_equal "oranges", queue_2.pop + end + end + end +end diff --git a/actioncable/test/subscription_adapter/inline_test.rb b/actioncable/test/subscription_adapter/inline_test.rb new file mode 100644 index 0000000000..6305626b2b --- /dev/null +++ b/actioncable/test/subscription_adapter/inline_test.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +require "test_helper" +require_relative "common" + +class InlineAdapterTest < ActionCable::TestCase + include CommonSubscriptionAdapterTest + + def setup + super + + @tx_adapter.shutdown + @tx_adapter = @rx_adapter + end + + def cable_config + { adapter: "inline" } + end +end diff --git a/actioncable/test/subscription_adapter/postgresql_test.rb b/actioncable/test/subscription_adapter/postgresql_test.rb new file mode 100644 index 0000000000..5fb26a8896 --- /dev/null +++ b/actioncable/test/subscription_adapter/postgresql_test.rb @@ -0,0 +1,65 @@ +# frozen_string_literal: true + +require "test_helper" +require_relative "common" + +require "active_record" + +class PostgresqlAdapterTest < ActionCable::TestCase + include CommonSubscriptionAdapterTest + + def setup + database_config = { "adapter" => "postgresql", "database" => "activerecord_unittest" } + ar_tests = File.expand_path("../../../activerecord/test", __dir__) + if Dir.exist?(ar_tests) + require File.join(ar_tests, "config") + require File.join(ar_tests, "support/config") + local_config = ARTest.config["connections"]["postgresql"]["arunit"] + database_config.update local_config if local_config + end + + ActiveRecord::Base.establish_connection database_config + + begin + ActiveRecord::Base.connection + rescue + @rx_adapter = @tx_adapter = nil + skip "Couldn't connect to PostgreSQL: #{database_config.inspect}" + end + + super + end + + def teardown + super + + ActiveRecord::Base.clear_all_connections! + end + + def cable_config + { adapter: "postgresql" } + end + + def test_clear_active_record_connections_adapter_still_works + server = ActionCable::Server::Base.new + server.config.cable = cable_config.with_indifferent_access + server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } + + adapter_klass = Class.new(server.config.pubsub_adapter) do + def active? + !@listener.nil? + end + end + + adapter = adapter_klass.new(server) + + subscribe_as_queue("channel", adapter) do |queue| + adapter.broadcast("channel", "hello world") + assert_equal "hello world", queue.pop + end + + ActiveRecord::Base.clear_reloadable_connections! + + assert adapter.active? + end +end diff --git a/actioncable/test/subscription_adapter/redis_test.rb b/actioncable/test/subscription_adapter/redis_test.rb new file mode 100644 index 0000000000..ac2d8ef724 --- /dev/null +++ b/actioncable/test/subscription_adapter/redis_test.rb @@ -0,0 +1,52 @@ +# frozen_string_literal: true + +require "test_helper" +require_relative "common" +require_relative "channel_prefix" + +require "action_cable/subscription_adapter/redis" + +class RedisAdapterTest < ActionCable::TestCase + include CommonSubscriptionAdapterTest + include ChannelPrefixTest + + def cable_config + { adapter: "redis", driver: "ruby" } + end +end + +class RedisAdapterTest::Hiredis < RedisAdapterTest + def cable_config + super.merge(driver: "hiredis") + end +end + +class RedisAdapterTest::AlternateConfiguration < RedisAdapterTest + def cable_config + alt_cable_config = super.dup + alt_cable_config.delete(:url) + alt_cable_config.merge(host: "127.0.0.1", port: 6379, db: 12) + end +end + +class RedisAdapterTest::Connector < ActionCable::TestCase + test "slices url, host, port, db, password and id from config" do + config = { url: 1, host: 2, port: 3, db: 4, password: 5, id: "Some custom ID" } + + assert_called_with ::Redis, :new, [ config ] do + connect config.merge(other: "unrelated", stuff: "here") + end + end + + test "adds default id if it is not specified" do + config = { url: 1, host: 2, port: 3, db: 4, password: 5, id: "ActionCable-PID-#{$$}" } + + assert_called_with ::Redis, :new, [ config ] do + connect config.except(:id) + end + end + + def connect(config) + ActionCable::SubscriptionAdapter::Redis.redis_connector.call(config) + end +end diff --git a/actioncable/test/subscription_adapter/subscriber_map_test.rb b/actioncable/test/subscription_adapter/subscriber_map_test.rb new file mode 100644 index 0000000000..ed81099cbc --- /dev/null +++ b/actioncable/test/subscription_adapter/subscriber_map_test.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +require "test_helper" + +class SubscriberMapTest < ActionCable::TestCase + test "broadcast should not change subscribers" do + setup_subscription_map + origin = @subscription_map.instance_variable_get(:@subscribers).dup + + @subscription_map.broadcast("not_exist_channel", "") + + assert_equal origin, @subscription_map.instance_variable_get(:@subscribers) + end + + private + def setup_subscription_map + @subscription_map = ActionCable::SubscriptionAdapter::SubscriberMap.new + end +end diff --git a/actioncable/test/subscription_adapter/test_adapter_test.rb b/actioncable/test/subscription_adapter/test_adapter_test.rb new file mode 100644 index 0000000000..3fe07adb4a --- /dev/null +++ b/actioncable/test/subscription_adapter/test_adapter_test.rb @@ -0,0 +1,47 @@ +# frozen_string_literal: true + +require "test_helper" +require_relative "common" + +class ActionCable::SubscriptionAdapter::TestTest < ActionCable::TestCase + include CommonSubscriptionAdapterTest + + def setup + super + + @tx_adapter.shutdown + @tx_adapter = @rx_adapter + end + + def cable_config + { adapter: "test" } + end + + test "#broadcast stores messages for streams" do + @tx_adapter.broadcast("channel", "payload") + @tx_adapter.broadcast("channel2", "payload2") + + assert_equal ["payload"], @tx_adapter.broadcasts("channel") + assert_equal ["payload2"], @tx_adapter.broadcasts("channel2") + end + + test "#clear_messages deletes recorded broadcasts for the channel" do + @tx_adapter.broadcast("channel", "payload") + @tx_adapter.broadcast("channel2", "payload2") + + @tx_adapter.clear_messages("channel") + + assert_equal [], @tx_adapter.broadcasts("channel") + assert_equal ["payload2"], @tx_adapter.broadcasts("channel2") + end + + test "#clear deletes all recorded broadcasts" do + @tx_adapter.broadcast("channel", "payload") + @tx_adapter.broadcast("channel2", "payload2") + + @tx_adapter.clear + + assert_equal [], @tx_adapter.broadcasts("channel") + assert_equal [], @tx_adapter.broadcasts("channel2") + end +end diff --git a/actioncable/test/test_helper.rb b/actioncable/test/test_helper.rb new file mode 100644 index 0000000000..c924f1e475 --- /dev/null +++ b/actioncable/test/test_helper.rb @@ -0,0 +1,43 @@ +# frozen_string_literal: true + +require "action_cable" +require "active_support/testing/autorun" +require "active_support/testing/method_call_assertions" + +require "puma" +require "rack/mock" + +begin + require "byebug" +rescue LoadError +end + +# Require all the stubs and models +Dir[File.expand_path("stubs/*.rb", __dir__)].each { |file| require file } + +# Set test adapter and logger +ActionCable.server.config.cable = { "adapter" => "test" } +ActionCable.server.config.logger = Logger.new(nil) + +class ActionCable::TestCase < ActiveSupport::TestCase + include ActiveSupport::Testing::MethodCallAssertions + + def wait_for_async + wait_for_executor Concurrent.global_io_executor + end + + def run_in_eventmachine + yield + wait_for_async + end + + def wait_for_executor(executor) + # do not wait forever, wait 2s + timeout = 2 + until executor.completed_task_count == executor.scheduled_task_count + sleep 0.1 + timeout -= 0.1 + raise "Executor could not complete all tasks in 2 seconds" unless timeout > 0 + end + end +end diff --git a/actioncable/test/test_helper_test.rb b/actioncable/test/test_helper_test.rb new file mode 100644 index 0000000000..90e3dbf01f --- /dev/null +++ b/actioncable/test/test_helper_test.rb @@ -0,0 +1,116 @@ +# frozen_string_literal: true + +require "test_helper" + +class BroadcastChannel < ActionCable::Channel::Base +end + +class TransmissionsTest < ActionCable::TestCase + def test_assert_broadcasts + assert_nothing_raised do + assert_broadcasts("test", 1) do + ActionCable.server.broadcast "test", "message" + end + end + end + + def test_assert_broadcasts_with_no_block + assert_nothing_raised do + ActionCable.server.broadcast "test", "message" + assert_broadcasts "test", 1 + end + + assert_nothing_raised do + ActionCable.server.broadcast "test", "message 2" + ActionCable.server.broadcast "test", "message 3" + assert_broadcasts "test", 3 + end + end + + def test_assert_no_broadcasts_with_no_block + assert_nothing_raised do + assert_no_broadcasts "test" + end + end + + def test_assert_no_broadcasts + assert_nothing_raised do + assert_no_broadcasts("test") do + ActionCable.server.broadcast "test2", "message" + end + end + end + + def test_assert_broadcasts_message_too_few_sent + ActionCable.server.broadcast "test", "hello" + error = assert_raises Minitest::Assertion do + assert_broadcasts("test", 2) do + ActionCable.server.broadcast "test", "world" + end + end + + assert_match(/2 .* but 1/, error.message) + end + + def test_assert_broadcasts_message_too_many_sent + error = assert_raises Minitest::Assertion do + assert_broadcasts("test", 1) do + ActionCable.server.broadcast "test", "hello" + ActionCable.server.broadcast "test", "world" + end + end + + assert_match(/1 .* but 2/, error.message) + end + + def test_assert_no_broadcasts_failure + error = assert_raises Minitest::Assertion do + assert_no_broadcasts "test" do + ActionCable.server.broadcast "test", "hello" + end + end + + assert_match(/0 .* but 1/, error.message) + end +end + +class TransmitedDataTest < ActionCable::TestCase + include ActionCable::TestHelper + + def test_assert_broadcast_on + assert_nothing_raised do + assert_broadcast_on("test", "message") do + ActionCable.server.broadcast "test", "message" + end + end + end + + def test_assert_broadcast_on_with_hash + assert_nothing_raised do + assert_broadcast_on("test", text: "hello") do + ActionCable.server.broadcast "test", text: "hello" + end + end + end + + def test_assert_broadcast_on_with_no_block + assert_nothing_raised do + ActionCable.server.broadcast "test", "hello" + assert_broadcast_on "test", "hello" + end + + assert_nothing_raised do + ActionCable.server.broadcast "test", "world" + assert_broadcast_on "test", "world" + end + end + + def test_assert_broadcast_on_message + ActionCable.server.broadcast "test", "hello" + error = assert_raises Minitest::Assertion do + assert_broadcast_on("test", "world") + end + + assert_match(/No messages sent/, error.message) + end +end diff --git a/actioncable/test/worker_test.rb b/actioncable/test/worker_test.rb new file mode 100644 index 0000000000..f7dc428441 --- /dev/null +++ b/actioncable/test/worker_test.rb @@ -0,0 +1,46 @@ +# frozen_string_literal: true + +require "test_helper" + +class WorkerTest < ActionCable::TestCase + class Receiver + attr_accessor :last_action + + def run + @last_action = :run + end + + def process(message) + @last_action = [ :process, message ] + end + + def connection + self + end + + def 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 + + setup do + @worker = ActionCable::Server::Worker.new + @receiver = Receiver.new + end + + teardown do + @receiver.last_action = nil + end + + test "invoke" do + @worker.invoke @receiver, :run, connection: @receiver.connection + assert_equal :run, @receiver.last_action + end + + test "invoke with arguments" do + @worker.invoke @receiver, :process, "Hello", connection: @receiver.connection + assert_equal [ :process, "Hello" ], @receiver.last_action + end +end |