Deconstructing Action Cable

I knew what Action Cable was about and roughly how it worked since it was released. But lately I've developed a much deeper understanding of it because I had to extend it and explain its internals to others.

From HTTP to WebSocket

And that's exactly what a browser does when it connects to a WebSocket. It first makes a request with the "Upgrade" header set to "websocket" which tells the server "hey, I want to switch to using WebSockets for this request". If the server supports WebSockets it responds with status 101 indicating that the upgrade was accepted and that, from now on, it will communicate using the WebSocket protocol on that request's connection.

But why do we need the HTTP request in the first place? Can't we just connect directly via the WebSocket protocol since we already know that the server supports it?

Connections

#!/usr/bin/ruby
module ApplicationCable
  class Connection < ActionCable::Connection::Base
    attr_accessor :current_person
    
    def connect
	  Rails.logger.debug "Someone wants to connect via WebSocket!"
	  set_current_person
	  
	  unless current_person
	    Rails.logger.debug "I don't know who this is. Closing the WebSocket."
	    reject_unauthorized_connection
	  end
	  
	  Rails.logger.debug "It's, #{current_person.name}"
    end

    private

    def set_current_person
      person = 
	    # We can access the param of the HTTP request
	    Person.find_by(token: request.params[:token]) ||
	    # We can access the session of the HTTP request
        Person.find_by(token: request.session[:person_id])
        
      unless person
        # We can access the cookies of the HTTP request
        session = Session.find_by(id: cookies.encrypted[:session_id])
        person = session&.person
      end
	  
	  self.current_person = person
    end
  end
end

#!/usr/bin/ruby
module ApplicationCable
  class Connection < ActionCable::Connection::Base
    attr_accessor :current_person
    
    def connect
	  set_current_person
	  reject_unauthorized_connection unless current_person

	  current_person.increment!(:connection_count)
    end

    def disconnect
      current_person&.decrement!(:connection_count)
    end

    private

    # ...
  end
end

#!/usr/bin/ruby
module ApplicationCable
  class Connection < ActionCable::Connection::Base
    #attr_accessor :current_person # CHANGE THIS
    identified_by :current_person # TO THIS
    
    def connect
	  set_current_person
	  reject_unauthorized_connection unless current_person
    end

    private

    # ...
  end
end

The Wild WebSocket West

When you send a message to a client - or vice versa - the the object you are sending will first be turned into a JSON-compatible object like a String, a Number, a Hash, or an Array. Then it will be wrapped inside a Hash under a key called "message" together with some additional things I'll explain in a bit. Then that Hash will be dumped as JSON and sent as an UTF-8 encoded String, byte for byte through the WebSocket. On the other end the message is assembled into a String again and parsed into an object.

Why do we need the wrapper Hash?

• subscribe
• unsubscribe
• message

• welcome
• disconnect
• ping
• confirm_subscription
• reject_subscription
• message

{
  "type": "welcome"
}

{
  "type": "ping",
  "message": 1705848059
}

• unauthorized - sent when "reject_unauthorized_connection" is called
• invalid_request - sent when the initial HTTP request was malformed
• server_restart - sent when the server is about to restart
• remote - send when the client is kicked for whatever reason

{
  "type": "disconnect",
  "reconnect": true,
  "reason": "server_restart"
}

Channels

#!/usr/bin/ruby
class ArticlesController
  # POST /articles
  def create
    if Article.create(params.require(:article).permit(:title, :content))
      redirect_to action: :show, status: :see_other
    else
      render :new, status: :unprocesssable_entity
    end
  end
end

#!/usr/bin/ruby
class ChatChannel < ApplicationCable::Channel
  def post_message(data)
	  Chat::Message.create(
	    content: data[:content], 
	    poster: current_person
	  )
  end
end

{
  "command": "subscribe",
  "identifier": {
    "channel": "ChatChannel",
    "room_name": "Ruby Zagreb"
  }
}

#!/usr/bin/ruby
class ChatChannel < ApplicationCable::Channel
  def post_message(data)
	  current_person
	    .chat_rooms
	    .find_by(name: params[:room_name]) # PARAMS FROM THE SUBSCIBE MESSAGE
	    &.messages
	    &.create(content: data[:content])
  end
end

#!/usr/bin/ruby
class ChatChannel < ApplicationCable::Channel
  def subscribed
    @chat_room = current_person
	  .chat_rooms
	  .find_by(name: params[:room_name])
	  
	reject unless @chat_room
  end
  
  def post_message(data)
	@chat_room.messages.create(content: data[:content])
  end
