1 files changed, 106 insertions, 36 deletions
diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb
index f51f19eb7d..9ad2213d47 100644
--- a/actioncable/test/channel/stream_test.rb
+++ b/actioncable/test/channel/stream_test.rb
@@ -1,6 +1,9 @@
-require 'test_helper'
-require 'stubs/test_connection'
-require 'stubs/room'
+# frozen_string_literal: true
+
+require "test_helper"
+require "minitest/mock"
+require "stubs/test_connection"
+require "stubs/room"
 
 module ActionCable::StreamTests
   class Connection < ActionCable::Connection::Base
@@ -23,22 +26,23 @@ module ActionCable::StreamTests
       transmit_subscription_confirmation
     end
 
-    private def pick_coder(coder)
-      case coder
-      when nil, 'json'
-        ActiveSupport::JSON
-      when 'custom'
-        DummyEncoder
-      when 'none'
-        nil
+    private
+      def pick_coder(coder)
+        case coder
+        when nil, "json"
+          ActiveSupport::JSON
+        when "custom"
+          DummyEncoder
+        when "none"
+          nil
+        end
       end
-    end
   end
 
   module DummyEncoder
     extend self
     def encode(*) '{ "foo": "encoded" }' end
-    def decode(*) { foo: 'decoded' } end
+    def decode(*) { foo: "decoded" } end
   end
 
   class SymbolChannel < ActionCable::Channel::Base
@@ -51,32 +55,58 @@ module ActionCable::StreamTests
     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 }
+        pubsub = Minitest::Mock.new connection.pubsub
+
+        pubsub.expect(:subscribe, nil, ["test_room_1", Proc, Proc])
+        pubsub.expect(:unsubscribe, nil, ["test_room_1", Proc])
+
+        connection.stub(:pubsub, pubsub) do
+          channel = ChatChannel.new connection, "{id: 1}", id: 1
+          channel.subscribe_to_channel
 
-        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) }
-        channel.unsubscribe_from_channel
+          wait_for_async
+          channel.unsubscribe_from_channel
+        end
+
+        assert pubsub.verify
       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, ""
+        pubsub = Minitest::Mock.new connection.pubsub
+
+        pubsub.expect(:subscribe, nil, ["channel", Proc, Proc])
+        pubsub.expect(:unsubscribe, nil, ["channel", Proc])
+
+        connection.stub(:pubsub, pubsub) do
+          channel = SymbolChannel.new connection, ""
+          channel.subscribe_to_channel
 
-        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) }
-        channel.unsubscribe_from_channel
+          wait_for_async
+
+          channel.unsubscribe_from_channel
+        end
+
+        assert pubsub.verify
       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:stream_tests:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
 
         channel = ChatChannel.new connection, ""
+        channel.subscribe_to_channel
         channel.stream_for Room.new(1)
+        wait_for_async
+
+        pubsub_call = channel.pubsub.class.class_variable_get "@@subscribe_called"
+
+        assert_equal "action_cable:stream_tests:chat:Room#1-Campfire", pubsub_call[:channel]
+        assert_instance_of Proc, pubsub_call[:callback]
+        assert_instance_of Proc, pubsub_call[:success_callback]
       end
     end
 
@@ -84,7 +114,9 @@ module ActionCable::StreamTests
       run_in_eventmachine do
         connection = TestConnection.new
 
-        ChatChannel.new connection, "{id: 1}", { id: 1 }
+        channel = ChatChannel.new connection, "{id: 1}", id: 1
+        channel.subscribe_to_channel
+
         assert_nil connection.last_transmission
 
         wait_for_async
@@ -114,45 +146,83 @@ module ActionCable::StreamTests
     end
   end
 
-  require 'action_cable/subscription_adapter/inline'
+  require "action_cable/subscription_adapter/async"
 
-  class StreamEncodingTest < ActionCable::TestCase
+  class UserCallbackChannel < ActionCable::Channel::Base
+    def subscribed
+      stream_from :channel do
+        Thread.current[:ran_callback] = true
+      end
+    end
+  end
+
+  class MultiChatChannel < ActionCable::Channel::Base
+    def subscribed
+      stream_from "main_room"
+      stream_from "test_all_rooms"
+    end
+  end
+
+  class StreamFromTest < ActionCable::TestCase
     setup do
-      @server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Inline)
+      @server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Async)
       @server.config.allowed_request_origins = %w( http://rubyonrails.com )
-      @server.stubs(:channel_classes).returns(ChatChannel.name => ChatChannel)
     end
 
-    test 'custom encoder' do
+    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
+        assert_called(connection.websocket, :transmit) do
+          @server.broadcast "test_room_1", { foo: "bar" }, { coder: DummyEncoder }
+          wait_for_async
+          wait_for_executor connection.server.worker_pool.executor
+        end
+      end
+    end
+
+    test "user supplied callbacks are run through the worker pool" do
+      run_in_eventmachine do
+        connection = open_connection
+        receive(connection, command: "subscribe", channel: UserCallbackChannel.name, identifiers: { id: 1 })
+
+        @server.broadcast "channel", {}
         wait_for_async
+        assert_not Thread.current[:ran_callback], "User callback was not run through the worker pool"
+      end
+    end
+
+    test "subscription confirmation should only be sent out once with multiple stream_from" do
+      run_in_eventmachine do
+        connection = open_connection
+        expected = { "identifier" => { "channel" => MultiChatChannel.name }.to_json, "type" => "confirm_subscription" }
+        assert_called_with(connection.websocket, :transmit, [expected.to_json]) do
+          receive(connection, command: "subscribe", channel: MultiChatChannel.name, identifiers: {})
+          wait_for_async
+        end
       end
     end
 
     private
       def subscribe_to(connection, identifiers:)
-        receive connection, command: 'subscribe', identifiers: 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'
+        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?
+          assert_predicate connection.websocket, :possible?
 
           wait_for_async
-          assert connection.websocket.alive?
+          assert_predicate connection.websocket, :alive?
         end
       end
 
-      def receive(connection, command:, identifiers:)
-        identifier = JSON.generate(channel: 'ActionCable::StreamTests::ChatChannel', **identifiers)
+      def receive(connection, command:, identifiers:, channel: "ActionCable::StreamTests::ChatChannel")
+        identifier = JSON.generate(identifiers.merge(channel: channel))
         connection.dispatch_websocket_message JSON.generate(command: command, identifier: identifier)
         wait_for_async
       end