Module: Rage::PubSub

Defined in:
lib/rage/pubsub/pubsub.rb

Overview

Note:

The low-level API is primarily intended for integrations and gems. Most applications should use higher-level abstractions like Rage::Cable or Rage::SSE instead.

The module enables communication between Rage instances across worker processes and nodes.

It provides the infrastructure for multi-server setups for Rage::Cable and Rage::SSE, allowing broadcasting messages across multiple servers or from different runtimes, e.g. Sidekiq. It also exposes a low-level API via the PubSub.subscribe, PubSub.unsubscribe, and PubSub.publish methods for direct inter-process communication.

Configuration

To use the module, add the redis-client gem to your Gemfile and create the environment-specific configuration in config/pubsub.yml:

production:
  adapter: redis
  url: <%= ENV["REDIS_URL"] %>

The configuration supports the following options:

  • adapter (required): The adapter to use for Pub/Sub. The only supported value is redis.
  • channel_prefix (optional): A prefix to use for the Redis stream name. This can be useful if you want to share a Redis instance with other applications or services.
  • pool_size (optional): The size of the Redis connection pool. Default is 10.
  • pool_timeout (optional): The timeout in seconds for acquiring a connection from the pool. Default is 1 second.

The rest of the options are passed directly to redis-client.

Defined Under Namespace

Modules: Adapters

Class Method Summary collapse

Class Method Details

.publish(message, topic:) ⇒ Object

Publish a message to all subscribers of a topic.

Messages are delivered to all subscribers across worker processes and nodes (when a PubSub adapter is configured).

Examples:

Publish an order update

Rage::PubSub.publish(JSON.generate({ id: 123, status: "shipped" }), topic: "orders")

Publish to a model-based topic

Rage::PubSub.publish("refresh", topic: current_user)

Parameters:

  • message (String) —

    The message to publish.

  • topic (String, Array, #id) —

    The channel name to publish to. Can be a string, an array, an Active Record model, or any object responding to id.

Raises:

  • (ArgumentError) —

    If message is not a String.



119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/rage/pubsub/pubsub.rb', line 119

def publish(message, topic:)
  raise ArgumentError, "Message must be a String, got #{message.class}" unless message.is_a?(String)

  serialized_topic = Rage::Internal.stream_name_for(topic)
  channel = "pubsub:#{serialized_topic}"

  Iodine.publish(channel, message) if Iodine.running?
  @__adapter&.publish(BROADCASTER_ID, channel, message)

  true
end

.subscribe(topic:, subscription_id:) {|message| ... } ⇒ Object

Subscribe to messages on a topic.

Examples:

Subscribe to order updates

Rage::PubSub.subscribe(topic: "orders", subscription_id: :order_listener) do |message|
  data = JSON.parse(message)
  puts "Received order update: #{data}"
end

Parameters:

  • topic (String, Array, #id) —

    The channel name to subscribe to. Can be a string, an array, an Active Record model, or any object responding to id.

  • subscription_id (Object) —

    A unique identifier for this subscription within the topic. Used to manage multiple subscriptions to the same topic and for cleanup via unsubscribe.

Yield Parameters:

  • message (String) —

    The raw message string.

Raises:

  • (ArgumentError) —

    If a subscription with the same topic and subscription_id already exists.



66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/rage/pubsub/pubsub.rb', line 66

def subscribe(topic:, subscription_id:, &block)
  serialized_topic = Rage::Internal.stream_name_for(topic)
  topic_subscriptions = (subscriptions[serialized_topic] ||= {})

  if topic_subscriptions.has_key?(subscription_id)
    raise ArgumentError, "A subscription with topic #{topic.inspect} and subscription_id #{subscription_id.inspect} already exists"
  end

  topic_subscriptions[subscription_id] = block
  subscribe_to_topic(serialized_topic) if topic_subscriptions.size == 1

  true
end

.unsubscribe(topic:, subscription_id: nil) ⇒ Object

Remove a subscription from a topic.

Examples:

Unsubscribe from order updates

Rage::PubSub.unsubscribe(topic: "orders", subscription_id: :order_listener)

Parameters:

  • topic (String, Array, #id) —

    The channel name to unsubscribe from. Must match the value used when calling subscribe.

  • subscription_id (Object, nil) (defaults to: nil) —

    The identifier of the subscription to remove. Omit the identifier to remove all subscriptions for the topic.



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/rage/pubsub/pubsub.rb', line 88

def unsubscribe(topic:, subscription_id: nil)
  serialized_topic = Rage::Internal.stream_name_for(topic)
  topic_subscriptions = subscriptions[serialized_topic]

  return false if topic_subscriptions.nil? || (subscription_id && !topic_subscriptions.has_key?(subscription_id))

  if subscription_id
    topic_subscriptions.delete(subscription_id)
    unsubscribe_from_topic(serialized_topic) if topic_subscriptions.empty?
  else
    unsubscribe_from_topic(serialized_topic)
  end

  true
end