Class: Sidekiq::Client
- Inherits:
-
Object
- Object
- Sidekiq::Client
- Includes:
- JobUtil, TestingClient
- Defined in:
- lib/sidekiq/client.rb
Constant Summary
Constants included from JobUtil
Instance Attribute Summary collapse
-
#redis_pool ⇒ Object
Returns the value of attribute redis_pool.
Class Method Summary collapse
-
.enqueue(klass, *args) ⇒ Object
Resque compatibility helpers.
-
.enqueue_in(interval, klass, *args) ⇒ Object
Example usage: Sidekiq::Client.enqueue_in(3.minutes, MyJob, ‘foo’, 1, :bat => ‘bar’).
-
.enqueue_to(queue, klass, *args) ⇒ Object
Example usage: Sidekiq::Client.enqueue_to(:queue_name, MyJob, ‘foo’, 1, :bat => ‘bar’).
-
.enqueue_to_in(queue, interval, klass, *args) ⇒ Object
Example usage: Sidekiq::Client.enqueue_to_in(:queue_name, 3.minutes, MyJob, ‘foo’, 1, :bat => ‘bar’).
- .push(item) ⇒ Object
- .push_bulk ⇒ Object
-
.via(pool) ⇒ Object
Allows sharding of jobs across any number of Redis instances.
Instance Method Summary collapse
-
#initialize(*args, **kwargs) ⇒ Client
constructor
Sidekiq::Client is responsible for pushing job payloads to Redis.
-
#middleware(&block) ⇒ Object
Define client-side middleware:.
-
#push(item) ⇒ Object
The main method used to push a job to Redis.
-
#push_bulk(items) ⇒ Object
Push a large number of jobs to Redis.
Methods included from JobUtil
#normalize_item, #normalized_hash, #validate, #verify_json
Constructor Details
#initialize(*args, **kwargs) ⇒ Client
45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
# File 'lib/sidekiq/client.rb', line 45 def initialize(*args, **kwargs) if args.size == 1 && kwargs.size == 0 warn "Sidekiq::Client.new(pool) is deprecated, please use Sidekiq::Client.new(pool: pool), #{caller(0..3)}" # old calling method, accept 1 pool argument @redis_pool = args[0] @chain = Sidekiq.default_configuration.client_middleware @config = Sidekiq.default_configuration else # new calling method: keyword arguments @config = kwargs[:config] || Sidekiq.default_configuration @redis_pool = kwargs[:pool] || Thread.current[:sidekiq_redis_pool] || @config&.redis_pool @chain = kwargs[:chain] || @config&.client_middleware raise ArgumentError, "No Redis pool available for Sidekiq::Client" unless @redis_pool end end |
Instance Attribute Details
#redis_pool ⇒ Object
Returns the value of attribute redis_pool.
31 32 33 |
# File 'lib/sidekiq/client.rb', line 31 def redis_pool @redis_pool end |
Class Method Details
.enqueue(klass, *args) ⇒ Object
189 190 191 |
# File 'lib/sidekiq/client.rb', line 189 def enqueue(klass, *args) klass.client_push("class" => klass, "args" => args) end |
.enqueue_in(interval, klass, *args) ⇒ Object
217 218 219 |
# File 'lib/sidekiq/client.rb', line 217 def enqueue_in(interval, klass, *args) klass.perform_in(interval, *args) end |
.enqueue_to(queue, klass, *args) ⇒ Object
196 197 198 |
# File 'lib/sidekiq/client.rb', line 196 def enqueue_to(queue, klass, *args) klass.client_push("queue" => queue, "class" => klass, "args" => args) end |
.enqueue_to_in(queue, interval, klass, *args) ⇒ Object
203 204 205 206 207 208 209 210 211 212 |
# File 'lib/sidekiq/client.rb', line 203 def enqueue_to_in(queue, interval, klass, *args) int = interval.to_f now = Time.now.to_f ts = ((int < 1_000_000_000) ? now + int : int) item = {"class" => klass, "args" => args, "at" => ts, "queue" => queue} item.delete("at") if ts <= now klass.client_push(item) end |
.push(item) ⇒ Object
173 174 175 |
# File 'lib/sidekiq/client.rb', line 173 def push(item) new.push(item) end |
.push_bulk ⇒ Object
177 178 179 |
# File 'lib/sidekiq/client.rb', line 177 def push_bulk(...) new.push_bulk(...) end |
.via(pool) ⇒ Object
Allows sharding of jobs across any number of Redis instances. All jobs defined within the block will use the given Redis connection pool.
pool = ConnectionPool.new { Redis.new }
Sidekiq::Client.via(pool) do
SomeJob.perform_async(1,2,3)
SomeOtherJob.perform_async(1,2,3)
end
Generally this is only needed for very large Sidekiq installs processing thousands of jobs per second. I do not recommend sharding unless you cannot scale any other way (e.g. splitting your app into smaller apps).
163 164 165 166 167 168 169 170 |
# File 'lib/sidekiq/client.rb', line 163 def self.via(pool) raise ArgumentError, "No pool given" if pool.nil? current_sidekiq_pool = Thread.current[:sidekiq_redis_pool] Thread.current[:sidekiq_redis_pool] = pool yield ensure Thread.current[:sidekiq_redis_pool] = current_sidekiq_pool end |
Instance Method Details
#middleware(&block) ⇒ Object
23 24 25 26 27 28 29 |
# File 'lib/sidekiq/client.rb', line 23 def middleware(&block) if block @chain = @chain.dup yield @chain end @chain end |
#push(item) ⇒ Object
The main method used to push a job to Redis. Accepts a number of options:
queue - the named queue to use, default 'default'
class - the job class to call, required
args - an array of simple arguments to the perform method, must be JSON-serializable
at - timestamp to schedule the job (optional), must be Numeric (e.g. Time.now.to_f)
retry - whether to retry this job if it fails, default true or an integer number of retries
retry_for - relative amount of time to retry this job if it fails, default nil
backtrace - whether to save any error backtrace, default false
If class is set to the class name, the jobs’ options will be based on Sidekiq’s default job options. Otherwise, they will be based on the job class’s options.
Any options valid for a job class’s sidekiq_options are also available here.
All keys must be strings, not symbols. NB: because we are serializing to JSON, all symbols in ‘args’ will be converted to strings. Note that backtrace: true can take quite a bit of space in Redis; a large volume of failing jobs can start Redis swapping if you aren’t careful.
Returns a unique Job ID. If middleware stops the job, nil will be returned instead.
Example:
push('queue' => 'my_queue', 'class' => MyJob, 'args' => ['foo', 1, :bat => 'bar'])
86 87 88 89 90 91 92 93 94 95 96 |
# File 'lib/sidekiq/client.rb', line 86 def push(item) normed = normalize_item(item) payload = middleware.invoke(item["class"], normed, normed["queue"], @redis_pool) do normed end if payload verify_json(payload) raw_push([payload]) payload["jid"] end end |
#push_bulk(items) ⇒ Object
Push a large number of jobs to Redis. This method cuts out the redis network round trip latency. It pushes jobs in batches if more than ‘:batch_size` (1000 by default) of jobs are passed. I wouldn’t recommend making ‘:batch_size` larger than 1000 but YMMV based on network quality, size of job args, etc. A large number of jobs can cause a bit of Redis command processing latency.
Takes the same arguments as #push except that args is expected to be an Array of Arrays. All other keys are duplicated for each job. Each job is run through the client middleware pipeline and each job gets its own Job ID as normal.
Returns an array of the of pushed jobs’ jids, may contain nils if any client middleware prevented a job push.
Example (pushing jobs in batches):
push_bulk('class' => MyJob, 'args' => (1..100_000).to_a, batch_size: 1_000)
116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 |
# File 'lib/sidekiq/client.rb', line 116 def push_bulk(items) batch_size = items.delete(:batch_size) || items.delete("batch_size") || 1_000 args = items["args"] at = items.delete("at") raise ArgumentError, "Job 'at' must be a Numeric or an Array of Numeric timestamps" if at && (Array(at).empty? || !Array(at).all? { |entry| entry.is_a?(Numeric) }) raise ArgumentError, "Job 'at' Array must have same size as 'args' Array" if at.is_a?(Array) && at.size != args.size jid = items.delete("jid") raise ArgumentError, "Explicitly passing 'jid' when pushing more than one job is not supported" if jid && args.size > 1 normed = normalize_item(items) slice_index = 0 result = args.each_slice(batch_size).flat_map do |slice| raise ArgumentError, "Bulk arguments must be an Array of Arrays: [[1], [2]]" unless slice.is_a?(Array) && slice.all?(Array) break [] if slice.empty? # no jobs to push payloads = slice.map.with_index { |job_args, index| copy = normed.merge("args" => job_args, "jid" => SecureRandom.hex(12)) copy["at"] = (at.is_a?(Array) ? at[slice_index + index] : at) if at result = middleware.invoke(items["class"], copy, copy["queue"], @redis_pool) do verify_json(copy) copy end result || nil } slice_index += batch_size to_push = payloads.compact raw_push(to_push) unless to_push.empty? payloads.map { |payload| payload&.[]("jid") } end result.is_a?(Enumerator::Lazy) ? result.force : result end |