3 files changed, 40 insertions, 35 deletions
diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb
index 2e589a2cfa..6739a62ba0 100644
--- a/actioncable/lib/action_cable/channel/base.rb
+++ b/actioncable/lib/action_cable/channel/base.rb
@@ -122,16 +122,16 @@ module ActionCable
           end
         end
 
-        protected
+        private
           # action_methods are cached and there is sometimes need to refresh
           # them. ::clear_action_methods! allows you to do that, so next time
           # you run action_methods, they will be recalculated.
-          def clear_action_methods!
+          def clear_action_methods! # :doc:
             @action_methods = nil
           end
 
           # Refresh the cached action_methods when a new action_method is added.
-          def method_added(name)
+          def method_added(name) # :doc:
             super
             clear_action_methods!
           end
@@ -144,13 +144,14 @@ module ActionCable
 
         # When a channel is streaming via pubsub, we want to delay the confirmation
         # transmission until pubsub subscription is confirmed.
-        @defer_subscription_confirmation = false
+        #
+        # The counter starts at 1 because it's awaiting a call to #subscribe_to_channel
+        @defer_subscription_confirmation_counter = Concurrent::AtomicFixnum.new(1)
 
         @reject_subscription = nil
         @subscription_confirmation_sent = nil
 
         delegate_connection_identifiers
-        subscribe_to_channel
       end
 
       # Extract the action name from the passed data and process it via the channel. The process will ensure
@@ -169,6 +170,17 @@ module ActionCable
         end
       end
 
+      # This method is called after subscription has been added to the connection
+      # and confirms or rejects the subscription.
+      def subscribe_to_channel
+        run_callbacks :subscribe do
+          subscribed
+        end
+
+        reject_subscription if subscription_rejected?
+        ensure_confirmation_sent
+      end
+
       # Called by the cable connection when it's cut, so the channel has a chance to cleanup with callbacks.
       # This method is not intended to be called directly by the user. Instead, overwrite the #unsubscribed callback.
       def unsubscribe_from_channel # :nodoc:
@@ -177,22 +189,22 @@ module ActionCable
         end
       end
 
-      protected
+      private
         # Called once a consumer has become a subscriber of the channel. Usually the place to setup any streams
         # you want this channel to be sending to the subscriber.
-        def subscribed
+        def subscribed # :doc:
           # Override in subclasses
         end
 
         # Called once a consumer has cut its cable connection. Can be used for cleaning up connections or marking
         # users as offline or the like.
-        def unsubscribed
+        def unsubscribed # :doc:
           # Override in subclasses
         end
 
         # Transmit a hash of data to the subscriber. The hash will automatically be wrapped in a JSON envelope with
         # the proper channel identifier marked as the recipient.
-        def transmit(data, via: nil)
+        def transmit(data, via: nil) # :doc:
           logger.info "#{self.class.name} transmitting #{data.inspect.truncate(300)}".tap { |m| m << " (via #{via})" if via }
 
           payload = { channel_class: self.class.name, data: data, via: via }
@@ -201,27 +213,32 @@ module ActionCable
           end
         end
 
-        def defer_subscription_confirmation!
-          @defer_subscription_confirmation = true
+        def ensure_confirmation_sent # :doc:
+          return if subscription_rejected?
+          @defer_subscription_confirmation_counter.decrement
+          transmit_subscription_confirmation unless defer_subscription_confirmation?
+        end
+
+        def defer_subscription_confirmation! # :doc:
+          @defer_subscription_confirmation_counter.increment
         end
 
-        def defer_subscription_confirmation?
-          @defer_subscription_confirmation
+        def defer_subscription_confirmation? # :doc:
+          @defer_subscription_confirmation_counter.value > 0
         end
 
-        def subscription_confirmation_sent?
+        def subscription_confirmation_sent? # :doc:
           @subscription_confirmation_sent
         end
 
-        def reject
+        def reject # :doc:
           @reject_subscription = true
         end
 
-        def subscription_rejected?
+        def subscription_rejected? # :doc:
           @reject_subscription
         end
 
-      private
         def delegate_connection_identifiers
           connection.identifiers.each do |identifier|
             define_singleton_method(identifier) do
@@ -230,18 +247,6 @@ module ActionCable
           end
         end
 
-        def subscribe_to_channel
-          run_callbacks :subscribe do
-            subscribed
-          end
-
-          if subscription_rejected?
-            reject_subscription
-          else
-            transmit_subscription_confirmation unless defer_subscription_confirmation?
-          end
-        end
-
         def extract_action(data)
           (data["action"].presence || :receive).to_sym
         end
diff --git a/actioncable/lib/action_cable/channel/naming.rb b/actioncable/lib/action_cable/channel/naming.rb
index b7e88bf73d..b565cb3cac 100644
--- a/actioncable/lib/action_cable/channel/naming.rb
+++ b/actioncable/lib/action_cable/channel/naming.rb
@@ -12,7 +12,7 @@ module ActionCable
         #   Chats::AppearancesChannel.channel_name # => 'chats:appearances'
         #   FooChats::BarAppearancesChannel.channel_name # => 'foo_chats:bar_appearances'
         def channel_name
-          @channel_name ||= name.sub(/Channel$/, "").gsub("::",":").underscore
+          @channel_name ||= name.sub(/Channel$/, "").gsub("::", ":").underscore
         end
       end
 
diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb
index 561750d713..dbba333353 100644
--- a/actioncable/lib/action_cable/channel/streams.rb
+++ b/actioncable/lib/action_cable/channel/streams.rb
@@ -69,8 +69,8 @@ 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.
-      # Pass `coder: ActiveSupport::JSON` to decode messages as JSON before passing to the callback.
-      # Defaults to `coder: nil` which does no decoding, passes raw messages.
+      # Pass <tt>coder: ActiveSupport::JSON</tt> to decode messages as JSON before passing to the callback.
+      # Defaults to <tt>coder: nil</tt> which does no decoding, passes raw messages.
       def stream_from(broadcasting, callback = nil, coder: nil, &block)
         broadcasting = String(broadcasting)
 
@@ -84,7 +84,7 @@ module ActionCable
 
         connection.server.event_loop.post do
           pubsub.subscribe(broadcasting, handler, lambda do
-            transmit_subscription_confirmation
+            ensure_confirmation_sent
             logger.info "#{self.class.name} is streaming from #{broadcasting}"
           end)
         end
@@ -94,8 +94,8 @@ module ActionCable
       # <tt>callback</tt> that'll be used instead of the default of just transmitting the updates straight
       # to the subscriber.
       #
-      # Pass `coder: ActiveSupport::JSON` to decode messages as JSON before passing to the callback.
-      # Defaults to `coder: nil` which does no decoding, passes raw messages.
+      # Pass <tt>coder: ActiveSupport::JSON</tt> to decode messages as JSON before passing to the callback.
+      # Defaults to <tt>coder: nil</tt> which does no decoding, passes raw messages.
       def stream_for(model, callback = nil, coder: nil, &block)
         stream_from(broadcasting_for([ channel_name, model ]), callback || block, coder: coder)
       end