Run Action Cable callbacks through the worker pool

Alternate implementation of #24162 with tests. The code had diverged
too far on master to pull that implemenation directly.

Fixes #23778
Close #24162

[Mattew Draper & Sean Griffin]

1 files changed, 25 insertions, 3 deletions

diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb
index f51f19eb7d..df129a7c62 100644
--- a/actioncable/test/channel/stream_test.rb
+++ b/actioncable/test/channel/stream_test.rb
@@ -116,11 +116,22 @@ module ActionCable::StreamTests
 
   require 'action_cable/subscription_adapter/inline'
 
+  class UserCallbackChannel < ActionCable::Channel::Base
+    def subscribed
+      stream_from :channel do
+        Thread.current[:ran_callback] = true
+      end
+    end
+  end
+
   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)
+      @server.stubs(:channel_classes).returns(
+        ChatChannel.name => ChatChannel,
+        UserCallbackChannel.name => UserCallbackChannel,
+      )
     end
 
     test 'custom encoder' do
@@ -134,6 +145,17 @@ module ActionCable::StreamTests
       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
+        refute Thread.current[:ran_callback], "User callback was not run through the worker pool"
+      end
+    end
+
     private
       def subscribe_to(connection, identifiers:)
         receive connection, command: 'subscribe', identifiers: identifiers
@@ -151,8 +173,8 @@ module ActionCable::StreamTests
         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(channel: channel, **identifiers)
         connection.dispatch_websocket_message JSON.generate(command: command, identifier: identifier)
         wait_for_async
       end