Support multiple isolated cable apps

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 multiple isolated cable apps (#406).
 
 ## [1.28.2] - 2026-10-09
 

lib/rage/cable/cable.rb

@@ -38,9 +38,21 @@ module Rage::Cable
   #   map "/cable" do
   #     run Rage.cable.application
   #   end
-  def self.application
-    handler = __build_handler(__protocol)
-    accept_response = [0, __protocol.protocol_definition, []]
+  def self.application(namespace: Object, protocol: nil)
+    unless namespace.is_a?(Module)
+      raise ArgumentError, "The `namespace` option must be a Module, got #{namespace.class}."
+    end
+
+    protocol = __protocol_for(protocol || Rage.config.cable.protocol)
+
+    unless __protocols.include?(protocol)
+      __protocols << protocol.tap(&:init)
+      @__adapter&.add_broadcaster(PUBSUB_BROADCASTER_ID, Relay)
+    end
+
+    router = Router.new(protocol, namespace)
+    handler = __build_handler(protocol, router)
+    accept_response = [0, protocol.protocol_definition, []]
 
     application = ->(env) do
       Rage::Telemetry.tracer.span_cable_websocket_handshake(env:) do
@@ -54,8 +66,11 @@ def self.application
       end
     end
 
+    application.define_singleton_method(:__rage_app_name) do
+      namespace == Object ? "Rage::Cable" : "Rage::Cable(namespace: #{namespace})"
+    end
+
     chain = Rage.with_middlewares(application, Rage.config.cable.middlewares)
-    application.define_singleton_method(:__rage_app_name) { "Rage::Cable" }
     chain.define_singleton_method(:__rage_root_app) { application }
 
     chain
@@ -64,55 +79,66 @@ def self.application
   # @private
   def self.__initialize
     if (adapter = Rage.config.pubsub.adapter)
-      adapter.add_broadcaster(PUBSUB_BROADCASTER_ID, __protocol)
+      adapter.add_broadcaster(PUBSUB_BROADCASTER_ID, Relay)
       @__adapter = adapter
     end
   end
 
   # @private
-  def self.__router
-    @__router ||= Router.new
+  def self.__protocols
+    @__protocols ||= []
   end
 
   # @private
-  def self.__protocol
-    @__protocol ||= Rage.config.cable.protocol.tap { |protocol| protocol.init(__router) }
+  # @param protocol [:actioncable_v1_json, :raw_websocket_json, Class]
+  def self.__protocol_for(protocol)
+    case protocol
+    when Class
+      protocol
+    when :actioncable_v1_json
+      Protocols::ActioncableV1Json
+    when :raw_websocket_json
+      Protocols::RawWebSocketJson
+    else
+      raise ArgumentError, "Unknown protocol. Supported values are `:actioncable_v1_json` and `:raw_websocket_json`."
+    end
   end
 
   # @private
-  def self.__build_handler(protocol)
+  def self.__build_handler(protocol, router)
     klass = Class.new do
-      def initialize(protocol)
+      def initialize(protocol, router)
         Iodine.on_state(:on_start) do
           unless Fiber.scheduler
             Fiber.set_scheduler(Rage::FiberScheduler.new)
           end
         end
 
         @protocol = protocol
+        @router = router
         @log_processor = Rage.__log_processor
       end
 
       def on_open(connection)
         init_connection(connection)
-        schedule_processing(connection) { |env| @protocol.on_open(connection, env) }
+        schedule_processing(connection) { |env| @protocol.on_open(connection, env, @router) }
       end
 
       def on_message(connection, data)
-        schedule_processing(connection) { |env| @protocol.on_message(connection, env, data) }
+        schedule_processing(connection) { |env| @protocol.on_message(connection, env, data, @router) }
       end
 
       if protocol.respond_to?(:on_close)
         def on_close(connection)
           return unless ::Iodine.running?
-          schedule_processing(connection) { |env| @protocol.on_close(connection, env) }
+          schedule_processing(connection) { |env| @protocol.on_close(connection, env, @router) }
           connection.env["rage.cable.queue"] << nil
         end
       end
 
       if protocol.respond_to?(:on_shutdown)
         def on_shutdown(connection)
-          @protocol.on_shutdown(connection)
+          @protocol.on_shutdown(connection, @router)
         rescue => e
           log_error(e)
         end
@@ -153,7 +179,7 @@ def log_error(e)
       end
     end
 
-    klass.new(protocol)
+    klass.new(protocol, router)
   end
 
   # Broadcast data directly to a named stream.
@@ -165,13 +191,22 @@ def log_error(e)
   #   Rage.cable.broadcast("chat", { message: "A new member has joined!" })
   def self.broadcast(stream, data)
     Rage::Telemetry.tracer.span_cable_stream_broadcast(stream:) do
-      __protocol.broadcast(stream, data)
+      Relay.broadcast(stream, data)
       @__adapter&.publish(PUBSUB_BROADCASTER_ID, stream, data)
     end
 
     true
   end
 
+  # @private
+  module Relay
+    def self.broadcast(stream, data)
+      Rage::Cable.__protocols.each do |protocol|
+        protocol.broadcast(stream, data)
+      end
+    end
+  end
+
   # @!parse [ruby]
   #   # @abstract
   #   class WebSocketConnection

lib/rage/cable/channel.rb

@@ -4,19 +4,18 @@ class Rage::Cable::Channel
 
   class << self
     # @private
-    attr_reader :__prepared_actions
-
-    # @private
-    attr_reader :__channels
+    attr_reader :__prepared_actions, :__channels, :__protocol
 
     # @private
     # returns a list of actions that can be called remotely
-    def __register_actions
+    def __register_actions(protocol)
+      @__protocol = protocol
+
       actions = (
         public_instance_methods(true) - Rage::Cable::Channel.public_instance_methods(true)
       ).reject { |m| m.start_with?("__rage_tmp") || m.start_with?("__run") }
 
-      actions.reject! { |m| m != :receive } unless Rage.cable.__protocol.supports_rpc?
+      actions.reject! { |m| m != :receive } unless @__protocol.supports_rpc?
 
       @__prepared_actions = (INTERNAL_ACTIONS + actions).each_with_object({}) do |action_name, memo|
         memo[action_name] = __register_action_proc(action_name)
@@ -417,6 +416,13 @@ def params
     @__params
   end
 
+  # Get the Rack env hash that is attached to the underlying WebSocket connection.
+  #
+  # @return [Hash]
+  def env
+    @__connection.env
+  end
+
   # Reject the subscription request. The method should only be called during the subscription
   # process (i.e. inside the {subscribed} method or {before_subscribe}/{after_subscribe} hooks).
   def reject
@@ -449,7 +455,7 @@ def subscription_rejected?
   #   Rage::Cable.broadcast("notifications", { message: "A new member has joined!" })
   def stream_from(stream)
     raise ArgumentError, "Stream name must be a String" unless stream.is_a?(String)
-    Rage.cable.__protocol.subscribe(@__connection, stream, @__params)
+    self.class.__protocol.subscribe(@__connection, stream, @__params)
   end
 
   # Subscribe to a local stream. Local streams are associated with a specific channel instance and can be used to send data to the current channel only.
@@ -490,7 +496,7 @@ def stream_for(streamable)
   #   end
   def stop_stream_from(stream)
     raise ArgumentError, "Stream name must be a String" unless stream.is_a?(String)
-    Rage.cable.__protocol.unsubscribe(@__connection, stream, @__params)
+    self.class.__protocol.unsubscribe(@__connection, stream, @__params)
   end
 
   # Unsubscribe from a local stream. The counterpart to {stream_for}.
@@ -527,7 +533,7 @@ def broadcast(stream, data)
   #     transmit({ message: "Hello!" })
   #   end
   def transmit(data)
-    message = Rage.cable.__protocol.serialize(@__params, data)
+    message = self.class.__protocol.serialize(@__params, data)
 
     if @__is_subscribing
       # we expect a confirmation message to be sent as a result of a successful subscribe call;

lib/rage/cable/protocols/actioncable_v1_json.rb

@@ -67,9 +67,7 @@ def self.protocol_definition
   end
 
   # This method serves as a constructor to prepare the object or set up recurring tasks (e.g. heartbeats).
-  #
-  # @param router [Rage::Cable::Router]
-  def self.init(router)
+  def self.init
     super
 
     Iodine.on_state(:on_start) do
@@ -87,9 +85,9 @@ def self.init(router)
   #
   # @param connection [Rage::Cable::WebSocketConnection] the connection object
   # @param env [Hash] the Rack environment
-  # @see Rage::Cable::Router
-  def self.on_open(connection, env)
-    accepted = @router.process_connection(env)
+  # @param router [Rage::Cable::Router] the Cable router
+  def self.on_open(connection, env, router)
+    accepted = router.process_connection(connection, env)
 
     if accepted
       connection.subscribe("cable:ping")
@@ -106,16 +104,16 @@ def self.on_open(connection, env)
   # @param connection [Rage::Cable::WebSocketConnection] the connection object
   # @param env [Hash] the Rack environment
   # @param raw_data [String] the message body
-  # @see Rage::Cable::Router
-  def self.on_message(connection, env, raw_data)
+  # @param router [Rage::Cable::Router] the Cable router
+  def self.on_message(connection, env, raw_data, router)
     parsed_data = Rage::ParamsParser.json_parse(raw_data)
 
     command, identifier = parsed_data[:command], parsed_data[:identifier]
     params = Rage::ParamsParser.json_parse(identifier)
 
     # process subscription messages
     if command == COMMAND::SUBSCRIBE
-      status = @router.process_subscription(connection, env, identifier, params[:channel], params)
+      status = router.process_subscription(connection, env, identifier, params[:channel], params)
       if status == :subscribed
         connection.write({ identifier: identifier, type: TYPE::CONFIRM }.to_json)
       elsif status == :rejected
@@ -126,7 +124,7 @@ def self.on_message(connection, env, raw_data)
 
       return
     elsif command == COMMAND::UNSUBSCRIBE
-      @router.process_disconnection(env)
+      router.process_disconnection(connection, env)
       return
     end
 
@@ -135,10 +133,10 @@ def self.on_message(connection, env, raw_data)
     data = JSON.parse(parsed_data[:data])
 
     message_status = if command == COMMAND::MESSAGE && data.has_key?("action")
-      @router.process_message(env, identifier, data["action"].to_sym, data)
+      router.process_message(connection, env, identifier, data["action"].to_sym, data)
 
     elsif command == COMMAND::MESSAGE
-      @router.process_message(env, identifier, :receive, data)
+      router.process_message(connection, env, identifier, :receive, data)
     end
 
     unless message_status == :processed
@@ -149,10 +147,11 @@ def self.on_message(connection, env, raw_data)
   # The method should process client disconnections and call {Rage::Cable::Router#process_disconnection}.
   #
   # @note This method is optional.
+  # @param connection [Rage::Cable::WebSocketConnection] the connection object
   # @param env [Hash] the Rack environment
-  # @see Rage::Cable::Router
-  def self.on_close(_, env)
-    @router.process_disconnection(env)
+  # @param router [Rage::Cable::Router] the Cable router
+  def self.on_close(connection, env, router)
+    router.process_disconnection(connection, env)
   end
 
   # Serialize a Ruby object into the format the client would understand.
@@ -162,4 +161,10 @@ def self.on_close(_, env)
   def self.serialize(params, data)
     { identifier: params.to_json, message: data }.to_json
   end
+
+  # A unique identifier of the protocol used in multi-server setups by {Rage::PubSub}.
+  # @return [String]
+  def self.protocol_id
+    "acv1"
+  end
 end

lib/rage/cable/protocols/base.rb

@@ -9,10 +9,12 @@
 # * `on_open`
 # * `on_message`
 # * `serialize`
+# * `protocol_id`
 #
 # The optional methods are:
 #
 # * `protocol_definition`
+# * `supports_rpc?`
 # * `on_shutdown`
 # * `on_close`
 #
@@ -21,26 +23,24 @@ class Rage::Cable::Protocols::Base
   HANDSHAKE_HEADERS = {}
 
   class << self
-    # @param router [Rage::Cable::Router]
-    def init(router)
-      @router = router
-
+    def init
       # Hash<String(stream name) => Set<Hash>(subscription params)>
       @subscription_identifiers = Hash.new { |hash, key| hash[key] = Set.new }
+      @sync_channel = "cable:#{protocol_id}:synchronize"
 
       Iodine.on_state(:pre_start) do
         # this is a fallback to synchronize subscription identifiers across different worker processes;
         # we expect connections to be distributed among all workers, so this code will almost never be called;
         # we also synchronize subscriptions with the master process so that the forks that are spun up instead
         # of the crashed ones also had access to the identifiers;
-        Iodine.subscribe("cable:synchronize") do |_, subscription_msg|
+        Iodine.subscribe(@sync_channel) do |_, subscription_msg|
           stream_name, params = Rage::ParamsParser.json_parse(subscription_msg)
           @subscription_identifiers[stream_name] << params
         end
       end
 
       Iodine.on_state(:on_finish) do
-        Iodine.unsubscribe("cable:synchronize")
+        Iodine.unsubscribe(@sync_channel)
       end
     end
 
@@ -54,11 +54,11 @@ def protocol_definition
     # @param name [String] the stream name
     # @param params [Hash] parameters associated with the client
     def subscribe(connection, name, params)
-      connection.subscribe("cable:#{name}:#{stream_id(params)}")
+      connection.subscribe("cable:#{protocol_id}:#{name}:#{stream_id(params)}")
 
       unless @subscription_identifiers[name].include?(params)
         @subscription_identifiers[name] << params
-        ::Iodine.publish("cable:synchronize", [name, params].to_json)
+        ::Iodine.publish(@sync_channel, [name, params].to_json)
       end
     end
 
@@ -68,7 +68,7 @@ def subscribe(connection, name, params)
     # @param name [String] the stream name
     # @param params [Hash] parameters associated with the client
     def unsubscribe(connection, name, params)
-      connection.unsubscribe("cable:#{name}:#{stream_id(params)}")
+      connection.unsubscribe("cable:#{protocol_id}:#{name}:#{stream_id(params)}")
     end
 
     # Broadcast data to all clients connected to a stream.
@@ -77,7 +77,7 @@ def unsubscribe(connection, name, params)
     # @param data [Object] the data to send
     def broadcast(name, data)
       @subscription_identifiers[name].each do |params|
-        ::Iodine.publish("cable:#{name}:#{stream_id(params)}", serialize(params, data))
+        ::Iodine.publish("cable:#{protocol_id}:#{name}:#{stream_id(params)}", serialize(params, data))
       end
     end