315 lines
10 KiB
Ruby
315 lines
10 KiB
Ruby
# 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.match?(/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 |_|
|
|
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
|