end

{
  "type": "confirm_subscription",
  "identifier": {
    "channel": "ChatChannel",
    "room_name": "Ruby Zagreb"
  }
}

{
  "type": "reject_subscription",
  "identifier": {
    "channel": "ChatChannel",
    "room_name": "Ruby Zagreb"
  }
}

{
  "command": "message",
  "identifier": {
    "channel": "ChatChannel",
    "room_name": "Ruby Zagreb"
  },
  "data": "Hello, Zagreb!"
}

{
  "command": "message",
  "identifier": {
    "channel": "ChatChannel",
    "room_name": "Ruby Zagreb"
  },
  "data": {
    "action": "post_message",
    "content": "Hello, Zagreb!"
  }
}

In Action Cable, our Channel object is the subscriber and our application (controllers, models, jobs, ...) is the publisher.

To create a stream we can use either "stream_from" which creates a PubSub channel using a String identifier that we pass to it (e.g. "chat:1337"). Or “stream_for” which accepts an object (e.g. "Chat.find(1337)"). "stream_for" internally generates a String identifier for the given object and calls "stream_from" with it.

#!/usr/bin/ruby
class ChatChannel < ApplicationCable::Channel
  def subscribed
    @chat_room = current_person
	  .chat_rooms
	  .find_by(name: params[:room_name])
	  
	reject unless @chat_room

    # creates a PubSub channel to which we can publish messages to from anywhere
    stream_for @chat_room
    # the above is basicaly the same as
    # stream_from "chat_room:#{@chat_room.id}"
  end
end

#!/usr/bin/ruby
ChatChannel.broacast_to(
  ChatRoom.find(params[:id]), # The same record that we gave to stream_for
  "Hello, Zagreb!" # The message I want to send
)

#!/usr/bin/ruby
ActionCable.server.broadcast("chat_room:123", "Hello, Zagreb!")

#!/usr/bin/ruby
class ChatChannel < ApplicationCable::Channel
  def subscribed
    @chat_room = current_person
	  .chat_rooms
	  .find_by(name: params[:room_name])
	  
	reject unless @chat_room

    transmit "Welcome back, #{current_person.first_name}!"
  end
end

{
  "identifier": {
    "channel": "ChatChannel",
    "room_name": "Ruby Zagreb"
  },
  "message": "Hello, Zagreb!"
}

In the browser

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

const consumer = createConsumer()

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

// Fetches the user's auth token from a meta tag in the DOM
const token = document.querySelector("meta[name=authe_token]")?.content

// Takes the current URL,
// changes it's path to "/ws",
// and adds "?token=#{token}" as the query params.
// The result looks something like "https://example.com/ws?token=123"
const webSocketURL = new URL(window.location.href)
webSocketURL.pathname = "/ws" // usually it's "/cable"
webSocketURL.search = `?token=${token}`

const consumer = createConsumer(webSocketURL.toString())

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

const consumer = createConsumer()

// Subscribe to the ChatChannel 
// and pass { room_name: "Ruby Zagreb" } as params
consumer.subscriptions.create(
  { channel: "ChatChannel", room_name: "Ruby Zagreb" }
)

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

const consumer = createConsumer()

// Subscribe to the ChatChannel 
// and pass { room_name: "Ruby Zagreb" } as params
const subscription = consumer.subscriptions.create(
  { channel: "ChatChannel", room_name: "Ruby Zagreb" }
)

subscription.received = function(message) {
  document
    .querySelector("#messages")
    ?.insertAdjacentHTML("beforeend", `<div>${message}</div>`)
}

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

const consumer = createConsumer()

// Subscribe to the ChatChannel 
// and pass { room_name: "Ruby Zagreb" } as params
const subscription = consumer.subscriptions.create(
  { channel: "ChatChannel", room_name: "Ruby Zagreb" },
  {
    received(message) {
	  this.appendMessage(message)
    }
    
    appendMessage(message) {
	  this.messageContainer()?.insertAdjacentHTML(
		"beforeend", 
		`<div>${message}</div>`
	  )
    }
    
    messageContainer() {
	  document.querySelector("#messages")
    }
  }
)

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

const consumer = createConsumer()

