ChainedJob
The library provides ability to use chained background jobs.
Installation
Add this line to your application's Gemfile:
gem 'chained_job'And then execute:
$ bundle install
Or install the gem yourself:
$ gem install chained_job
How it works
chained_job stores background job arguments array in Redis and starts processing them in parallel by number of workers provided.
Job re-enqueues itself until there are no more stored arguments in Redis for that specific job.
Configuration
require 'logger'
require 'redis_config'
ChainedJob.configure do |config|
# Default value is 1_000
config.arguments_batch_size = 2_000
# Default value is 7 days
config.arguments_queue_expiration = 3 * 24 * 60 * 60 # 3 days
# Error will be raised while running job if redis is not setup
config.redis = Svc.redis # redis-client
config.logger = ::Logger.new(STDOUT)
endCustom Hooks
There are three types of hooks: around_start_chains, around_chain_process and around_array_of_job_arguments as lambda functions. Both of the hooks gets options hash.
For around_start_chains callback options hash contains four keys:
{
job_class: CheckUsersActivityJob,
array_of_job_arguments: [1, 2 ,3],
parallelism: 2,
args: [{"job_argument" => "value"}],
}For around_chain_process:
{
job_class: CheckUsersActivityJob,
worker_id: 2,
args: [{"job_argument" => "value"}],
}For around_array_of_job_arguments:
{
job_class: CheckUsersActivityJob,
args: [{"job_argument" => "value"}],
}You can configure callbacks:
ChainedJob.configure do |config|
config.around_start_chains = ->(options, &block) do
time = Benchmark.ms { block.call }
puts "It took #{time} to start the chains for #{options[:job_class]} job"
end
endUsage
Example of creating new background ActiveJob:
# frozen_string_literal: true
class CheckUsersActivityJob < ActiveJob::Base
include ChainedJob::Middleware
def parallelism
2
end
def array_of_job_arguments
User.last(100).ids
end
def process(user_id)
user = User.find_by(id: user_id)
return unless user
UserActivities::Check.run(user)
end
endOn Failure (retry queue)
In this case, if one of your chained workers fails to process some ids - it will go into the retry queue and restarts as you would expect. However important to note that args picked from Redis are no longer available, and hence those ids won't be processed anymore.
In case you want that during exception your arguments would be pushed back to Redis, you can use
# frozen_string_literal: true
class CheckUsersActivityJob < ActiveJob::Base
include ChainedJob::Middleware
def handle_retry?
true
end
endDevelopment
For running tests use bundle exec rake test.
To check linter errors use rubocop: bundle exec rubocop
Contributing
Bug reports and pull requests are welcome on GitHub at https://github.com/vinted/chained_job.