Merge pull request #98 from rails/subscription-confirmation

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

5 files changed, 43 insertions, 9 deletions

diff --git a/lib/action_cable/channel/base.rb b/lib/action_cable/channel/base.rb
index df87064195..0b9aa2f4ad 100644
--- a/lib/action_cable/channel/base.rb
+++ b/lib/action_cable/channel/base.rb
@@ -73,6 +73,8 @@ module ActionCable
       include Naming
       include Broadcasting
 
+      SUBSCRIPTION_CONFIRMATION_INTERNAL_MESSAGE = 'confirm_subscription'.freeze
+
       on_subscribe   :subscribed
       on_unsubscribe :unsubscribed
 
@@ -120,6 +122,10 @@ module ActionCable
         @identifier = identifier
         @params     = params
 
+        # When a channel is streaming via redis pubsub, we want to delay the confirmation
+        # transmission until redis pubsub subscription is confirmed.
+        @defer_subscription_confirmation = false
+
         delegate_connection_identifiers
         subscribe_to_channel
       end
@@ -165,6 +171,15 @@ module ActionCable
         end
 
 
+      protected
+        def defer_subscription_confirmation!
+          @defer_subscription_confirmation = true
+        end
+
+        def defer_subscription_confirmation?
+          @defer_subscription_confirmation
+        end
+
       private
         def delegate_connection_identifiers
           connection.identifiers.each do |identifier|
@@ -177,6 +192,7 @@ module ActionCable
 
         def subscribe_to_channel
           run_subscribe_callbacks
+          transmit_subscription_confirmation unless defer_subscription_confirmation?
         end
 
 
@@ -213,6 +229,12 @@ module ActionCable
         def run_unsubscribe_callbacks
           self.class.on_unsubscribe_callbacks.each { |callback| send(callback) }
         end
+
+        def transmit_subscription_confirmation
+          logger.info "#{self.class.name} is transmitting the subscription confirmation"
+          connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: SUBSCRIPTION_CONFIRMATION_INTERNAL_MESSAGE)
+        end
+
     end
   end
 end
diff --git a/lib/action_cable/channel/streams.rb b/lib/action_cable/channel/streams.rb
index 9fffdf1789..b5ffa17f72 100644
--- a/lib/action_cable/channel/streams.rb
+++ b/lib/action_cable/channel/streams.rb
@@ -69,12 +69,18 @@ module ActionCable
       # Start streaming from the named <tt>broadcasting</tt> pubsub queue. Optionally, you can pass a <tt>callback</tt> that'll be used
       # instead of the default of just transmitting the updates straight to the subscriber.
       def stream_from(broadcasting, callback = nil)
-        callback ||= default_stream_callback(broadcasting)
+        # Hold off the confirmation until pubsub#subscribe is successful
+        defer_subscription_confirmation!
 
+        callback ||= default_stream_callback(broadcasting)
         streams << [ broadcasting, callback ]
-        EM.next_tick { pubsub.subscribe broadcasting, &callback }
 
-        logger.info "#{self.class.name} is streaming from #{broadcasting}"
+        EM.next_tick do
+          pubsub.subscribe(broadcasting, &callback).callback do |reply|
+            transmit_subscription_confirmation
+            logger.info "#{self.class.name} is streaming from #{broadcasting}"
+          end
+        end
       end
 
       # Start streaming the pubsub queue for the <tt>model</tt> in this channel. Optionally, you can pass a
diff --git a/lib/assets/javascripts/cable.coffee b/lib/assets/javascripts/cable.coffee
index 0bd1757505..476d90ef72 100644
--- a/lib/assets/javascripts/cable.coffee
+++ b/lib/assets/javascripts/cable.coffee
@@ -3,6 +3,8 @@
 
 @Cable =
   PING_IDENTIFIER: "_ping"
+  INTERNAL_MESSAGES:
+    SUBSCRIPTION_CONFIRMATION: 'confirm_subscription'
 
   createConsumer: (url) ->
     new Cable.Consumer url
diff --git a/lib/assets/javascripts/cable/connection.coffee b/lib/assets/javascripts/cable/connection.coffee
index 90d8fac3e1..33159130c7 100644
--- a/lib/assets/javascripts/cable/connection.coffee
+++ b/lib/assets/javascripts/cable/connection.coffee
@@ -52,8 +52,14 @@ class Cable.Connection
 
   events:
     message: (event) ->
-      {identifier, message} = JSON.parse(event.data)
-      @consumer.subscriptions.notify(identifier, "received", message)
+      {identifier, message, type} = JSON.parse(event.data)
+
+      if type?
+        switch type
+          when Cable.INTERNAL_MESSAGES.SUBSCRIPTION_CONFIRMATION
+            @consumer.subscriptions.notify(identifier, "connected")
+      else
+        @consumer.subscriptions.notify(identifier, "received", message)
 
     open: ->
       @disconnected = false
diff --git a/lib/assets/javascripts/cable/subscriptions.coffee b/lib/assets/javascripts/cable/subscriptions.coffee
index eeaa697081..4efb384ee2 100644
--- a/lib/assets/javascripts/cable/subscriptions.coffee
+++ b/lib/assets/javascripts/cable/subscriptions.coffee
@@ -21,13 +21,11 @@ class Cable.Subscriptions
   add: (subscription) ->
     @subscriptions.push(subscription)
     @notify(subscription, "initialized")
-    if @sendCommand(subscription, "subscribe")
-      @notify(subscription, "connected")
+    @sendCommand(subscription, "subscribe")
 
   reload: ->
     for subscription in @subscriptions
-      if @sendCommand(subscription, "subscribe")
-        @notify(subscription, "connected")
+      @sendCommand(subscription, "subscribe")
 
   remove: (subscription) ->
     @subscriptions = (s for s in @subscriptions when s isnt subscription)