const subscription = consumer.subscriptions.create(
  { channel: "ChatChannel", room_name: "Ruby Zagreb" },
  {
    // Called right after the Subscription object is created
    initialized() {
      console.log(`A subscription to ${this.identifier} was created`)
    }

    // Called if the subscription was rejected by the server
	rejected() {
	  console.log(`The server rejected the subscription to ${this.identifier}`)
	}

    // Called when the subscription gets confirmed by the server
    connected(data) {
	  // The data object has a single property `reconnected`
	  // which indicates if this was a resubscribe after a disconnect
      console.log(`The server confirmed the subscription to ${this.identifier}! Reconnected: ${data.reconnected}`)
    }

    // Called when the WebSocket closes
	disconnected(data) {
	  // The data object has a single propert `willAttemptReconnect`
	  // which indicates if a reconnect attempt will be made or not
	  console.log(`WebSocket to ${this.consumer.url} closed! Will attempt reconnect: ${data.willAttemptReconnect}`)
	}

    // Called when a message is sent from the server
    received(message) {
	  console.log(`Received message: ${message}`)
    }
  }
)

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

const consumer = createConsumer()

const subscription = consumer.subscriptions.create(
  { channel: "ChatChannel", room_name: "Ruby Zagreb" },
  {
    received(message) {
      console.log(`Received message: ${message}`)
      
      if (message.toLowerCase() === "avada kedavra") this.unsubscribe()
    }
  }
)

#!/usr/bin/node
import { createConsumer } from "@rails/actioncable"

const consumer = createConsumer()

const subscription = consumer.subscriptions.create(
  { channel: "ChatChannel", room_name: "Ruby Zagreb" }
)

subscription.send({ action: "post_message", content: "Hello, Zagreb!" })

Thousands of connections, one server

Once it's notified that a socket is ready, it reads its bytes, parses them into a message, and then passes the message to the Connection object, which then puts in in a thread pool to be processed - called the worker pool - to process that message.

Another thread pool?

Things I wish I knew right away

#!/usr/bin/ruby
class StockTickerChannel < ApplicationCable::Channel
  # this will send a message to the client every 2 seconds 
  # as long as they are subscribed
  periodically every: 2.seconds do
    transmit value: @stock.value, timestamp: Time.now.to_i
  end
  
  def subscribed
    @stock = Stock.find_by(symbol: params[:symbol])
	  
	reject unless @stock
  end
end

#!/usr/bin/ruby
module ApplicationCable
  class Connection < ActionCable::Connection::Base
    rescue_from Exception do |error|
      MyErrorTracker.capture_exception(error)
    end

    # ...
  end
end

#!/usr/bin/ruby
module ApplicationCable
  class Connection < ActionCable::Connection::Base
    around_command do
	  Current.set(person: current_person) { yield }
    end

    # ...
  end
end

#!/usr/bin/ruby
module AppearanceTrackable
  extend ActiveSupport::Concern

  included do
	after_subscribe unless: :subscription_rejected? do
	  Current.person.came_online!
	end
	
	after_unsubscribe do
	  Current.person.went_offline!
    end
  end
end

class ChatChannel < ApplicationCable::Channel
  include AppearanceTrackable
  
  def subscribed
    @chat_room = current_person
	  .chat_rooms
	  .find_by(name: params[:room_name])
	  
	reject unless @chat_room

    stream_for @chat_room
  end
end

#!/usr/bin/ruby
# config/puma.rb
require File.expand_path("../config/environment", File.dirname(__FILE__))

# ...

# Kill a worker thread if it didn't generate a responsw in 8 hours
worker_timeout 8 * 3600 if ENV.fetch("RAILS_ENV", "development") == "development"

#!/usr/bin/ruby
# config/puma.rb
max_threads_count = ENV.fetch("RAILS_MAX_THREADS") { 5 }
min_threads_count = ENV.fetch("RAILS_MIN_THREADS") { max_threads_count }
threads min_threads_count, max_threads_count

# config/application.rb
config.action_cable.worker_pool_size = ENV.fetch("RAILS_MAX_THREADS") { 5 }

#!/usr/bin/ruby
class Person < Applicationrecord
  def ban!
    update!(banned: true)
    
    ActionCable::RemoteConnections
      .where(current_person: self)
      &.disconnect(reconnect: false)
  end
end

# config/cable.yml
production:
  adapter: postgres

Final thought

Action Cable is an amazing piece of software. It can seem complex, but that’s because it does a lot. When you look at each piece individually the complexity goes away.