GodObject

To understand the following code you should first familiarize yourself with the basics of Action Cable, the Readme is a good start.

The goal here is to identify logged in users and do permission checks per message. One could also check permissions during initiation of the connection or the subscription of a channel, the most granular option is to verify permissions for each message. This can be beneficial if multiple types of messages or messages regarding different resources which require distinct permissions are delivered from the same queue.
Also imagine permissions change while a channel is subscribed, you would propably want to stop sending messages immediately if a user gets the permission to receive them revoked.

In the ApplicationCable we define methods to get the user from the session and Cancancan’s Ability through which we can check permissions.

module ApplicationCable
  class Connection < ActionCable::Connection::Base
    identified_by :current_user

    def connect
      self.current_user = find_verified_user
    end

    def session
      cookies.encrypted[Rails.application.config.session_options[:key]]
    end

    def ability
      @ability ||= Ability.new(current_user)
    end

    protected
    def find_verified_user
      User.find_by(id: session["user_id"])
    end
  end
end

We give Channel access to the session and the ability object. The current user is already accessable through current_user.

module ApplicationCable
  class Channel < ActionCable::Channel::Base
    delegate :session, :ability, to: :connection
    # dont allow the clients to call those methods
    protected :session, :ability
  end
end

So far we setup everything we need to verify permissions in our own channels.
So now we can use the ability object to deny subscription in general, or in this case to filter which messages are sent.

Notice: Currently using ActiveRecord from inside a stream callback depletes the connection pool. I reported this issue under #23778: ActionCable can deplete AR’s connection pool. Therefore we have to ensure the connection is checked back into the pool ourselfs.

class StreamUpdatesChannel < ApplicationCable::Channel
  def subscribed
    queue = "stream_updates:#{params[:stream_id]}"
    stream_from queue, -> (message) do
      ActiveRecord::Base.connection_pool.with_connection do
        if ability.can? :show, Stream.find(params[:stream_id])
          transmit ActiveSupport::JSON.decode(message), via: queue
        end
      end
    end
  end
end