Skip to content
River Docs

Using River from Ruby

River’s Ruby client inserts and works jobs using Postgres or SQLite, with drivers for Rails’ Active Record and Sequel. It shares the River schema with Go, Rust, and TypeScript, so applications can enqueue jobs for workers in any of those languages.

The shared getting-started guide and feature pages include Ruby examples. Source is in river/ruby; the package is riverqueue.

Basic usage

Your project’s Gemfile should contain the riverqueue gem, a driver like riverqueue-activerecord, and the adapter for your database (see drivers). For example, with ActiveRecord and Postgres:

gem "pg"
gem "riverqueue"
gem "riverqueue-activerecord"

Initialize a client with:

require "riverqueue"
require "riverqueue-activerecord"


ActiveRecord::Base.establish_connection("postgres://...")
client = River::Client.new(River::Driver::ActiveRecord.new)

Define a job and insert it:

class SortArgs
  attr_accessor :strings


  def initialize(strings:)
    self.strings = strings
  end


  def kind = "sort"


  def to_json = JSON.generate(strings: strings)
end


insert_res = client.insert(SortArgs.new(strings: ["whale", "tiger", "bear"]))
insert_res.job # inserted job row

Job args should:

  • Respond to #kind with a unique string that identifies them in the database, which the receiving worker will recognize.
  • Respond to #to_json with a JSON object encoding.

They may also respond to #insert_opts with an instance of InsertOpts to define insertion options that’ll be used for all jobs of the kind.

Working jobs

Register a worker and start a client with a queue:

class SortWorker
  def self.kind = "sort"


  def work(job)
    puts job.args.fetch("strings").sort.inspect
  end
end


workers = River::Workers.new.add(SortWorker)


client = River::Client.new(River::Driver::ActiveRecord.new,
  config: River::Config.new(queues: {default: 10}, workers: workers))


client.start


# Later, during application shutdown:
client.stop

For a dedicated process, use the Ruby worker command, which boots the application and handles signals. See graceful shutdown.

Insertion options

Inserts take an insert_opts parameter to customize features of the inserted job:

insert_res = client.insert(
  SortArgs.new(strings: ["whale", "tiger", "bear"]),
  insert_opts: River::InsertOpts.new(
    max_attempts: 17,
    priority: 3,
    queue: "my_queue",
    tags: ["custom"]
  )
)

Inserting unique jobs

Unique jobs are supported through InsertOpts#unique_opts, and can be made unique by args, period, queue, and state. If a job matching unique properties is found on insert, the insert is skipped and the existing job returned.

insert_res = client.insert(args, insert_opts: River::InsertOpts.new(
  unique_opts: River::UniqueOpts.new(
    by_args: true,
    by_period: 15 * 60,
    by_queue: true
  )
))


# contains either a newly inserted job, or an existing one if insertion was skipped
insert_res.job


# true if insertion was skipped
insert_res.unique_skipped_as_duplicate?

Inserting jobs in bulk

Use #insert_many to bulk insert jobs as a single operation for improved efficiency:

results = client.insert_many([
  SortArgs.new(strings: ["whale", "tiger", "bear"]),
  SortArgs.new(strings: ["lion", "dolphin", "eagle"]),
])

Or with InsertManyParams, which may include insertion options:

results = client.insert_many([
  River::InsertManyParams.new(
    SortArgs.new(strings: ["whale", "tiger", "bear"]),
    insert_opts: River::InsertOpts.new(max_attempts: 5)
  ),
  River::InsertManyParams.new(
    SortArgs.new(strings: ["lion", "dolphin", "eagle"]),
    insert_opts: River::InsertOpts.new(queue: "high_priority")
  )
])

Inserting in a transaction

No extra code is needed to insert jobs from inside a transaction. Just make sure that one is open from your ORM of choice, call the normal #insert or #insert_many methods, and insertions will take part in it.

ActiveRecord::Base.transaction do
  client.insert(SortArgs.new(strings: ["whale", "tiger", "bear"]))
end
DB.transaction do
  client.insert(SortArgs.new(strings: ["whale", "tiger", "bear"]))
end

Inserting with a Ruby hash

JobArgsHash can be used to insert with a kind and JSON hash so that it’s not necessary to define a class:

insert_res = client.insert(River::JobArgsHash.new("hash_kind", {
    job_num: 1
}))

RBS and type checking

The gem bundles RBS files containing type annotations for its API to support type checking in Ruby through an RBS-compatible tool like Steep.

Drivers

Both drivers support insertion and execution. Create and update the schema with River’s migrations, and use the same connection pool for application transactions and River insertions.

ActiveRecord

Use River with Rails’ ActiveRecord by putting the riverqueue-activerecord driver in your Gemfile:

gem "riverqueue"
gem "riverqueue-activerecord"

Database adapters are optional dependencies. Add the adapter used by your application.

Postgres

Add pg to your Gemfile:

gem "pg"

Then initialize a client with ActiveRecord’s established connection:

ActiveRecord::Base.establish_connection("postgres://localhost/my_app")
client = River::Client.new(River::Driver::ActiveRecord.new)

SQLite

Add sqlite3 to your Gemfile:

gem "sqlite3"

Then initialize a client with an SQLite connection:

ActiveRecord::Base.establish_connection(
  adapter: "sqlite3",
  database: "storage/river.sqlite3",
  timeout: 5_000
)


client = River::Client.new(River::Driver::ActiveRecord.new)

Sequel

Use River with Sequel by putting the riverqueue-sequel driver in your Gemfile:

gem "riverqueue"
gem "riverqueue-sequel"

Database adapters are optional dependencies. Add the adapter used by your application.

Postgres

Add pg to your Gemfile:

gem "pg"

Then initialize a client with a Sequel database:

db = Sequel.connect("postgres://localhost/my_app")
client = River::Client.new(River::Driver::Sequel.new(db))

SQLite

Add sqlite3 to your Gemfile:

gem "sqlite3"

Then initialize a client with an SQLite database:

db = Sequel.connect("sqlite://storage/river.sqlite3", timeout: 5_000)
client = River::Client.new(River::Driver::Sequel.new(db))