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.

3 files changed, 118 insertions, 48 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