3 files changed, 193 insertions, 48 deletions

diff --git a/actioncable/test/channel/base_test.rb b/actioncable/test/channel/base_test.rb
index d41bf3064b..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
 
@@ -166,6 +166,81 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
     end
   end
 
+  test "notification for perform_action" do
+    begin
+      events = []
+      ActiveSupport::Notifications.subscribe "perform_action.action_cable" do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      data = {'action' => :speak, 'content' => 'hello'}
+      @channel.perform_action data
+
+      assert_equal 1, events.length
+      assert_equal 'perform_action.action_cable', events[0].name
+      assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class]
+      assert_equal :speak, events[0].payload[:action]
+      assert_equal data, events[0].payload[:data]
+    ensure
+      ActiveSupport::Notifications.unsubscribe "perform_action.action_cable"
+    end
+  end
+
+  test "notification for transmit" do
+    begin
+      events = []
+      ActiveSupport::Notifications.subscribe 'transmit.action_cable' do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      @channel.perform_action 'action' => :get_latest
+      expected_data = {data: 'latest'}
+
+      assert_equal 1, events.length
+      assert_equal 'transmit.action_cable', events[0].name
+      assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class]
+      assert_equal expected_data, events[0].payload[:data]
+      assert_nil events[0].payload[:via]
+    ensure
+      ActiveSupport::Notifications.unsubscribe 'transmit.action_cable'
+    end
+  end
+
+  test "notification for transmit_subscription_confirmation" do
+    begin
+      events = []
+      ActiveSupport::Notifications.subscribe 'transmit_subscription_confirmation.action_cable' do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      @channel.stubs(:subscription_confirmation_sent?).returns(false)
+      @channel.send(:transmit_subscription_confirmation)
+
+      assert_equal 1, events.length
+      assert_equal 'transmit_subscription_confirmation.action_cable', events[0].name
+      assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class]
+    ensure
+      ActiveSupport::Notifications.unsubscribe 'transmit_subscription_confirmation.action_cable'
+    end
+  end
+
+  test "notification for transmit_subscription_rejection" do
+    begin
+      events = []
+      ActiveSupport::Notifications.subscribe 'transmit_subscription_rejection.action_cable' do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      @channel.send(:transmit_subscription_rejection)
+
+      assert_equal 1, events.length
+      assert_equal 'transmit_subscription_rejection.action_cable', events[0].name
+      assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class]
+    ensure
+      ActiveSupport::Notifications.unsubscribe 'transmit_subscription_rejection.action_cable'
+    end
+  end
+
   private
     def assert_logged(message)
       old_logger = @connection.logger
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