An insert-only Ruby client for River packaged in the riverqueue gem. Allows jobs to be inserted in Ruby and run by a Go worker, but doesn't support working jobs in Ruby.
Your project's Gemfile should contain the riverqueue gem and a driver like riverqueue-sequel (see drivers):
gem"riverqueue"gem"riverqueue-sequel"Initialize a client with:
require"riverqueue"require"riverqueue-activerecord"client=River::Client.new(River::Driver::ActiveRecord.new)Define a job and insert it:
classSortArgsattr_accessor:stringsdefinitialize(strings:)self.strings=stringsenddefkind="sort"defto_json=JSON.dump({strings: strings})endinsert_res=client.insert(SortArgs.new(strings: ["whale","tiger","bear"]))insert_res.job# inserted job rowJob args should:
- Respond to
#kindwith a unique string that identifies them in the database, and which a Go worker will recognize. - Response to
#to_jsonwith 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.
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"]))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 skippedinsert_res.job# true if insertion was skippedinsert_res.unique_skipped_as_duplicatedUse #insert_many to bulk insert jobs as a single operation for improved efficiency:
num_inserted=client.insert_many([SortArgs.new(strings: ["whale","tiger","bear"]),SortArgs.new(strings: ["lion","dolphin","eagle"])])Or with InsertManyParams, which may include insertion options:
num_inserted=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"))])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.transactiondoclient.insert(SortArgs.new(strings: ["whale","tiger","bear"]))endDB.transactiondoclient.insert(SortArgs.new(strings: ["whale","tiger","bear"]))endJobArgsHash 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}))The gem bundles RBS files containing type annotations for its API to support type checking in Ruby through a tool like Sorbet or Steep.
Use River with ActiveRecord by putting the riverqueue-activerecord driver in your Gemfile:
gem"riverqueue"gem"riverqueue-activerecord"Then initialize driver and client:
ActiveRecord::Base.establish_connection("postgres://...")client=River::Client.new(River::Driver::ActiveRecord.new)Use River with Sequel by putting the riverqueue-sequel driver in your Gemfile:
gem"riverqueue"gem"riverqueue-sequel"Then initialize driver and client:
DB=Sequel.connect("postgres://...")client=River::Client.new(River::Driver::Sequel.new(DB))See development.