Broadcaster

Broadcasting based on Redis PubSub

Requires a running Redis server.

Installation

Add this line to your application's Gemfile:

gem 'broadcaster'

And then execute:

$ bundle

Or install it yourself as:

$ gem install broadcaster

Usage

broadcaster = Broadcaster.new

subscription_id = broadcaster.subscribe 'channel_1' do |message|
  puts message
end

broadcaster.publish 'channel_1', 'text'
broadcaster.publish 'channel_1', 1
broadcaster.publish 'channel_1', key: 'value'

broadcaster.unsubscribe subscription_id
broadcaster.unsubscribe_all

Any callable object can be used instead of a block:

class Receiver
  def call(message)
    puts message
  end
end

broadcaster.subscribe 'channel_1', Receiver.new

Messages are serialized with Marshal, so any serializable Ruby object can be published.

Each subscriber runs isolated from the others: if one raises an error it is logged and the message is still delivered to the remaining subscribers.

Scoping

Channels are scoped by the broadcaster id. Without an explicit id each instance gets a random one, so instances are isolated from each other:

broadcaster_1 = Broadcaster.new
broadcaster_2 = Broadcaster.new

broadcaster_2.subscribe('channel_1') { |message| puts message }
broadcaster_1.publish 'channel_1', 'text' # not received by broadcaster_2

Sharing the same id makes them see each other's messages, even across processes:

broadcaster_1 = Broadcaster.new id: 'my_app'
broadcaster_2 = Broadcaster.new id: 'my_app'

broadcaster_2.subscribe('channel_1') { |message| puts message }
broadcaster_1.publish 'channel_1', 'text' # received by broadcaster_2

Global configuration

Broadcaster.configure do |config|
  config.logger = Logger.new '/file.log'
  config.redis_settings = 'redis://host_name:6379'
  # or
  config.redis_client = Redic::Sentinels
  config.redis_settings = hosts: [sentinel_1, sentinel_2], master_name: 'mymaster'
end

Setting	Default	Description
redis_client	Redic	Redis client class
redis_settings	'redis://localhost:6379'	Connection settings passed to the client
logger	Logger.new(STDOUT)	Logger used by every broadcaster
reconnection_timeout	10	Seconds to wait before retrying a broken listener connection
watchdog_interval	30	Seconds between watchdog checks
watchdog_timeout	120	Seconds without listener activity before restarting it

Options

options = {
  id: 'my_app',                             # Shared broadcaster for multiple processes
  redis_settings: 'redis://host_name:6379', # Custom redis connection
  logger: Logger.new('/file.log'),          # Custom logger
  watchdog_interval: 30,                    # Watchdog tuning for this instance
  watchdog_timeout: 120
  # or
  redis_client: Redic::Sentinels,
  redis_settings: hosts: [sentinel_1, sentinel_2], master_name: 'mymaster'
}

Broadcaster.new options

reconnection_timeout can only be set globally.

Reconnection

The listener runs on its own thread. If the connection to Redis drops, it is re-established automatically after reconnection_timeout seconds, keeping the current subscriptions. Publishing while the connection is down raises an error.

Watchdog

Reconnection relies on the connection raising an error. A half-open socket never does: if the TCP connection is silently dropped, or a Redis Sentinel failover promotes a new master, the listener stays blocked reading from a socket that will never deliver another message. The process looks healthy while no message is received, and nothing recovers on its own.

The watchdog covers that case. It publishes a ping on its own channel every watchdog_interval seconds, and if no message reaches the listener for watchdog_timeout seconds it restarts the listener over a new connection — which also re-resolves the current master when using Redic::Sentinels.

watchdog_timeout must be comfortably greater than watchdog_interval, so that a single lost ping does not trigger a restart.

Shutdown

Stops the watchdog and the listener, and closes both connections:

broadcaster.shutdown

Contributing

Bug reports and pull requests are welcome on GitHub at https://github.com/gabynaiman/broadcaster.

License

The gem is available as open source under the terms of the MIT License.