Allow rejecting subscriptions from the channel

3 files changed, 34 insertions, 8 deletions

diff --git a/lib/action_cable/channel/base.rb b/lib/action_cable/channel/base.rb
index 221730dbc4..31b8dece4a 100644
--- a/lib/action_cable/channel/base.rb
+++ b/lib/action_cable/channel/base.rb
@@ -74,11 +74,12 @@ module ActionCable
       include Broadcasting
 
       SUBSCRIPTION_CONFIRMATION_INTERNAL_MESSAGE = 'confirm_subscription'.freeze
+      SUBSCRIPTION_REJECTION_INTERNAL_MESSAGE = 'reject_subscription'.freeze
 
       on_subscribe   :subscribed
       on_unsubscribe :unsubscribed
 
-      attr_reader :params, :connection
+      attr_reader :params, :connection, :identifier
       delegate :logger, to: :connection
 
       class << self
@@ -170,8 +171,6 @@ module ActionCable
           connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data)
         end
 
-
-      protected
         def defer_subscription_confirmation!
           @defer_subscription_confirmation = true
         end
@@ -184,6 +183,14 @@ module ActionCable
           @subscription_confirmation_sent
         end
 
+        def reject!
+          @reject_subscription = true
+        end
+
+        def subscription_rejected?
+          @reject_subscription
+        end
+
       private
         def delegate_connection_identifiers
           connection.identifiers.each do |identifier|
@@ -196,7 +203,12 @@ module ActionCable
 
         def subscribe_to_channel
           run_subscribe_callbacks
-          transmit_subscription_confirmation unless defer_subscription_confirmation?
+
+          if subscription_rejected?
+            reject_subscription
+          else
+            transmit_subscription_confirmation unless defer_subscription_confirmation?
+          end
         end
 
 
@@ -243,6 +255,16 @@ module ActionCable
           end
         end
 
+        def reject_subscription
+          connection.subscriptions.remove_subscription self
+          transmit_subscription_rejection
+        end
+
+        def transmit_subscription_rejection
+          logger.info "#{self.class.name} is transmitting the subscription rejection"
+          connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: SUBSCRIPTION_REJECTION_INTERNAL_MESSAGE)
+        end
+
     end
   end
 end
diff --git a/lib/action_cable/connection/base.rb b/lib/action_cable/connection/base.rb
index 9f74226f98..b3de4dd4a9 100644
--- a/lib/action_cable/connection/base.rb
+++ b/lib/action_cable/connection/base.rb
@@ -48,7 +48,7 @@ module ActionCable
       include InternalChannel
       include Authorization
 
-      attr_reader :server, :env
+      attr_reader :server, :env, :subscriptions
       delegate :worker_pool, :pubsub, to: :server
 
       attr_reader :logger
@@ -140,7 +140,7 @@ module ActionCable
 
       private
         attr_reader :websocket
-        attr_reader :subscriptions, :message_buffer
+        attr_reader :message_buffer
 
         def on_open
           connect if respond_to?(:connect)
diff --git a/lib/action_cable/connection/subscriptions.rb b/lib/action_cable/connection/subscriptions.rb
index 229be2a316..6199db4898 100644
--- a/lib/action_cable/connection/subscriptions.rb
+++ b/lib/action_cable/connection/subscriptions.rb
@@ -37,8 +37,12 @@ module ActionCable
 
       def remove(data)
         logger.info "Unsubscribing from channel: #{data['identifier']}"
-        subscriptions[data['identifier']].unsubscribe_from_channel
-        subscriptions.delete(data['identifier'])
+        remove_subscription subscriptions[data['identifier']]
+      end
+
+      def remove_subscription(subscription)
+        subscription.unsubscribe_from_channel
+        subscriptions.delete(subscription.identifier)
       end
 
       def perform_action(data)