aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorMatthew Draper <matthew@trebex.net>2016-01-25 03:55:53 +1030
committerMatthew Draper <matthew@trebex.net>2016-01-30 03:46:37 +1030
commitce37de4a19447fc89d2d16f15ba9314fba30d47e (patch)
treec1e562251ea6dccb2b9be1bd8ecf64294600146f
parent16a6603956551703e3bbd06101c568a73bcdaa52 (diff)
downloadrails-ce37de4a19447fc89d2d16f15ba9314fba30d47e.tar.gz
rails-ce37de4a19447fc89d2d16f15ba9314fba30d47e.tar.bz2
rails-ce37de4a19447fc89d2d16f15ba9314fba30d47e.zip
Add a couple of tests that connect with a WS client
-rw-r--r--Gemfile2
-rw-r--r--Gemfile.lock4
-rw-r--r--actioncable/test/client/echo_channel.rb13
-rw-r--r--actioncable/test/client_test.rb238
4 files changed, 257 insertions, 0 deletions
diff --git a/Gemfile b/Gemfile
index 9dd841d535..f8df08ee43 100644
--- a/Gemfile
+++ b/Gemfile
@@ -66,6 +66,8 @@ group :cable do
gem 'em-hiredis', require: false
gem 'redis', require: false
+
+ gem 'faye-websocket', require: false
end
# Add your own local bundler stuff.
diff --git a/Gemfile.lock b/Gemfile.lock
index ddf5cc88a9..d2d616b39d 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -127,6 +127,9 @@ GEM
erubis (2.7.0)
eventmachine (1.0.9.1)
execjs (2.6.0)
+ faye-websocket (0.10.2)
+ eventmachine (>= 0.12.0)
+ websocket-driver (>= 0.5.1)
ffi (1.9.10)
ffi (1.9.10-x64-mingw32)
ffi (1.9.10-x86-mingw32)
@@ -282,6 +285,7 @@ DEPENDENCIES
delayed_job
delayed_job_active_record
em-hiredis
+ faye-websocket
jquery-rails
json
kindlerb (= 0.1.1)
diff --git a/actioncable/test/client/echo_channel.rb b/actioncable/test/client/echo_channel.rb
new file mode 100644
index 0000000000..9a54080d4d
--- /dev/null
+++ b/actioncable/test/client/echo_channel.rb
@@ -0,0 +1,13 @@
+class EchoChannel < ActionCable::Channel::Base
+ def subscribed
+ stream_from "global"
+ end
+
+ def ding(data)
+ transmit(dong: data['message'])
+ end
+
+ def bulk(data)
+ ActionCable.server.broadcast "global", wide: data['message']
+ end
+end
diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb
new file mode 100644
index 0000000000..b185654c71
--- /dev/null
+++ b/actioncable/test/client_test.rb
@@ -0,0 +1,238 @@
+require 'test_helper'
+require 'concurrent'
+
+require 'active_support/core_ext/hash/indifferent_access'
+require 'pathname'
+
+require 'faye/websocket'
+require 'json'
+
+class ClientTest < ActionCable::TestCase
+ WAIT_WHEN_EXPECTING_EVENT = 3
+ WAIT_WHEN_NOT_EXPECTING_EVENT = 0.2
+
+ def setup
+ # TODO: ActionCable requires a *lot* of setup at the moment...
+ ::Object.const_set(:ApplicationCable, Module.new)
+ ::ApplicationCable.const_set(:Connection, Class.new(ActionCable::Connection::Base))
+
+ ::Object.const_set(:Rails, Module.new)
+ ::Rails.singleton_class.send(:define_method, :root) { Pathname.new(__dir__) }
+
+ ActionCable.instance_variable_set(:@server, nil)
+ server = ActionCable.server
+ server.config = ActionCable::Server::Configuration.new
+ inner_logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN }
+ server.config.logger = ActionCable::Connection::TaggedLoggerProxy.new(inner_logger, tags: [])
+
+ server.config.cable = { adapter: 'async' }.with_indifferent_access
+
+ # and now the "real" setup for our test:
+ server.config.disable_request_forgery_protection = true
+ server.config.channel_load_paths = [File.expand_path('client', __dir__)]
+
+ @reactor_mutex = Mutex.new
+ @reactor_users = 0
+ @reactor_running = Concurrent::Event.new
+
+ # faye-websocket is warning-rich
+ @previous_verbose, $VERBOSE = $VERBOSE, nil
+ end
+
+ def teardown
+ $VERBOSE = @previous_verbose
+
+ if @reactor_running.set?
+ EM.stop
+ end
+
+ begin
+ ::Object.send(:remove_const, :ApplicationCable)
+ rescue NameError
+ end
+ begin
+ ::Object.send(:remove_const, :Rails)
+ rescue NameError
+ end
+ end
+
+ def with_puma_server(rack_app = ActionCable.server, port = 3099)
+ 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
+
+ t = Thread.new { server.run.join }
+ yield port
+
+ ensure
+ server.stop(true)
+ t.join
+ end
+
+ def start_event_machine
+ @reactor_mutex.synchronize do
+ unless @reactor_running.set?
+ @reactor ||= Thread.new do
+ EM.next_tick do
+ @reactor_running.set
+ end
+ EM.run
+ end
+ @reactor_running.wait(WAIT_WHEN_EXPECTING_EVENT)
+ end
+ @reactor_users += 1
+ end
+ end
+
+ def stop_event_machine
+ @reactor_mutex.synchronize do
+ @reactor_users -= 1
+ if @reactor_users < 1
+ @reactor_running.reset
+ EM.stop
+ @reactor = nil
+ end
+ end
+ end
+
+ class SyncClient
+ attr_reader :pings
+
+ def initialize(em_controller, port)
+ em_controller.start_event_machine
+
+ @ws = Faye::WebSocket::Client.new("ws://127.0.0.1:#{port}/")
+ @messages = Queue.new
+ @closed = Concurrent::Event.new
+ @has_messages = Concurrent::Event.new
+ @pings = 0
+
+ open = Concurrent::Event.new
+ error = nil
+
+ @ws.on(:error) do |event|
+ if open.set?
+ @messages << RuntimeError.new(event.message)
+ else
+ error = event.message
+ open.set
+ end
+ end
+
+ @ws.on(:open) do |event|
+ open.set
+ end
+
+ @ws.on(:message) do |event|
+ hash = JSON.parse(event.data)
+ if hash['identifier'] == '_ping'
+ @pings += 1
+ else
+ @messages << hash
+ @has_messages.set
+ end
+ end
+
+ @ws.on(:close) do |event|
+ @ws = nil
+ em_controller.stop_event_machine
+ @closed.set
+ end
+
+ open.wait(WAIT_WHEN_EXPECTING_EVENT)
+ raise error if error
+ end
+
+ def read_message
+ @has_messages.wait(WAIT_WHEN_EXPECTING_EVENT) if @messages.empty?
+ @has_messages.reset if @messages.size < 2
+
+ msg = @messages.pop(true)
+ raise msg if msg.is_a?(Exception)
+
+ msg
+ end
+
+ def read_messages
+ list = []
+ loop do
+ @has_messages.wait(WAIT_WHEN_NOT_EXPECTING_EVENT)
+ if @has_messages.set?
+ list << read_message
+ else
+ break
+ end
+ end
+ list
+ end
+
+ def send_message(hash)
+ @ws.send(JSON.dump(hash))
+ end
+
+ def close
+ sleep WAIT_WHEN_NOT_EXPECTING_EVENT
+
+ unless @messages.empty?
+ raise "#{@messages.size} messages unprocessed"
+ end
+
+ @ws.close
+ @closed.wait(WAIT_WHEN_EXPECTING_EVENT)
+ end
+ end
+
+ def faye_client(port)
+ SyncClient.new(self, port)
+ end
+
+ def test_single_client
+ with_puma_server do |port|
+ c = faye_client(port)
+ c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel')
+ assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "type"=>"confirm_subscription"}, c.read_message)
+ c.send_message command: 'message', identifier: JSON.dump(channel: 'EchoChannel'), data: JSON.dump(action: 'ding', message: 'hello')
+ assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "message"=>{"dong"=>"hello"}}, c.read_message)
+ c.close
+ end
+ end
+
+ def test_interacting_clients
+ with_puma_server do |port|
+ clients = 20.times.map { faye_client(port) }
+
+ barrier_1 = Concurrent::CyclicBarrier.new(clients.size)
+ barrier_2 = Concurrent::CyclicBarrier.new(clients.size)
+
+ clients.map {|c| Concurrent::Future.execute {
+ c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel')
+ assert_equal({"identifier"=>'{"channel":"EchoChannel"}', "type"=>"confirm_subscription"}, c.read_message)
+ c.send_message command: 'message', identifier: JSON.dump(channel: 'EchoChannel'), data: JSON.dump(action: 'ding', message: 'hello')
+ assert_equal({"identifier"=>'{"channel":"EchoChannel"}', "message"=>{"dong"=>"hello"}}, c.read_message)
+ barrier_1.wait WAIT_WHEN_EXPECTING_EVENT
+ c.send_message command: 'message', identifier: JSON.dump(channel: 'EchoChannel'), data: JSON.dump(action: 'bulk', message: 'hello')
+ barrier_2.wait WAIT_WHEN_EXPECTING_EVENT
+ sleep 1
+ assert_equal clients.size, c.read_messages.size
+ } }.each(&:wait!)
+
+ clients.map {|c| Concurrent::Future.execute { c.close } }.each(&:wait!)
+ end
+ end
+
+ def test_many_clients
+ with_puma_server do |port|
+ clients = 200.times.map { faye_client(port) }
+
+ clients.map {|c| Concurrent::Future.execute {
+ c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel')
+ assert_equal({"identifier"=>'{"channel":"EchoChannel"}', "type"=>"confirm_subscription"}, c.read_message)
+ c.send_message command: 'message', identifier: JSON.dump(channel: 'EchoChannel'), data: JSON.dump(action: 'ding', message: 'hello')
+ assert_equal({"identifier"=>'{"channel":"EchoChannel"}', "message"=>{"dong"=>"hello"}}, c.read_message)
+ } }.each(&:wait!)
+
+ clients.map {|c| Concurrent::Future.execute { c.close } }.each(&:wait!)
+ end
+ end
+end