CableRoom

Build live Rooms on top of ActionCable.

A Room is a long-lived, server-side object that owns a piece of shared realtime state — a quiz session, a collaborative document, a game lobby, a live dashboard. Exactly one instance of a Room runs across your whole cluster at a time, it processes messages one at a time on its own thread, and clients attach to it as ports.

ActionCable gives you channels, which are per-connection and stateless. CableRoom gives you the thing on the other side of those channels: a single authoritative object that outlives any one connection, holds state in memory, and shuts itself down when nobody needs it any more.

Contents.

How it works.

Every Room is backed by a synthetic ActionCable channel that has no browser connection behind it. That channel holds a Redis lock on the Room's key, which is what guarantees a single instance cluster-wide. Members don't talk to the Room directly. They publish to, and subscribe from, Redis pubsub keys called ports.

   Browser              Your ActionCable Channel                 The Room
      |                  (include RoomMember)                (Room::Base subclass)
      |                            |                                  |
      |  --- websocket message --> |  --- to_room port ------------->  |  handle_received_message
      |                            |                                  |
      |  <-- websocket message --- |  <-- from_room port -----------   |  broadcast / self <<
      |                            |  <-- <token> port -------------   |  reply
      |                            |  <-- <user> / <tag> port ------   |  broadcast(tag: :admin)

Four things follow from that design:

  • One instance, many processes. Any process can call MyRoom.ensure(key). The first one to win the Redis lock runs the Room; the rest get false and just publish to its ports.
  • No connection affinity. Members can be spread across every app server. They only need Redis.
  • Single-threaded Room state. Messages and timers run one at a time, so you can touch instance variables without locks.
  • Rooms are disposable. A Room is expected to die when it's idle and be re-created on demand. Persist anything you can't lose.

Requirements.

  • Ruby 3.4 (CI runs 3.4)
  • Rails 7.2 through 8.x
  • Redis, for both the ActionCable adapter and the Room locks

Installation.

gem "cable_room"

Then point it at Redis:

export CABLEROOM_REDIS_URL=redis://localhost:6379/1

See Configuration for the full list of variables.

Quick start.

Define a Room:

class ChatRoom < CableRoom::Room::Base
  # Shut down 30 seconds after the last member leaves
  reap_when { connected_clients.empty? }

  after_startup do
    @history = []
  end

  # Handles { "type": "chat", "body": "..." } from any member
  def on_chat(msg)
    entry = { user: message_origin.user, body: msg["body"], at: Time.current }
    @history << entry
    broadcast({ type: "chat", **entry })
  end

  # Send the backlog only to the member who just connected
  on_port_connected do
    reply({ type: "history", entries: @history })
  end
end

Define a channel that joins it:

class ChatChannel < ApplicationCable::Channel
  include CableRoom::RoomProxyChannel

  subscribe_to_room do
    join_room ChatRoom, params[:room_id], create: true, tags: params[:tags]
  end
end

That's the whole loop. RoomProxyChannel forwards everything the client sends into the Room and everything the Room broadcasts back out to the client. create: true means this channel will provision the Room if it isn't already running somewhere.

Defining a Room.

Lifecycle.

class MyRoom < CableRoom::Room::Base
  before_startup  { }   # streams aren't open yet
  after_startup   { }   # aliased as on_startup
  before_shutdown { }   # last chance to broadcast
  after_shutdown  { }   # aliased as on_shutdown
end

You can also just define startup and shutdown methods; they run inside the corresponding callback chain.

Out of the box a Room broadcasts { type: "room_opened" } after startup and { type: "room_closed", reason: ... } before shutdown.

To stop a Room from inside itself:

shutdown!("everyone left")   # graceful: drains queued messages first
stop!                        # immediate: drops anything pending

lifecycle_state returns :initializing, :starting, :started, :shutting_down, or :dead.

Handling messages.

Inbound messages arrive on the :to_room port and dispatch by type. A message of type "start_quiz" (or "StartQuiz") calls on_start_quiz. Unknown types log a warning.

def on_start_quiz(msg)
  logger.info "starting with #{msg['question_count']} questions"
end

Inside a handler:

Helper What it gives you
message The raw message hash
message_origin The PortClient that sent it
reply(data) Send back to that port alone
broadcast(data) Send to every member

Override handle_received_message(message) if you'd rather dispatch yourself. Call super for anything you don't handle, so the built-in port and user bookkeeping keeps working.

Sending the string "KILL" to the :to_room port shuts the Room down. It's a blunt instrument, useful in a console.

Timers.

class MyRoom < CableRoom::Room::Base
  periodically :tick, every: 5.seconds
  periodically -> { broadcast({ type: "still_here" }) }, every: 1.minute

  def tick; end
