message_bus

MessageBus

A reliable, robust messaging bus for Ruby processes and web clients.

MessageBus implements a Server to Server channel based protocol and Server to Web Client protocol (using polling, long-polling or long-polling + streaming)

Since long-polling is implemented using Rack Hijack and Thin::Async, all common Ruby web servers (Pitchfork/Unicorn, Puma, Thin, Passenger) can run MessageBus and handle a large number of concurrent connections that wait on messages.

MessageBus is implemented as Rack middleware and can be used by any Rails / Sinatra or pure Rack application.

Read the generated docs: https://www.rubydoc.info/gems/message_bus

Ruby version support

MessageBus only supports Ruby versions currently maintained by the Ruby core team. As of 2026-09-04, the supported versions are Ruby 3.3, 3.4, and 4.0; the minimum required version is Ruby 3.3.

Can you handle concurrent requests?

Yes, MessageBus uses Rack Hijack and this interface allows us to take control of the underlying socket. MessageBus can handle thousands of concurrent long polls on all popular Ruby webservers. MessageBus runs as middleware in your Rack (or by extension Rails) application and does not require a dedicated server. Background work is minimized to ensure it does not interfere with existing non-MessageBus traffic.

Is this used in production at scale?

Yes, MessageBus was extracted out of Discourse and is used in thousands of production Discourse sites at scale.

Installation

Add this line to your application's Gemfile:

gem 'message_bus'

And then execute:

$ bundle

Or install it yourself as:

$ gem install message_bus

Usage

Server to Server messaging

message_id = MessageBus.publish "/channel", "message"

# in another process / spot

MessageBus.subscribe "/channel" do |msg|
  # block called in a background thread when message is received
end

# subscribe to channel and receive the entire backlog
MessageBus.subscribe "/channel", 0 do |msg|
  # block called in a background thread when message is received
end

# subscribe to channel and receive the backlog starting at message 6
MessageBus.subscribe "/channel", 5 do |msg|
  # block called in a background thread when message is received
end

# get the ID of the last message on a channel
id = MessageBus.last_id("/channel")

# returns all messages after some id
MessageBus.backlog "/channel", id

Targeted messages

Messages can be targeted to particular clients by supplying the client_ids option when publishing a message.

MessageBus.publish "/channel", "hello", client_ids: ["XXX", "YYY"] # (using MessageBus.clientId)

By configuring the user_id_lookup and group_ids_lookup options with a Proc or Lambda which will be called with a Rack specification environment, messages can be targeted to particular clients users or groups by supplying either the user_ids or group_ids options when publishing a message.

…

If both user_ids and group_ids options are supplied when publishing a message, the message will be targeted at clients with lookup return values that matches on either the user_ids or the group_ids options.

MessageBus.publish "/channel", "hello", user_ids: [1, 2, 3], group_ids: [1, 2, 3]

If the client_ids option is supplied with either the user_ids or group_ids options when publishing a message, the client_ids option will be applied unconditionally and messages will be filtered further using user_id or group_id clauses.

MessageBus.publish "/channel", "hello", client_ids: ["XXX", "YYY"], user_ids: [1, 2, 3], group_ids: [1, 2, 3]

Passing nil or [] to either client_ids, user_ids or group_ids is equivalent to allowing all values on each option.

Filtering Client Messages

Custom client message filters can be registered via MessageBus#register_client_message_filter. This can be useful for filtering away messages from the client based on the message's payload.

For example, ensuring that only messages seen by the server in the last 20 seconds are published to the client:

MessageBus.register_client_message_filter('/test') do |message|
  (Time.now.to_i - message.data[:published_at]) <= 20
end

MessageBus.publish('/test/5', { data: "somedata", published_at: Time.now.to_i })

Error handling

MessageBus.configure(on_middleware_error: proc do |env, e|
   # If you wish to add special handling based on error
   # return a rack result array: [status, headers, body]
   # If you just want to pass it on return nil
end)

Disabling message_bus

