1 files changed, 53 insertions, 37 deletions

diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb
index 874ebe2e71..a866044f95 100644
--- a/actioncable/lib/action_cable/channel/base.rb
+++ b/actioncable/lib/action_cable/channel/base.rb
@@ -1,4 +1,4 @@
-require 'set'
+require "set"
 
 module ActionCable
   module Channel
@@ -32,8 +32,8 @@ module ActionCable
     #
     # == Action processing
     #
-    # Unlike subclasses of ActionController::Base, channels do not follow a REST
-    # constraint form for their actions. Instead, ActionCable operates through a
+    # Unlike subclasses of ActionController::Base, channels do not follow a RESTful
+    # constraint form for their actions. Instead, Action Cable operates through a
     # remote-procedure call model. You can declare any public method on the
     # channel (optionally taking a <tt>data</tt> argument), and this method is
     # automatically exposed as callable to the client.
@@ -63,10 +63,10 @@ module ActionCable
     #       end
     #   end
     #
-    # In this example, subscribed/unsubscribed are not callable methods, as they
+    # In this example, the subscribed and unsubscribed methods are not callable methods, as they
     # were already declared in ActionCable::Channel::Base, but <tt>#appear</tt>
     # and <tt>#away</tt> are. <tt>#generate_connection_token</tt> is also not
-    # callable as it's a private method. You'll see that appear accepts a data
+    # callable, since it's a private method. You'll see that appear accepts a data
     # parameter, which it then uses as part of its model call. <tt>#away</tt>
     # does not, since it's simply a trigger action.
     #
@@ -125,7 +125,7 @@ module ActionCable
         protected
           # 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
+          # you run action_methods, they will be recalculated.
           def clear_action_methods!
             @action_methods = nil
           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
@@ -160,21 +161,34 @@ module ActionCable
         action = extract_action(data)
 
         if processable_action?(action)
-          dispatch_action(action, data)
+          payload = { channel_class: self.class.name, action: action, data: data }
+          ActiveSupport::Notifications.instrument("perform_action.action_cable", payload) do
+            dispatch_action(action, data)
+          end
         else
           logger.error "Unable to process #{action_signature(action, data)}"
         end
       end
 
-      # Called by the cable connection when its cut so the channel has a chance to cleanup with callbacks.
+      # 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
+      def unsubscribe_from_channel # :nodoc:
         run_callbacks :unsubscribe do
           unsubscribed
         end
       end
 
-
       protected
         # 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.
@@ -183,7 +197,7 @@ module ActionCable
         end
 
         # Called once a consumer has cut its cable connection. Can be used for cleaning up connections or marking
-        # people as offline or the like.
+        # users as offline or the like.
         def unsubscribed
           # Override in subclasses
         end
@@ -191,16 +205,26 @@ module ActionCable
         # 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)
-          logger.info "#{self.class.name} transmitting #{data.inspect}".tap { |m| m << " (via #{via})" if via }
-          connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data)
+          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 }
+          ActiveSupport::Notifications.instrument("transmit.action_cable", payload) do
+            connection.transmit identifier: @identifier, message: data
+          end
+        end
+
+        def ensure_confirmation_sent
+          return if subscription_rejected?
+          @defer_subscription_confirmation_counter.decrement
+          transmit_subscription_confirmation unless defer_subscription_confirmation?
         end
 
         def defer_subscription_confirmation!
-          @defer_subscription_confirmation = true
+          @defer_subscription_confirmation_counter.increment
         end
 
         def defer_subscription_confirmation?
-          @defer_subscription_confirmation
+          @defer_subscription_confirmation_counter.value > 0
         end
 
         def subscription_confirmation_sent?
@@ -224,26 +248,12 @@ 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
+          (data["action"].presence || :receive).to_sym
         end
 
         def processable_action?(action)
-          self.class.action_methods.include?(action.to_s)
+          self.class.action_methods.include?(action.to_s) unless subscription_rejected?
         end
 
         def dispatch_action(action, data)
@@ -258,7 +268,7 @@ module ActionCable
 
         def action_signature(action, data)
           "#{self.class.name}##{action}".tap do |signature|
-            if (arguments = data.except('action')).any?
+            if (arguments = data.except("action")).any?
               signature << "(#{arguments.inspect})"
             end
           end
@@ -267,8 +277,11 @@ module ActionCable
         def transmit_subscription_confirmation
           unless subscription_confirmation_sent?
             logger.info "#{self.class.name} is transmitting the subscription confirmation"
-            connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:confirmation])
-            @subscription_confirmation_sent = true
+
+            ActiveSupport::Notifications.instrument("transmit_subscription_confirmation.action_cable", channel_class: self.class.name) do
+              connection.transmit identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:confirmation]
+              @subscription_confirmation_sent = true
+            end
           end
         end
 
@@ -279,7 +292,10 @@ module ActionCable
 
         def transmit_subscription_rejection
           logger.info "#{self.class.name} is transmitting the subscription rejection"
-          connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:rejection])
+
+          ActiveSupport::Notifications.instrument("transmit_subscription_rejection.action_cable", channel_class: self.class.name) do
+            connection.transmit identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:rejection]
+          end
         end
     end
   end