end

Timer bodies run on the Room's thread, so they're serialized against message handling.

Ports and messaging.

A port is one Redis pubsub key derived from the Room class, the Room key, and a port name. Two are reserved:

  • :to_room — many-to-one. Members publish here; the Room streams from it.
  • :from_room — one-to-many. The Room publishes here; every member streams from it.

Every member also gets a private port named after its random token, plus a port for the user it joined as and one for each tag it carries. That's how targeted delivery works without the Room tracking connections.

Sending.

self << { type: "tick" }                            # to :from_room, i.e. everyone
broadcast({ type: "tick" })                         # same thing
broadcast({ type: "secret" }, client_port: token)   # one port
broadcast({ type: "hi" }, user: "user_42")          # every port that user joined from
broadcast({ type: "tools" }, tag: :admin)           # every port carrying the tag
reply({ type: "pong" })                             # the port whose message you're handling
message_origin << { type: "pong" }                  # the same, spelled differently

Combining a target with a tag makes the tag a filter, not a second audience. broadcast(msg, user: "user_42", tag: :admin) reaches that user only if one of their ports is tagged admin, and sends nothing otherwise.

Scoping.

with_port_scope sets an ambient target so nested code doesn't have to pass it around:

with_port_scope(tag: :admin) do
  broadcast({ type: "diagnostics", data: expensive_report })
end

Scopes merge when nested. without_port_scope clears them. with_port_scope! skips the block entirely when nothing matches, which is the cheap way to avoid building a payload nobody will receive. The block form of reply does the same for a single port:

reply do
  broadcast({ type: "a" })
  broadcast({ type: "b" })
end

Custom ports.

Ports aren't limited to the built-ins. Open your own for a side channel:

ports[:telemetry] << { fps: 60 }

stream_port(:control) do |message|
  logger.info "control: #{message.inspect}"
end

Ports opened with stream_port close automatically at shutdown.

Port liveness.

Members ping every 10 seconds. A port that goes quiet for 30 seconds (PortManagement::PORT_TIMEOUT) is dropped, and on_port_disconnected runs for it with message_origin still set, so cleanup can tell which port went away.

on_port_connected    { logger.info "port #{message_origin.token} joined" }
on_port_disconnected { logger.info "port #{message_origin.token} gone" }

connected_clients returns the live PortClient objects. Each one carries a token, its tags, its user, and any extra metadata the member passed in. Read and write metadata with [] and []=.

Users.

Members can join as a user. CableRoom then collapses that user's ports into a single identity, so a person with three browser tabs joins once and leaves once.

class MyRoom < CableRoom::Room::Base
  on_user_joined { broadcast({ type: "joined", user: message_origin.user }) }
  on_user_left   { broadcast({ type: "left",   user: message_origin.user }) }
end

on_user_joined fires on the first port for that user; on_user_left fires when the last one goes away. connected_users lists them, and all_user_tags(user) unions the tags across every port that user is connected from.

A RoomMember channel that defines current_user passes it automatically. Pass as: to override it, or as: nil for an anonymous port. The value is serialized with ActiveJob's argument serializer, so an ActiveRecord object survives the trip and arrives as the same record.

Authorization.

Two layers, and they compose. Use guards for anything that depends on the message; use tag policies for anything that depends on who's asking.

Guards.

class MyRoom < CableRoom::Room::Base
  # Block, symbol, or proc. Return false to drop the message.
  authorize_inbound { |message| message["body"].to_s.length < 1_000 }
  authorize_inbound :quiz_running?, only: [:answer, :skip]
  authorize_inbound :not_locked?, except: :leave

  protected

  # Zero-arity guards read `message` themselves
  def quiz_running? = @state == :running
  def not_locked?(msg) = !@locked
end

A dropped message logs a warning and never reaches a handler.

Tag policies.

Members join with tags (join_room MyRoom, key, tags: [:admin]). Policies then say which tags may trigger which handlers.

class MyRoom < CableRoom::Room::Base
  inbound_tag_policy do
    deny  :muted, :chat          # muted members can't chat...
    allow :*,     :chat          # ...but everyone else can
    allow :admin, [:kick, :ban]  # admins get the moderation verbs
  end
end

Two rules govern how this resolves:

  1. Declaring any policy flips the default to deny. Before you write one, everything is allowed. After, only what you allow is allowed. The built-in connection and user messages stay permitted, so members can still join and leave.
  2. Highest priority wins. Rules default to priority 10. Pass priority: to layer a base policy under, or an override over, another. inbound_tag_policy(priority: -10) adds permissions without flipping the default.

