Rrq

Round-robin Queue, with focus on reliability.

Installation

Add this line to your application's Gemfile:

gem 'rrq'

And then execute:

$ bundle

Usage

$rrq = Rrq::Connection.new
$rrq.redis = $redis

Rrq::Manager.new($rrq).recover! # Run this on startup for one process

1. Create a message payload as a hash

payload = {
  klass: "MessageProcessor",
  args: [1, 2 ,3]
}

2. Write the message to Rrq

$rrq.push("messages", company.id, payload)

3. Pull a message from the queue, perform processing than acknowledge

loop do 
  message = $rrq.pop
  payload = message["payload"]

  worker = payload["klass"].constantize.new
  worker.perform(*payload["args"])

  $rrq.ack(message)
end

Message structure:

{
  message_id: "d41d8cd98f00b204e9800998ecf8427e",
  queue: "queue_name",
  partition: "123",
  payload: {
    key: "value"
  }
}

Purpose of keys:

• rrq:queues (Set) => Contains a list of all queues
• rrq:queue:<queue name>:all_partitions (Set) => Contains a list of partitions for a particular queue
• rrq:queue:<queue name>:active_partitions (List) => A copy of the all_partitions set but to be used for round-robin distribution of partitions.
• rrq:queue:<queue name>:<partition> (List) => A list of messages for a particular queue + partition
• rrq:queue:<queue name>:<partition>:working (List) => A list of unaknowledged messages for a particular queue + partition

Contributing

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