In certain cases, it is undesirable for message_bus to start up on application start, for example in a Rails application during the db:create rake task when using the Postgres backend (which will error trying to connect to the non-existent database to subscribe). You can invoke MessageBus.off before the middleware chain is loaded in order to prevent subscriptions and publications from happening; in a Rails app you might do this in an initializer based on some environment variable or some other conditional means. If you want to just disable subscribing to the bus but want to continue to allow publications to be made, you can do MessageBus.off(disable_publish: false).

Debugging

When setting up MessageBus, it's useful to manually inspect channels before integrating a client application.

You can curl MessageBus; this is helpful when trying to debug what may be going wrong. This example uses https://chat.samsaffron.com:

curl -H "Content-Type: application/x-www-form-urlencoded" -X POST --data "/message=0" https://chat.samsaffron.com/message-bus/client-id/poll\?dlp\=t

You should see a reply with the messages of that channel you requested (in this case /message) starting at the message ID you requested (0). The URL parameter dlp=t disables long-polling: we do not want this request to stay open.

Transport

MessageBus ships with 3 transport mechanisms.

1. Long Polling with chunked encoding (streaming)
2. Long Polling
3. Polling

Long Polling with chunked encoding allows a single connection to stream multiple messages to a client, and this requires HTTP/1.1.

Chunked encoding provides all the benefits of EventSource with greater browser support (as it works on IE10 and up as well)

To setup NGINX to proxy to your app correctly be sure to enable HTTP1.1 and disable buffering:

location /message-bus/ {
  ...
  proxy_http_version 1.1;
  proxy_buffering off;
  ...
}

NOTE: do not set proxy_buffering off globally, it may have unintended consequences.

In order to disable chunked encoding for a specific client in Javascript:

MessageBus.enableChunkedEncoding = false;

or as a server-side policy in Ruby for all clients:

MessageBus.configure(chunked_encoding_enabled: false)

Long Polling requires no special setup; as soon as new data arrives on the channel the server delivers the data and closes the connection.

Polling also requires no special setup; MessageBus will fallback to polling after a tab becomes inactive and remains inactive for a period.

Multisite support

MessageBus can be used in an environment that hosts multiple sites by multiplexing channels. To use this mode:

# define a site_id lookup method, which is executed
# when `MessageBus.publish` is called
MessageBus.configure(site_id_lookup: proc do
  some_method_that_returns_site_id_string
end)

# you may post messages just to this site
MessageBus.publish "/channel", "some message"

# you can also choose to pass the `:site_id`.
# This takes precedence over whatever `site_id_lookup`
# returns
MessageBus.publish "/channel", "some message", site_id: "site-id"

# you may publish messages to ALL sites using the /global/ prefix
MessageBus.publish "/global/channel", "will go to all sites"

Client support

JavaScript Client

MessageBus ships a simple ~300 line JavaScript library which provides an API to interact with the server.

JavaScript clients can listen on any channel and receive messages via polling or long polling. You may simply include the source file (located in assets/ within the message_bus source code):

<script src="message-bus.js" type="text/javascript"></script>

or when used in a Rails application, import it through the asset pipeline:

//= require message-bus

In your application Javascript, you can then subscribe to particular channels and define callback functions to be executed when messages are received:

…

JavaScript Client settings

All client settings are settable via MessageBus.OPTION

Javascript Client API

MessageBus.start() : Starts up the MessageBus poller

MessageBus.subscribe(channel,func,lastId) : Subscribes to a channel. You may optionally specify the id of the last message you received in the channel. The callback receives three arguments on message delivery: func(payload, globalId, messageId). You may save globalId or messageId of received messages and use then at a later time when client needs to subscribe, receiving the backlog since that id.

MessageBus.unsubscribe(channel,func) : Removes a subscription from a particular channel that was defined with a particular callback function (optional).

MessageBus.pause() : Pauses all MessageBus activity

MessageBus.resume() : Resumes MessageBus activity

MessageBus.stop() : Stops all MessageBus activity

MessageBus.status() : Returns status (started, paused, stopped)

MessageBus.diagnostics() : Returns a log that may be used for diagnostics on the status of message bus.

Ruby

The gem ships with a Ruby implementation of the client library available with an API very similar to that of the Ja