1 files changed, 142 insertions, 0 deletions
diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb
new file mode 100644
index 0000000000..d4a13be889
--- /dev/null
+++ b/actioncable/test/subscription_adapter/common.rb
@@ -0,0 +1,142 @@
+require 'test_helper'
+require 'concurrent'
+
+require 'action_cable/process/logging'
+require 'active_support/core_ext/hash/indifferent_access'
+require 'pathname'
+
+module CommonSubscriptionAdapterTest
+  WAIT_WHEN_EXPECTING_EVENT = 3
+  WAIT_WHEN_NOT_EXPECTING_EVENT = 0.2
+
+  def setup
+    # TODO: ActionCable requires a *lot* of setup at the moment...
+    ::Object.const_set(:ApplicationCable, Module.new)
+    ::ApplicationCable.const_set(:Connection, Class.new(ActionCable::Connection::Base))
+
+    ::Object.const_set(:Rails, Module.new)
+    ::Rails.singleton_class.send(:define_method, :root) { Pathname.new(__dir__) }
+
+    server = ActionCable::Server::Base.new
+    server.config = ActionCable::Server::Configuration.new
+    inner_logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN }
+    server.config.logger = ActionCable::Connection::TaggedLoggerProxy.new(inner_logger, tags: [])
+
+
+    # and now the "real" setup for our test:
+    spawn_eventmachine
+
+    server.config.cable = cable_config.with_indifferent_access
+
+    adapter_klass = server.config.pubsub_adapter
+
+    @rx_adapter = adapter_klass.new(server)
+    @tx_adapter = adapter_klass.new(server)
+  end
+
+  def teardown
+    @tx_adapter.shutdown if @tx_adapter && @tx_adapter != @rx_adapter
+    @rx_adapter.shutdown if @rx_adapter
+
+    begin
+      ::Object.send(:remove_const, :ApplicationCable)
+    rescue NameError
+    end
+    begin
+      ::Object.send(:remove_const, :Rails)
+    rescue NameError
+    end
+  end
+
+
+  def subscribe_as_queue(channel, adapter = @rx_adapter)
+    queue = Queue.new
+
+    callback = -> data { queue << data }
+    subscribed = Concurrent::Event.new
+    adapter.subscribe(channel, callback, Proc.new { subscribed.set })
+    subscribed.wait(WAIT_WHEN_EXPECTING_EVENT)
+    assert subscribed.set?
+
+    yield queue
+
+    sleep WAIT_WHEN_NOT_EXPECTING_EVENT
+    assert_empty queue
+  ensure
+    adapter.unsubscribe(channel, callback) if subscribed.set?
+  end
+
+
+  def test_subscribe_and_unsubscribe
+    subscribe_as_queue('channel') do |queue|
+    end
+  end
+
+  def test_basic_broadcast
+    subscribe_as_queue('channel') do |queue|
+      @tx_adapter.broadcast('channel', 'hello world')
+
+      assert_equal 'hello world', queue.pop
+    end
+  end
+
+  def test_broadcast_after_unsubscribe
+    keep_queue = nil
+    subscribe_as_queue('channel') do |queue|
+      keep_queue = queue
+
+      @tx_adapter.broadcast('channel', 'hello world')
+
+      assert_equal 'hello world', queue.pop
+    end
+
+    @tx_adapter.broadcast('channel', 'hello void')
+
+    sleep WAIT_WHEN_NOT_EXPECTING_EVENT
+    assert_empty keep_queue
+  end
+
+  def test_multiple_broadcast
+    subscribe_as_queue('channel') do |queue|
+      @tx_adapter.broadcast('channel', 'bananas')
+      @tx_adapter.broadcast('channel', 'apples')
+
+      received = []
+      2.times { received << queue.pop }
+      assert_equal ['apples', 'bananas'], received.sort
+    end
+  end
+
+  def test_identical_subscriptions
+    subscribe_as_queue('channel') do |queue|
+      subscribe_as_queue('channel') do |queue_2|
+        @tx_adapter.broadcast('channel', 'hello')
+
+        assert_equal 'hello', queue_2.pop
+      end
+
+      assert_equal 'hello', queue.pop
+    end
+  end
+
+  def test_simultaneous_subscriptions
+    subscribe_as_queue('channel') do |queue|
+      subscribe_as_queue('other channel') do |queue_2|
+        @tx_adapter.broadcast('channel', 'apples')
+        @tx_adapter.broadcast('other channel', 'oranges')
+
+        assert_equal 'apples', queue.pop
+        assert_equal 'oranges', queue_2.pop
+      end
+    end
+  end
+
+  def test_channel_filtered_broadcast
+    subscribe_as_queue('channel') do |queue|
+      @tx_adapter.broadcast('other channel', 'one')
+      @tx_adapter.broadcast('channel', 'two')
+
+      assert_equal 'two', queue.pop
+    end
+  end
+end