Cable message encoding

* 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.

10 files changed, 168 insertions, 88 deletions

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 }