1 files changed, 70 insertions, 14 deletions

diff --git a/lib/action_cable/channel/base.rb b/lib/action_cable/channel/base.rb
index 2f1b4a187d..ca903a810d 100644
--- a/lib/action_cable/channel/base.rb
+++ b/lib/action_cable/channel/base.rb
@@ -1,6 +1,8 @@
+require 'set'
+
 module ActionCable
   module Channel
-    # The channel provides the basic structure of grouping behavior into logical units when communicating over the websocket connection.
+    # The channel provides the basic structure of grouping behavior into logical units when communicating over the WebSocket connection.
     # You can think of a channel like a form of controller, but one that's capable of pushing content to the subscriber in addition to simply
     # responding to the subscriber's direct requests.
     #
@@ -64,6 +66,22 @@ module ActionCable
     #
     # Also note that in this example, current_user is available because it was marked as an identifying attribute on the connection.
     # All such identifiers will automatically create a delegation method of the same name on the channel instance.
+    #
+    # == Rejecting subscription requests
+    #
+    # A channel can reject a subscription request in the #subscribed callback by invoking #reject!
+    #
+    # Example:
+    #
+    #   class ChatChannel < ApplicationCable::Channel
+    #     def subscribed
+    #       @room = Chat::Room[params[:room_number]]
+    #       reject unless current_user.can_access?(@room)
+    #     end
+    #   end
+    #
+    # In this example, the subscription will be rejected if the current_user does not have access to the chat room.
+    # On the client-side, Channel#rejected callback will get invoked when the server rejects the subscription request.
     class Base
       include Callbacks
       include PeriodicTimers
@@ -71,10 +89,7 @@ module ActionCable
       include Naming
       include Broadcasting
 
-      on_subscribe   :subscribed
-      on_unsubscribe :unsubscribed
-
-      attr_reader :params, :connection
+      attr_reader :params, :connection, :identifier
       delegate :logger, to: :connection
 
       class << self
@@ -118,6 +133,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
@@ -138,8 +157,9 @@ module ActionCable
       # Called by the cable connection when its 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
-        run_unsubscribe_callbacks
-        logger.info "#{self.class.name} unsubscribed"
+        run_callbacks :unsubscribe do
+          unsubscribed
+        end
       end
 
 
@@ -160,9 +180,28 @@ module ActionCable
         # 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({ identifier: @identifier, message: data }.to_json)
+          connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data)
         end
 
+        def defer_subscription_confirmation!
+          @defer_subscription_confirmation = true
+        end
+
+        def defer_subscription_confirmation?
+          @defer_subscription_confirmation
+        end
+
+        def subscription_confirmation_sent?
+          @subscription_confirmation_sent
+        end
+
+        def reject
+          @reject_subscription = true
+        end
+
+        def subscription_rejected?
+          @reject_subscription
+        end
 
       private
         def delegate_connection_identifiers
@@ -175,8 +214,15 @@ module ActionCable
 
 
         def subscribe_to_channel
-          logger.info "#{self.class.name} subscribing"
-          run_subscribe_callbacks
+          run_callbacks :subscribe do
+            subscribed
+          end
+
+          if subscription_rejected?
+            reject_subscription
+          else
+            transmit_subscription_confirmation unless defer_subscription_confirmation?
+          end
         end
 
 
@@ -206,12 +252,22 @@ module ActionCable
           end
         end
 
-        def run_subscribe_callbacks
-          self.class.on_subscribe_callbacks.each { |callback| send(callback) }
+        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
+          end
+        end
+
+        def reject_subscription
+          connection.subscriptions.remove_subscription self
+          transmit_subscription_rejection
         end
 
-        def run_unsubscribe_callbacks
-          self.class.on_unsubscribe_callbacks.each { |callback| send(callback) }
+        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])
         end
     end
   end