Delayed Rabbit

A lightweight Ruby gem for scheduling delayed background jobs using RabbitMQ's delayed message exchange plugin.

• Schedules jobs with custom delays using RabbitMQ's x-delayed-message plugin
• Uses JSON serialization for job data
• Configurable exchange and queue settings
• Simple API for publishing delayed jobs
• Rails integration support
• Automatic connection management
• Persistent message delivery
• Support for custom routing keys
• Dead-letter queue support

delayed_rabbit only publishes jobs. Consume delayed_jobs_queue with your own worker (e.g. Bunny, Sneakers or a service in another language).

• Ruby 3.2+
• RabbitMQ 3.8+ with x-delayed-message plugin enabled
• Bunny gem (~> 2.18)
• JSON serialization support

Before using this gem, you need to enable the delayed message plugin in RabbitMQ:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

Add this line to your application's Gemfile:

gem 'delayed_rabbit'

And then execute:

bundle install

Or install it yourself as:

gem install delayed_rabbit

Add to your Gemfile:

gem 'delayed_rabbit'

Run the generator to set up configuration files:

rails generate delayed_rabbit:install

This will create:

• config/initializers/delayed_rabbit.rb - Main configuration file
• config/rabbitmq.yml - Environment-specific RabbitMQ settings

Edit config/rabbitmq.yml with your RabbitMQ settings:

development:
  host: localhost
  port: 5672
  user: guest
  password: guest
  vhost: /

production:
  host: <%= ENV['RABBITMQ_HOST'] %>
  port: <%= ENV['RABBITMQ_PORT'] || 5672 %>
  user: <%= ENV['RABBITMQ_USER'] %>
  password: <%= ENV['RABBITMQ_PASSWORD'] %>
  vhost: <%= ENV['RABBITMQ_VHOST'] || '/' %>

Settings from rabbitmq.yml are loaded before the app's initializers run, so anything set in config/initializers/delayed_rabbit.rb overrides them. If the file is missing, the defaults are used.

class UserNotifier
  def self.send_welcome_notification(user)
    DelayedRabbit.publish(
      {
        type: "welcome_email",
        user_id: user.id,
        email: user.email
      },
      delay_ms: 5000,  # 5 seconds delay
      routing_key: "notifications.welcome"
    )
  end
end

DelayedRabbit.publish reuses one shared connection (guarded by a mutex) across calls. DelayedRabbit::JobPublisher.publish takes the same arguments but opens and closes a connection for each message.

class UsersController < ApplicationController
  def create
    user = User.create(user_params)
    UserNotifier.send_welcome_notification(user)
    redirect_to root_path, notice: 'User created successfully'
  end
end

The gem provides a rake task for testing:

# Enqueue a test job (delay defaults to 5000ms)
bundle exec rake delayed_rabbit:enqueue_test_job
bundle exec rake delayed_rabbit:enqueue_test_job[10000]

If you're not using Rails:

require 'delayed_rabbit'

# Configure manually
DelayedRabbit.configure do |config|
  config.connection_options = {
    host: "localhost",
    port: 5672,
    user: "guest",
    password: "guest"
  }
end

# Create a publisher (uses DelayedRabbit.configuration)
publisher = DelayedRabbit::JobPublisher.new

# Publish a job
publisher.publish(
  {
    type: "email_notification",
    recipient: "user@example.com"
  },
  delay_ms: 5000
)

# Close connection
publisher.close

1. Add the gem to your Gemfile
2. Run the rake task to enqueue a test job:

bundle exec rake delayed_rabbit:enqueue_test_job[5000]

You can also create your own rake tasks for specific job types:

namespace :jobs do
  desc "Enqueue a notification job"
  task :enqueue_notification, [:delay_ms, :user_id] => :environment do |t, args|
    delay_ms = Integer(args[:delay_ms] || 5000)
    user_id = args[:user_id]
    
    job_data = {
      type: "notification",
      user_id: user_id,
      message: "Welcome notification"
    }

    DelayedRabbit::JobPublisher.publish(job_data, delay_ms: delay_ms, routing_key: "notifications.#{user_id}")
  end
end