Within a priority, the first matching rule decides, and rules match in declaration order. That means a deny exception has to come before the broad allow it carves out of — write allow :*, :chat first and it swallows every member, muted ones included. When the ordering matters a lot, give the two rules different priorities instead of relying on where they sit in the block:

inbound_tag_policy(priority: 20) { deny :muted, :chat }
inbound_tag_policy(priority: 10) { allow :*, :chat }

Group related handlers behind one name with define_tag_alias. A rule written against the alias covers everything it implies:

class MyRoom < CableRoom::Room::Base
  define_tag_alias :moderation, [:kick, :ban, :mute]

  inbound_tag_policy do
    allow :admin, :moderation
  end
end

Aliases are per Room class and inherited by subclasses, so one Room's vocabulary can't change how another Room reads its policies.

System message types.

Some message types are the framework's, not the client's. Members can't forge them:

class MyRoom < CableRoom::Room::Base
  system_message_types :score_awarded, :quiz_finished
end

Attempts to send one from a member are dropped with a warning at the sender. port_connected, port_disconnected, port_ping, user_joined, and user_left are already protected.

Reaping and the watchdog.

Rooms hold memory and a Redis lock, so they need to know when to quit. reap_when declares a check that runs on a timer:

class MyRoom < CableRoom::Room::Base
  # Idle for 30 seconds with nobody connected -> shut down
  reap_when { connected_clients.empty? }

  # Tighter window, and named so the shutdown reason says which check fired
  reap_when(key: :abandoned, grace: 5.minutes, interval: 30.seconds) do
    connected_users.empty?
  end

  # Return :reap to skip the grace period entirely
  reap_when(grace: 1.hour) { @cancelled ? :reap : false }
end
  • Truthy starts the grace clock. Once the condition has held for grace: (default 30 seconds), the Room shuts down.
  • Falsey resets the clock and pings the watchdog.
  • :reap shuts down now, whatever the grace period says.

Declare as many checks as you like; each gets its own timer and its own grace clock. Call check_reapers_now! to run them all immediately instead of waiting for the next tick.

The watchdog.

Separately, every Room is supervised. Every five seconds its channel extends the Redis lock and confirms the Room has pinged its watchdog within the last 15 seconds (Room::Base::WATCH_DOG_INTERVAL). Lose the lock and the Room stops, since another process may now own the key. Miss the ping and it shuts down as wedged.

Reaper checks are what ping the watchdog. A Room that declares no reap_when has nothing pinging it, so the watchdog will shut it down about 15 seconds after startup. Every long-lived Room needs at least one reap_when — or its own timer calling ping_watchdog — to stay up.

Joining a Room from a channel.

The proxy shortcut.

When the client only needs a pipe to the Room, RoomProxyChannel is the whole channel:

class QuizChannel < ApplicationCable::Channel
  include CableRoom::RoomProxyChannel

  subscribe_to_room do
    join_room QuizRoom, params[:quiz_id], create: true
  end
end

It wires up subscribed, receive, and unsubscribed, and forwards messages both ways.

Full control.

RoomMember gives you the membership without the forwarding, so the channel can filter, transform, or fan out:

class QuizChannel < ApplicationCable::Channel
  include CableRoom::RoomMember

  def subscribed
    @membership = join_room(
      QuizRoom,
      params[:quiz_id],
      create: true,
      tags: current_user.teacher? ? [:admin] : [:student],
      extra: { device: params[:device] },
      on_joined:      ->(m) { transmit(type: "ready") },
      on_message:     ->(msg) { transmit(msg) if msg["type"] != "internal" },
      on_room_closed: ->(m) { transmit(type: "over") },
      on_left:        ->(m) { logger.info "left #{m.key}" }
    )
  end

  def answer(data)
    @membership << { type: "answer", choice: data["choice"] }
  end

  def unsubscribed
    @membership&.leave!
  end
end

join_room options:

Option Meaning
create: Provision the Room if it isn't running. Defaults to false.
as: The user identity. Defaults to current_user when the channel has one.
tags: Tags this port carries, for policies and targeted broadcasts.
extra: Extra metadata, readable on the Room's PortClient.
forward: Pipe every Room message straight to the websocket.
on_joined: The Room acknowledged this port.
on_message: Any message from the Room.
on_room_opened: The Room opened while we were connecting. Not guaranteed.
on_room_closed: The Room closed while we were connected. Not guaranteed.
on_left: This membership ended.

The returned membership responds to <<, connected?, left?, key, ping!, leave!, and rejoin!.

With create: true, provisioning is retried on every ping, not just at join. If the Room's host process dies, the next ping from any member brings it back somewhere else.

From outside a channel.

