diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c12e222..06ac4375 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/lib/rage/cable/channel.rb b/lib/rage/cable/channel.rb index 3a1c790d..e8290cb2 100644 --- a/lib/rage/cable/channel.rb +++ b/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!" }) diff --git a/lib/rage/cable/protocols/actioncable_v1_json.rb b/lib/rage/cable/protocols/actioncable_v1_json.rb index 1574278f..476ded55 100644 --- a/lib/rage/cable/protocols/actioncable_v1_json.rb +++ b/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 diff --git a/spec/integration/test_app/app/channels/time_channel.rb b/spec/integration/test_app/app/channels/time_channel.rb index b98956f5..4e2fa1d0 100644 --- a/spec/integration/test_app/app/channels/time_channel.rb +++ b/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 diff --git a/spec/integration/websockets/actioncable_v1_json_spec.rb b/spec/integration/websockets/actioncable_v1_json_spec.rb index 65965fc0..e0f0cb9c 100644 --- a/spec/integration/websockets/actioncable_v1_json_spec.rb +++ b/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