Options passed to JobPublisher.new are merged over DelayedRabbit.configuration. The arguments hashes are merged key by key, so you only need to pass the arguments you add. To reuse an existing Bunny session, pass connection:; the publisher then won't close it.

DelayedRabbit.configure do |config|
  config.exchange_name = "delayed_jobs"
  config.queue_name = "delayed_jobs_queue"
  config.routing_key = "delayed_jobs"

  # Rejected or expired messages are routed to this fanout exchange and queue.
  # Set dead_letter_exchange_name to nil to disable.
  config.dead_letter_exchange_name = "delayed_jobs.dlx"
  config.dead_letter_queue_name = "delayed_jobs.dead"
end

publisher = DelayedRabbit::JobPublisher.new(
  exchange_options: {
    arguments: {"x-delayed-type" => "direct"}  # Change to direct exchange
  }
)

publisher = DelayedRabbit::JobPublisher.new(
  queue_options: {
    arguments: {"x-message-ttl" => 3600000}  # 1 hour TTL, then dead-lettered
  }
)

RabbitMQ rejects a declaration whose options differ from an existing exchange or queue (PRECONDITION_FAILED). After changing queue arguments, delete the queue so it can be declared again.

You can configure different settings based on Rails environment:

# config/initializers/delayed_rabbit.rb
DelayedRabbit.configure do |config|
  config.connection_options = {
    host: Rails.env.production? ? "rabbitmq-prod" : "localhost",
    port: 5672,
    user: Rails.application.credentials.rabbitmq[:user],
    password: Rails.application.credentials.rabbitmq[:password]
  }
end

• host: RabbitMQ server hostname
• port: RabbitMQ server port (default: 5672)
• user: Username for authentication
• password: Password for authentication
• vhost: Virtual host to connect to (default: "/")
• automatic_recovery: Enable automatic connection recovery
• network_recovery_interval: Time between recovery attempts

• type: Exchange type (default: "x-delayed-message")
• durable: If true, exchange will survive broker restarts
• auto_delete: If true, exchange will be deleted when last queue unbinds
• arguments: Additional exchange arguments

• durable: If true, queue will survive broker restarts
• exclusive: If true, queue can only be consumed by this connection
• auto_delete: If true, queue will be deleted when last consumer disconnects
• arguments: Additional queue arguments (e.g., TTL, dead-letter exchange)

1. Clone the repository
2. Install dependencies:

gem install bundler
bundle install

3. Run the test suite:

bundle exec rake test

4. Start the development console:

bundle exec irb -Ilib -rdelayed_rabbit

The unit tests use mocks and need no broker:

bundle exec rake test

The integration test publishes a delayed message to a real broker that has the delayed message plugin enabled:

RABBITMQ_INTEGRATION=1 bundle exec rake test

1. Update the version number in lib/delayed_rabbit/version.rb
2. Run the release command:

bundle exec rake release

This will:

• Create a git tag for the version
• Push git commits and tags
• Push the .gem file to rubygems.org

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -am 'Add some amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Create a Pull Request

1. Plugin Not Enabled (PRECONDITION_FAILED - invalid exchange type 'x-delayed-message')

   • Ensure the delayed message plugin is enabled in RabbitMQ
   • Run: rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2. PRECONDITION_FAILED - inequivalent arg ...

   • An exchange or queue with the same name already exists with different settings (versions before 1.1.0 declared delayed_jobs as a plain topic exchange)
   • Delete it, e.g. rabbitmqctl delete_exchange delayed_jobs / rabbitmqctl delete_queue delayed_jobs_queue

3. Connection Issues

   • Verify RabbitMQ server is running
   • Check connection credentials
   • Verify network connectivity

4. Message Not Received

   • Check if exchange exists
   • Verify queue bindings
   • Check message TTL settings

1. Enable Bunny logging:

Bunny.logger.level = Logger::DEBUG

2. Use RabbitMQ management UI to:
   • Monitor exchanges and queues
   • Check message delivery
   • View connection status

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

For support, please:

1. Check the documentation
2. Search existing issues
3. Open a new issue if needed
4. For urgent issues, consider professional support options

If you discover a security vulnerability, please contact the maintainers directly instead of opening a public issue. We will work to address the vulnerability as quickly as possible.