Send subscription confirmation from server to the client to avoid race conditions.

Without this, it's very easy to send messages over a subscription even before the
redis pubsub has been fully initialized.

Now we delay calling the subscription#connected method on the client side until we
receive a subscription confirmation message from the server.

3 files changed, 33 insertions, 9 deletions

diff --git a/test/channel/base_test.rb b/test/channel/base_test.rb
index bac8569780..7eb8e15845 100644
--- a/test/channel/base_test.rb
+++ b/test/channel/base_test.rb
@@ -139,4 +139,10 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
     expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "message" => { "data" => "latest" }
     assert_equal expected, @connection.last_transmission
   end
+
+  test "subscription confirmation" do
+    expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription"
+    assert_equal expected, @connection.last_transmission
+  end
+
 end
diff --git a/test/channel/stream_test.rb b/test/channel/stream_test.rb
index 4e0248d7b4..cd0d3d1b83 100644
--- a/test/channel/stream_test.rb
+++ b/test/channel/stream_test.rb
@@ -12,28 +12,45 @@ class ActionCable::Channel::StreamTest < ActionCable::TestCase
     end
   end
 
-  setup do
-    @connection = TestConnection.new
-  end
-
   test "streaming start and stop" do
     run_in_eventmachine do
-      @connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1").returns stub_everything(:pubsub) }
-      channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
+      connection = TestConnection.new
+      connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1").returns stub_everything(:pubsub) }
+      channel = ChatChannel.new connection, "{id: 1}", { id: 1 }
 
-      @connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe_proc) }
+      connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe_proc) }
       channel.unsubscribe_from_channel
     end
   end
 
   test "stream_for" do
     run_in_eventmachine do
+      connection = TestConnection.new
       EM.next_tick do
-        @connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire").returns stub_everything(:pubsub) }
+        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire").returns stub_everything(:pubsub) }
       end
 
-      channel = ChatChannel.new @connection, ""
+      channel = ChatChannel.new connection, ""
       channel.stream_for Room.new(1)
     end
   end
+
+  test "stream_from subscription confirmation" do
+    EM.run do
+      connection = TestConnection.new
+      connection.expects(:pubsub).returns EM::Hiredis.connect.pubsub
+
+      channel = ChatChannel.new connection, "{id: 1}", { id: 1 }
+      assert_nil connection.last_transmission
+
+      EM::Timer.new(0.1) do
+        expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription"
+        assert_equal expected, connection.last_transmission, "Did not receive verification confirmation within 0.1s"
+
+        EM.run_deferred_callbacks
+        EM.stop
+      end
+    end
+  end
+
 end
diff --git a/test/test_helper.rb b/test/test_helper.rb
index f8a9971077..935e50e900 100644
--- a/test/test_helper.rb
+++ b/test/test_helper.rb
@@ -8,6 +8,7 @@ Bundler.setup
 Bundler.require :default, :test
 
 require 'puma'
+require 'em-hiredis'
 require 'mocha/mini_test'
 
 require 'rack/mock'