Support unsubscribe commands in Action Cable

CHANGELOG.md

@@ -6,6 +6,7 @@
 - [Cable] Serialize WS messages per connection (#400).
 - Update fiber scheduler to support Ruby 4.1 (#397).
 - Improve blocking mechanism for fibers (#401).
+- Support `unsubscribe` commands in Action Cable.
 
 ## [1.28.0] - 2026-09-15
 

lib/rage/cable/channel.rb

@@ -434,11 +434,16 @@ def subscription_rejected?
   #
   # @param stream [String] the name of the stream
   # @raise [ArgumentError] if the stream name is not a String
+  # @note Calling {stop_stream_from} in {unsubscribed} is not strictly required, but is important when clients are dynamically subscribing/unsubscribing from channels on a single connection.
   # @example Subscribe to a stream
   #   class NotificationsChannel < Rage::Cable::Channel
   #     def subscribed
   #       stream_from "notifications"
   #     end
+  #
+  #     def unsubscribed
+  #       stop_stream_from "notifications"
+  #     end
   #   end
   # @example Broadcast to the stream
   #   Rage::Cable.broadcast("notifications", { message: "A new member has joined!" })
@@ -451,11 +456,16 @@ def stream_from(stream)
   #
   # @param streamable [#id, String, Symbol, Numeric, Array] an object that will be used to generate the stream name
   # @raise [ArgumentError] if the streamable object does not satisfy the type requirements
+  # @note Calling {stop_stream_for} in {unsubscribed} is not strictly required, but is important when clients are dynamically subscribing/unsubscribing from channels on a single connection.
   # @example Subscribe to a stream
   #   class NotificationsChannel < Rage::Cable::Channel
   #     def subscribed
   #       stream_for current_user
   #     end
+  #
+  #     def unsubscribed
+  #       stop_stream_for current_user
+  #     end
   #   end
   # @example Broadcast to the stream
   #   NotificationsChannel.broadcast_to(current_user, { message: "You have a new notification!" })

lib/rage/cable/protocols/actioncable_v1_json.rb

@@ -49,6 +49,7 @@ module REASON
 
   module COMMAND
     SUBSCRIBE = "subscribe"
+    UNSUBSCRIBE = "unsubscribe"
     MESSAGE = "message"
   end
 
@@ -123,6 +124,9 @@ def self.on_message(connection, env, raw_data)
         connection.write(MESSAGES::INVALID)
       end
 
+      return
+    elsif command == COMMAND::UNSUBSCRIBE
+      router.process_disconnection(env)
       return
     end
 

spec/integration/test_app/app/channels/time_channel.rb

@@ -16,4 +16,9 @@ def remote_sync_time(data)
     sleep 1
     transmit({ message: "synced from #{data["remote"]}" })
   end
+
+  def unsubscribed
+    stop_stream_from "current_time"
+    transmit({ message: "unsubscribed" })
+  end
 end

spec/integration/websockets/actioncable_v1_json_spec.rb

@@ -17,6 +17,7 @@
   let(:get_time_message) { { identifier: { client: "1", channel: "TimeChannel" }.to_json, command: "message", data: { action: "what_time_is_it" }.to_json }.to_json }
   let(:sync_time_message) { { identifier: { client: "1", channel: "TimeChannel" }.to_json, command: "message", data: { action: "sync_time" }.to_json }.to_json }
   let(:remote_sync_time_message) { { identifier: { client: "1", channel: "TimeChannel" }.to_json, command: "message", data: { action: "remote_sync_time", remote: "time.com" }.to_json }.to_json }
+  let(:unsubscribe_message) { { identifier: { client: "1", channel: "TimeChannel" }.to_json, command: "unsubscribe" }.to_json }
 
   it "rejects a connection from unknown origin" do
     with_websocket_connection("ws://localhost:3000/cable") do |client|
@@ -104,4 +105,12 @@
       expect(client.messages.last).to include("synced from time.com")
     end
   end
+
+  it "unsubscribes from streams" do
+    with_websocket_connection("ws://localhost:3000/cable?user_id=1", headers: { Origin: "localhost:3000" }) do |client|
+      client.send(subscribe_message)
+      client.send(unsubscribe_message)
+      expect(client.messages.last).to include("unsubscribed")
+    end
+  end
 end