Skip to content
River Docs

Inserting jobs from Ruby

River supports inserting jobs from Ruby and have them worked in Go, a feature that may be desirable in performance sensitive cases so that jobs can take advantage of Go's considerably faster runtime speed.

Insertion is supported for both Postgres and SQLite through Rails' ActiveRecord and Sequel.

Client source code is available in the riverqueue-ruby GitHub repository.

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.dump({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, and which a Go worker will recognize.
  • Response to #to_json with a JSON serialization that'll be parseable as an object in Go.

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.

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,
    by_state: [River::JOB_STATE_AVAILABLE]
  )
)


# 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_duplicated

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 a tool like Sorbet or Steep.

Drivers

The Ruby client only inserts jobs. Configure a Go worker with River's SQLite driver and point it at the same database to work jobs inserted through SQLite. Create and update the database using River's SQLite migrations.

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))