3 files changed, 35 insertions, 17 deletions
diff --git a/actioncable/test/stubs/room.rb b/actioncable/test/stubs/room.rb
index cd66a0b687..9e521cf3a6 100644
--- a/actioncable/test/stubs/room.rb
+++ b/actioncable/test/stubs/room.rb
@@ -1,7 +1,7 @@
 class Room
   attr_reader :id, :name
 
-  def initialize(id, name='Campfire')
+  def initialize(id, name="Campfire")
     @id = id
     @name = name
   end
diff --git a/actioncable/test/stubs/test_connection.rb b/actioncable/test/stubs/test_connection.rb
index da98201900..cd2e219d88 100644
--- a/actioncable/test/stubs/test_connection.rb
+++ b/actioncable/test/stubs/test_connection.rb
@@ -1,25 +1,33 @@
-require 'stubs/user'
+require "stubs/user"
 
 class TestConnection
-  attr_reader :identifiers, :logger, :current_user, :transmissions
+  attr_reader :identifiers, :logger, :current_user, :server, :transmissions
 
-  def initialize(user = User.new("lifo"))
+  delegate :pubsub, to: :server
+
+  def initialize(user = User.new("lifo"), coder: ActiveSupport::JSON, subscription_adapter: SuccessAdapter)
+    @coder = coder
     @identifiers = [ :current_user ]
 
     @current_user = user
     @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new)
+    @server = TestServer.new(subscription_adapter: subscription_adapter)
     @transmissions = []
   end
 
-  def pubsub
-    SuccessAdapter.new(TestServer.new)
+  def transmit(cable_message)
+    @transmissions << encode(cable_message)
   end
 
-  def transmit(data)
-    @transmissions << data
+  def last_transmission
+    decode @transmissions.last if @transmissions.any?
   end
 
-  def last_transmission
-    @transmissions.last
+  def decode(websocket_message)
+    @coder.decode websocket_message
+  end
+
+  def encode(cable_message)
+    @coder.encode cable_message
   end
 end
diff --git a/actioncable/test/stubs/test_server.rb b/actioncable/test/stubs/test_server.rb
index 56d132b30a..5bf2a151dc 100644
--- a/actioncable/test/stubs/test_server.rb
+++ b/actioncable/test/stubs/test_server.rb
@@ -1,20 +1,30 @@
-require 'ostruct'
+require "ostruct"
 
 class TestServer
   include ActionCable::Server::Connections
+  include ActionCable::Server::Broadcasting
 
-  attr_reader :logger, :config
+  attr_reader :logger, :config, :mutex
 
-  def initialize
+  def initialize(subscription_adapter: SuccessAdapter)
     @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new)
-    @config = OpenStruct.new(log_tags: [], subscription_adapter: SuccessAdapter)
+
+    @config = OpenStruct.new(log_tags: [], subscription_adapter: subscription_adapter)
+
+    @mutex = Monitor.new
   end
 
   def pubsub
-    @config.subscription_adapter.new(self)
+    @pubsub ||= @config.subscription_adapter.new(self)
+  end
+
+  def event_loop
+    @event_loop ||= ActionCable::Connection::StreamEventLoop.new.tap do |loop|
+      loop.instance_variable_set(:@executor, Concurrent.global_io_executor)
+    end
   end
 
-  def stream_event_loop
-    @stream_event_loop ||= ActionCable::Connection::StreamEventLoop.new
+  def worker_pool
+    @worker_pool ||= ActionCable::Server::Worker.new(max_size: 5)
   end
 end