Cable: Extract stream handler construction

* Use separate stream handler builders for easy override and testing.
* Fix worker pool execution that was silently failing since it only
  expected connection receivers.

Sparked by code in #24162.

4 files changed, 13 insertions, 7 deletions

diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb
index df129a7c62..0b0c72ccf6 100644
--- a/actioncable/test/channel/stream_test.rb
+++ b/actioncable/test/channel/stream_test.rb
@@ -142,6 +142,7 @@ module ActionCable::StreamTests
         connection.websocket.expects(:transmit)
         @server.broadcast 'test_room_1', { foo: 'bar' }, coder: DummyEncoder
         wait_for_async
+        wait_for_executor connection.server.worker_pool.executor
       end
     end
 
diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb
index 5ac453db35..fe503fd703 100644
--- a/actioncable/test/client_test.rb
+++ b/actioncable/test/client_test.rb
@@ -226,7 +226,9 @@ class ClientTest < ActionCable::TestCase
       assert_equal(1, app.connections.count)
       assert(app.remote_connections.where(identifier: identifier))
 
-      channel = app.connections.first.subscriptions.send(:subscriptions).first[1]
+      subscriptions = app.connections.first.subscriptions.send(:subscriptions)
+      assert_not_equal 0, subscriptions.size, 'Missing EchoChannel subscription'
+      channel = subscriptions.first[1]
       channel.expects(:unsubscribed)
       c.close
       sleep 0.1 # Data takes a moment to process
diff --git a/actioncable/test/test_helper.rb b/actioncable/test/test_helper.rb
index de1ee96770..0a9ee7ce77 100644
--- a/actioncable/test/test_helper.rb
+++ b/actioncable/test/test_helper.rb
@@ -49,10 +49,7 @@ end
 
 module ConcurrentRubyConcurrencyHelpers
   def wait_for_async
-    e = Concurrent.global_io_executor
-    until e.completed_task_count == e.scheduled_task_count
-      sleep 0.1
-    end
+    wait_for_executor Concurrent.global_io_executor
   end
 
   def run_in_eventmachine
@@ -67,4 +64,10 @@ class ActionCable::TestCase < ActiveSupport::TestCase
   else
     include ConcurrentRubyConcurrencyHelpers
   end
+
+  def wait_for_executor(executor)
+    until executor.completed_task_count == executor.scheduled_task_count
+      sleep 0.1
+    end
+  end
 end
diff --git a/actioncable/test/worker_test.rb b/actioncable/test/worker_test.rb
index 7016da3493..e2c81fe312 100644
--- a/actioncable/test/worker_test.rb
+++ b/actioncable/test/worker_test.rb
@@ -33,12 +33,12 @@ class WorkerTest < ActiveSupport::TestCase
   end
 
   test "invoke" do
-    @worker.invoke @receiver, :run
+    @worker.invoke @receiver, :run, connection: @receiver.connection
     assert_equal :run, @receiver.last_action
   end
 
   test "invoke with arguments" do
-    @worker.invoke @receiver, :process, "Hello"
+    @worker.invoke @receiver, :process, "Hello", connection: @receiver.connection
     assert_equal [ :process, "Hello" ], @receiver.last_action
   end
 end