QuizRoom.ensure("quiz_9")                                        # => true if this process now runs it
QuizRoom.send_message("quiz_9", { type: "extend", by: 60 })      # publish to :to_room
QuizRoom.room_port_key("quiz_9", :from_room)                     # the raw pubsub key

Background work.

A Room is single-threaded on purpose. Slow work belongs off its thread:

def on_export(msg)
  token = message_origin.token   # capture before leaving the Room's thread

  async do
    report = build_expensive_report

    on_room_thread do
      broadcast({ type: "export_ready", url: report.url }, client_port: token)
    end
  end
end

async borrows a thread from the pool shared by every Room in the process and runs concurrently with the Room, so the block must not touch Room state. Capture what it needs first. Inside it, message is nil, and message_origin and reply point at whatever the Room is handling now rather than what it was handling when you called async.

on_room_thread queues work back onto the Room's thread, where state is safe again. Prefer handing results back that way over blocking on async work, since a blocked Room thread can starve its neighbours.

Instrumentation and errors.

Rooms swallow exceptions so one bad message can't take the Room down. That makes the error handler the only place you'll hear about it:

CableRoom.error_handler = ->(error, context) do
  Sentry.capture_exception(error, extra: context)
end

The context includes the Room, its class, its key, and the channel. An error.cable_room notification fires either way.

ActiveSupport notifications:

Event Payload
room_opened.cable_room room
room_closed.cable_room room, reason
message_received.cable_room room, message
port_connected.cable_room room, message
port_disconnected.cable_room room, reason, message
user_joined.cable_room room, user
user_left.cable_room room, user
error.cable_room error, plus context

port_disconnected reports a reason of :left for a clean departure and :timeout for a port that stopped pinging.

Every Room also gets a tagged logger, so logger.info from inside a Room is prefixed with the Room class and a short UUID. That UUID is how you follow one instance through the logs.

Configuration.

CableRoom keeps its own Redis pool, separate from the ActionCable adapter's, for locking. Configure it with these variables:

Variable Purpose
CABLEROOM_REDIS_URL The connection URL.
CABLEROOM_REDIS_PROVIDER Name of another variable holding the URL.
CABLEROOM_REDIS_POOL_SIZE Pool size. Defaults to RAILS_MAX_THREADS, then five.

Without a prefixed variable it falls back to REDIS_PROVIDER and REDIS_URL, so a single-Redis app needs no CableRoom-specific configuration at all. Reach the pool directly with CableRoom.redis { |conn| ... } and the lock manager with CableRoom.lock_manager.

Room threads come from a pool sized by ActionCable's own worker_pool_size.

Timings live in constants:

Constant Default What it controls
Room::Base::LOCK_DURATION 15.seconds Redis lock TTL, extended on every beat.
Room::Base::WATCH_DOG_INTERVAL 15.seconds How stale a watchdog ping may get.
PortManagement::PORT_TIMEOUT 30.seconds How long a silent port survives.
ChannelTracker::BEAT_INTERVAL 5.seconds Lock extension and watchdog sweep.

The first two are read as self::CONSTANT, so a Room subclass can redefine them. The other two are module constants that apply process-wide.

Shutdown.

On process exit, CableRoom asks every local Room to shut down gracefully and waits up to 15 seconds for them to drain. It also hooks ActionCable's restart, so a code reload in development stops Rooms instead of orphaning their locks.

Introspection.

CableRoom::Room.locally_open_rooms        # every Room running in this process
QuizRoom.locally_running_instances       # just the QuizRooms

Both are process-local. There's no cluster-wide registry — the Redis lock is the only source of truth about who owns a key.

Subclassing.

Room classes build a private Channel and PortClient for each subclass, chained to the parent's. Periodic timers, callbacks, policies, and tag aliases all inherit correctly through however many levels you need:

class BaseGameRoom < CableRoom::Room::Base
  periodically :tick, every: 1.second
  reap_when { connected_users.empty? }
end

class TriviaRoom < BaseGameRoom
  # keeps tick and the reaper, adds its own
  periodically :rotate_question, every: 30.seconds
end

Note that a Room's pubsub keys derive from its class name, so anonymous Room classes won't work.

Development.

Rooms need Redis and, for the test suite, Postgres:

bundle install
bundle exec rspec

To run against every supported Rails version:

bundle exec appraisal install
bundle exec appraisal rspec

The suite has two halves. Unit specs use RoomHarness#build_room, which runs a Room against a stub channel with no Redis and no pubsub, so logic is testable synchronously. End-to-end specs run the async ActionCable adapter and real message delivery, and wait on observable conditions with wait_until rather than sleeping.

spec/internal holds a Combustion app, so rackup boots a minimal Rails host if you want to poke at Rooms by hand.