From b168eb5819fa5fea940c9865d5c9a3ec5ba2a7ec Mon Sep 17 00:00:00 2001 From: Jeremy Daer Date: Fri, 11 Mar 2016 16:32:02 -0700 Subject: Cable message encoding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Introduce a connection coder responsible for encoding Cable messages as WebSocket messages, defaulting to `ActiveSupport::JSON` and duck- typing to any object responding to `#encode` and `#decode`. * Consolidate encoding responsibility to the connection. No longer explicitly JSON-encode from channels or other sources. Pass Cable messages as Hashes to `#transmit` and rely on it to encode. * Introduce stream encoders responsible for decoding pubsub messages. Preserve the currently raw encoding, but make it easy to use JSON. Same duck type as the connection encoder. * Revert recent data normalization/quoting (#23649) which treated `identifier` and `data` values as nested JSON objects rather than as opaque JSON-encoded strings. That dealt us an awkward hand where we'd decode JSON strings… or not, but always encode as JSON. Embedding JSON object values directly is preferably, no extra JSON encoding, but that should be a purposeful protocol version change rather than ambiguously, inadvertently supporting multiple message formats. --- actioncable/test/channel/base_test.rb | 4 +- actioncable/test/channel/rejection_test.rb | 2 +- actioncable/test/channel/stream_test.rb | 160 ++++++++++++++++------ actioncable/test/client_test.rb | 36 ++--- actioncable/test/connection/identifier_test.rb | 6 +- actioncable/test/connection/subscriptions_test.rb | 2 +- actioncable/test/stubs/test_connection.rb | 23 ++-- actioncable/test/stubs/test_server.rb | 12 +- actioncable/test/subscription_adapter/common.rb | 3 +- actioncable/test/test_helper.rb | 8 +- 10 files changed, 168 insertions(+), 88 deletions(-) (limited to 'actioncable/test') diff --git a/actioncable/test/channel/base_test.rb b/actioncable/test/channel/base_test.rb index bed54eb6b3..daa782eeb3 100644 --- a/actioncable/test/channel/base_test.rb +++ b/actioncable/test/channel/base_test.rb @@ -146,12 +146,12 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase test "transmitting data" do @channel.perform_action 'action' => :get_latest - expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "message" => { "data" => "latest" } + expected = { "identifier" => "{id: 1}", "message" => { "data" => "latest" }} assert_equal expected, @connection.last_transmission end test "subscription confirmation" do - expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription" + expected = { "identifier" => "{id: 1}", "type" => "confirm_subscription" } assert_equal expected, @connection.last_transmission end diff --git a/actioncable/test/channel/rejection_test.rb b/actioncable/test/channel/rejection_test.rb index aa93396d44..15db57d6ba 100644 --- a/actioncable/test/channel/rejection_test.rb +++ b/actioncable/test/channel/rejection_test.rb @@ -18,7 +18,7 @@ class ActionCable::Channel::RejectionTest < ActiveSupport::TestCase @connection.expects(:subscriptions).returns mock().tap { |m| m.expects(:remove_subscription).with instance_of(SecretChannel) } @channel = SecretChannel.new @connection, "{id: 1}", { id: 1 } - expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "reject_subscription" + expected = { "identifier" => "{id: 1}", "type" => "reject_subscription" } assert_equal expected, @connection.last_transmission end diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb index 526ea92e4f..f51f19eb7d 100644 --- a/actioncable/test/channel/stream_test.rb +++ b/actioncable/test/channel/stream_test.rb @@ -2,18 +2,43 @@ require 'test_helper' require 'stubs/test_connection' require 'stubs/room' -class ActionCable::Channel::StreamTest < ActionCable::TestCase +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}" + 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 @@ -22,69 +47,114 @@ class ActionCable::Channel::StreamTest < ActionCable::TestCase end end - test "streaming start and stop" do - run_in_eventmachine do - connection = TestConnection.new - connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } - channel = ChatChannel.new connection, "{id: 1}", { id: 1 } + class StreamTest < ActionCable::TestCase + test "streaming start and stop" do + run_in_eventmachine do + connection = TestConnection.new + connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } + channel = ChatChannel.new connection, "{id: 1}", { id: 1 } - connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) } - channel.unsubscribe_from_channel + connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) } + channel.unsubscribe_from_channel + end end - end - test "stream from non-string channel" do - run_in_eventmachine do - connection = TestConnection.new - connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("channel", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } - channel = SymbolChannel.new connection, "" + test "stream from non-string channel" do + run_in_eventmachine do + connection = TestConnection.new + connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("channel", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } + channel = SymbolChannel.new connection, "" - connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) } - channel.unsubscribe_from_channel + connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) } + channel.unsubscribe_from_channel + end end - end - test "stream_for" do - run_in_eventmachine do - connection = TestConnection.new - connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } + test "stream_for" do + run_in_eventmachine do + connection = TestConnection.new + connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:stream_tests:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } - channel = ChatChannel.new connection, "" - channel.stream_for Room.new(1) + channel = ChatChannel.new connection, "" + channel.stream_for Room.new(1) + end end - end - test "stream_from subscription confirmation" do - run_in_eventmachine do - connection = TestConnection.new + test "stream_from subscription confirmation" do + run_in_eventmachine do + connection = TestConnection.new - ChatChannel.new connection, "{id: 1}", { id: 1 } - assert_nil connection.last_transmission + ChatChannel.new connection, "{id: 1}", { id: 1 } + assert_nil connection.last_transmission - wait_for_async + wait_for_async - expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription" - connection.transmit(expected) + confirmation = { "identifier" => "{id: 1}", "type" => "confirm_subscription" } + connection.transmit(confirmation) - assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation within 0.1s" + assert_equal confirmation, connection.last_transmission, "Did not receive subscription confirmation within 0.1s" + end end - end - test "subscription confirmation should only be sent out once" do - run_in_eventmachine do - connection = TestConnection.new + 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 + channel = ChatChannel.new connection, "test_channel" + channel.send_confirmation + channel.send_confirmation - wait_for_async + wait_for_async - expected = ActiveSupport::JSON.encode "identifier" => "test_channel", "type" => "confirm_subscription" - assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation" + expected = { "identifier" => "test_channel", "type" => "confirm_subscription" } + assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation" - assert_equal 1, connection.transmissions.size + assert_equal 1, connection.transmissions.size + end end end + require 'action_cable/subscription_adapter/inline' + + class StreamEncodingTest < ActionCable::TestCase + setup do + @server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Inline) + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + @server.stubs(:channel_classes).returns(ChatChannel.name => ChatChannel) + end + + test 'custom encoder' do + run_in_eventmachine do + connection = open_connection + subscribe_to connection, identifiers: { id: 1 } + + connection.websocket.expects(:transmit) + @server.broadcast 'test_room_1', { foo: 'bar' }, coder: DummyEncoder + wait_for_async + 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 connection.websocket.possible? + + wait_for_async + assert connection.websocket.alive? + end + end + + def receive(connection, command:, identifiers:) + identifier = JSON.generate(channel: 'ActionCable::StreamTests::ChatChannel', **identifiers) + connection.dispatch_websocket_message JSON.generate(command: command, identifier: identifier) + wait_for_async + end + end end diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb index 5f5c09d1a1..5ac453db35 100644 --- a/actioncable/test/client_test.rb +++ b/actioncable/test/client_test.rb @@ -75,11 +75,11 @@ class ClientTest < ActionCable::TestCase end @ws.on(:message) do |event| - hash = JSON.parse(event.data) - if hash['type'] == 'ping' + message = JSON.parse(event.data) + if message['type'] == 'ping' @pings += 1 else - @messages << hash + @messages << message @has_messages.release end end @@ -116,8 +116,8 @@ class ClientTest < ActionCable::TestCase list end - def send_message(hash) - @ws.send(JSON.dump(hash)) + def send_message(message) + @ws.send(JSON.generate(message)) end def close @@ -148,9 +148,9 @@ class ClientTest < ActionCable::TestCase with_puma_server do |port| c = faye_client(port) assert_equal({"type" => "welcome"}, c.read_message) # pop the first welcome message off the stack - c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') + c.send_message command: 'subscribe', identifier: JSON.generate(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') + c.send_message command: 'message', identifier: JSON.generate(channel: 'EchoChannel'), data: JSON.generate(action: 'ding', message: 'hello') assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "message"=>{"dong"=>"hello"}}, c.read_message) c.close end @@ -165,12 +165,12 @@ class ClientTest < ActionCable::TestCase clients.map {|c| Concurrent::Future.execute { assert_equal({"type" => "welcome"}, c.read_message) # pop the first welcome message off the stack - c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') + c.send_message command: 'subscribe', identifier: JSON.generate(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') + c.send_message command: 'message', identifier: JSON.generate(channel: 'EchoChannel'), data: JSON.generate(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') + c.send_message command: 'message', identifier: JSON.generate(channel: '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 } }.each(&:wait!) @@ -185,9 +185,9 @@ class ClientTest < ActionCable::TestCase clients.map {|c| Concurrent::Future.execute { assert_equal({"type" => "welcome"}, c.read_message) # pop the first welcome message off the stack - c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') + c.send_message command: 'subscribe', identifier: JSON.generate(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') + c.send_message command: 'message', identifier: JSON.generate(channel: 'EchoChannel'), data: JSON.generate(action: 'ding', message: 'hello') assert_equal({"identifier"=>'{"channel":"EchoChannel"}', "message"=>{"dong"=>"hello"}}, c.read_message) } }.each(&:wait!) @@ -199,16 +199,16 @@ class ClientTest < ActionCable::TestCase with_puma_server do |port| c = faye_client(port) assert_equal({"type" => "welcome"}, c.read_message) # pop the first welcome message off the stack - c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') + c.send_message command: 'subscribe', identifier: JSON.generate(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: 'delay', message: 'hello') + c.send_message command: 'message', identifier: JSON.generate(channel: 'EchoChannel'), data: JSON.generate(action: 'delay', message: 'hello') c.close # disappear before write c = faye_client(port) assert_equal({"type" => "welcome"}, c.read_message) # pop the first welcome message off the stack - c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') + c.send_message command: 'subscribe', identifier: JSON.generate(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') + c.send_message command: 'message', identifier: JSON.generate(channel: 'EchoChannel'), data: JSON.generate(action: 'ding', message: 'hello') assert_equal({"identifier"=>'{"channel":"EchoChannel"}', "message"=>{"dong"=>"hello"}}, c.read_message) c.close # disappear before read end @@ -217,7 +217,7 @@ class ClientTest < ActionCable::TestCase def test_unsubscribe_client with_puma_server do |port| app = ActionCable.server - identifier = JSON.dump(channel: 'EchoChannel') + identifier = JSON.generate(channel: 'EchoChannel') c = faye_client(port) assert_equal({"type" => "welcome"}, c.read_message) @@ -240,7 +240,7 @@ class ClientTest < ActionCable::TestCase with_puma_server do |port| c = faye_client(port) assert_equal({"type" => "welcome"}, c.read_message) - c.send_message command: 'subscribe', identifier: JSON.dump(channel: 'EchoChannel') + c.send_message command: 'subscribe', identifier: JSON.generate(channel: 'EchoChannel') assert_equal({"identifier"=>"{\"channel\":\"EchoChannel\"}", "type"=>"confirm_subscription"}, c.read_message) ActionCable.server.restart diff --git a/actioncable/test/connection/identifier_test.rb b/actioncable/test/connection/identifier_test.rb index c3d5f1f90b..b48d9af809 100644 --- a/actioncable/test/connection/identifier_test.rb +++ b/actioncable/test/connection/identifier_test.rb @@ -40,8 +40,7 @@ class ActionCable::Connection::IdentifierTest < ActionCable::TestCase open_connection_with_stubbed_pubsub @connection.websocket.expects(:close) - message = ActiveSupport::JSON.encode('type' => 'disconnect') - @connection.process_internal_message message + @connection.process_internal_message 'type' => 'disconnect' end end @@ -50,8 +49,7 @@ class ActionCable::Connection::IdentifierTest < ActionCable::TestCase open_connection_with_stubbed_pubsub @connection.websocket.expects(:close).never - message = ActiveSupport::JSON.encode('type' => 'unknown') - @connection.process_internal_message message + @connection.process_internal_message 'type' => 'unknown' end end diff --git a/actioncable/test/connection/subscriptions_test.rb b/actioncable/test/connection/subscriptions_test.rb index f91597f567..53e8547245 100644 --- a/actioncable/test/connection/subscriptions_test.rb +++ b/actioncable/test/connection/subscriptions_test.rb @@ -88,7 +88,7 @@ class ActionCable::Connection::SubscriptionsTest < ActionCable::TestCase channel1 = subscribe_to_chat_channel - channel2_id = ActiveSupport::JSON.encode({ id: 2, channel: 'ActionCable::Connection::SubscriptionsTest::ChatChannel' }) + channel2_id = ActiveSupport::JSON.encode(id: 2, channel: 'ActionCable::Connection::SubscriptionsTest::ChatChannel') channel2 = subscribe_to_chat_channel(channel2_id) channel1.expects(:unsubscribe_from_channel) diff --git a/actioncable/test/stubs/test_connection.rb b/actioncable/test/stubs/test_connection.rb index 8ba284fdc6..885450dda6 100644 --- a/actioncable/test/stubs/test_connection.rb +++ b/actioncable/test/stubs/test_connection.rb @@ -3,24 +3,31 @@ require 'stubs/user' class TestConnection attr_reader :identifiers, :logger, :current_user, :server, :transmissions - def initialize(user = User.new("lifo")) + 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 + @server = TestServer.new(subscription_adapter: subscription_adapter) @transmissions = [] end - def pubsub - SuccessAdapter.new(server) + def transmit(cable_message) + @transmissions << encode(cable_message) end - def transmit(data) - @transmissions << data + def last_transmission + decode @transmissions.last if @transmissions.any? end - def last_transmission - @transmissions.last + 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 index 9e860825f3..b86f422a13 100644 --- a/actioncable/test/stubs/test_server.rb +++ b/actioncable/test/stubs/test_server.rb @@ -2,22 +2,26 @@ require 'ostruct' class TestServer include ActionCable::Server::Connections + include ActionCable::Server::Broadcasting - attr_reader :logger, :config + attr_reader :logger, :config, :mutex - def initialize + def initialize(subscription_adapter: SuccessAdapter) @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new) - @config = OpenStruct.new(log_tags: [], subscription_adapter: SuccessAdapter) + + @config = OpenStruct.new(log_tags: [], subscription_adapter: subscription_adapter) @config.use_faye = ENV['FAYE'].present? @config.client_socket_class = if @config.use_faye ActionCable::Connection::FayeClientSocket else ActionCable::Connection::ClientSocket end + + @mutex = Monitor.new end def pubsub - @config.subscription_adapter.new(self) + @pubsub ||= @config.subscription_adapter.new(self) end def event_loop diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb index 82f0abbbf3..285c690df0 100644 --- a/actioncable/test/subscription_adapter/common.rb +++ b/actioncable/test/subscription_adapter/common.rb @@ -20,8 +20,7 @@ module CommonSubscriptionAdapterTest end def teardown - @tx_adapter.shutdown if @tx_adapter && @tx_adapter != @rx_adapter - @rx_adapter.shutdown if @rx_adapter + [@rx_adapter, @tx_adapter].uniq.each(&:shutdown) end diff --git a/actioncable/test/test_helper.rb b/actioncable/test/test_helper.rb index 030362d512..de1ee96770 100644 --- a/actioncable/test/test_helper.rb +++ b/actioncable/test/test_helper.rb @@ -2,11 +2,13 @@ require 'action_cable' require 'active_support/testing/autorun' require 'puma' - require 'mocha/setup' - require 'rack/mock' -require 'active_support/core_ext/hash/indifferent_access' + +begin + require 'byebug' +rescue LoadError +end # Require all the stubs and models Dir[File.dirname(__FILE__) + '/stubs/*.rb'].each {|file| require file } -- cgit v1.2.3