Class: Sidekiq::Client

Inherits:
Object
  • Object
show all
Includes:
TestingClient
Defined in:
lib/sidekiq/client.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis_pool = nil) ⇒ Client

Sidekiq::Client normally uses the default Redis pool but you may pass a custom ConnectionPool if you want to shard your Sidekiq jobs across several Redis instances (for scalability reasons, e.g.)

Sidekiq::Client.new(ConnectionPool.new { Redis.new })

Generally this is only needed for very large Sidekiq installs processing thousands of jobs per second. I don’t recommend sharding unless you cannot scale any other way (e.g. splitting your app into smaller apps).



41
42
43
 
# File 'lib/sidekiq/client.rb', line 41

def initialize(redis_pool = nil)
  @redis_pool = redis_pool || Thread.current[:sidekiq_via_pool] || Sidekiq.redis_pool
end

Instance Attribute Details

#redis_pool ⇒ Object

Returns the value of attribute redis_pool.



29
30
31
 
# File 'lib/sidekiq/client.rb', line 29

def redis_pool
  @redis_pool
end

Class Method Details

.enqueue(klass, *args) ⇒ Object

Resque compatibility helpers. Note all helpers should go through Worker#client_push.

Example usage:

Sidekiq::Client.enqueue(MyWorker, 'foo', 1, :bat => 'bar')

Messages are enqueued to the ‘default’ queue.



151
152
153
 
# File 'lib/sidekiq/client.rb', line 151

def enqueue(klass, *args)
  klass.client_push("class" => klass, "args" => args)
end

.enqueue_in(interval, klass, *args) ⇒ Object

Example usage:

Sidekiq::Client.enqueue_in(3.minutes, MyWorker, 'foo', 1, :bat => 'bar')



179
180
181
 
# File 'lib/sidekiq/client.rb', line 179

def enqueue_in(interval, klass, *args)
  klass.perform_in(interval, *args)
end

.enqueue_to(queue, klass, *args) ⇒ Object

Example usage:

Sidekiq::Client.enqueue_to(:queue_name, MyWorker, 'foo', 1, :bat => 'bar')



158
159
160
 
# File 'lib/sidekiq/client.rb', line 158

def enqueue_to(queue, klass, *args)
  klass.client_push("queue" => queue, "class" => klass, "args" => args)
end

.enqueue_to_in(queue, interval, klass, *args) ⇒ Object

Example usage:

Sidekiq::Client.enqueue_to_in(:queue_name, 3.minutes, MyWorker, 'foo', 1, :bat => 'bar')



165
166
167
168
169
170
171
172
173
174
 
# File 'lib/sidekiq/client.rb', line 165

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 



135
136
137
 
# File 'lib/sidekiq/client.rb', line 135

def push(item)
  new.push(item)
end

.push_bulk(items) ⇒ Object 



139
140
141
 
# File 'lib/sidekiq/client.rb', line 139

def push_bulk(items)
  new.push_bulk(items)
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
  SomeWorker.perform_async(1,2,3)
  SomeOtherWorker.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).



125
126
127
128
129
130
131
132
 
# File 'lib/sidekiq/client.rb', line 125

def self.via(pool)
  raise ArgumentError, "No pool given" if pool.nil?
  current_sidekiq_pool = Thread.current[:sidekiq_via_pool]
  Thread.current[:sidekiq_via_pool] = pool
  yield
ensure
  Thread.current[:sidekiq_via_pool] = current_sidekiq_pool
end

Instance Method Details

#middleware(&block) ⇒ Object

Define client-side middleware:

client = Sidekiq::Client.new
client.middleware do |chain|
  chain.use MyClientMiddleware
end
client.push('class' => 'SomeWorker', 'args' => [1,2,3])

All client instances default to the globally-defined Sidekiq.client_middleware but you can change as necessary.



20
21
22
23
24
25
26
27
 
# File 'lib/sidekiq/client.rb', line 20

def middleware(&block)
  @chain ||= Sidekiq.client_middleware
  if block_given?
    @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 worker 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
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 worker options. Otherwise, they will be based on the job class’s options.

Any options valid for a worker class’s sidekiq_options are also available here.

All options 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' => MyWorker, 'args' => ['foo', 1, :bat => 'bar'])



69
70
71
72
73
74
75
76
77
 
# File 'lib/sidekiq/client.rb', line 69

def push(item)
  normed = normalize_item(item)
  payload = process_single(item["class"], normed)

  if 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. I wouldn’t recommend pushing more than 1000 per call 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. The number of jobs pushed can be less than the number given if the middleware stopped processing for one or more jobs.

Raises:

  • (ArgumentError) 


92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/sidekiq/client.rb', line 92

def push_bulk(items)
  arg = items["args"].first
  return [] unless arg # no jobs to push
  raise ArgumentError, "Bulk arguments must be an Array of Arrays: [[1], [2]]" unless arg.is_a?(Array)

  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?(Numeric))

  normed = normalize_item(items)
  payloads = items["args"].map.with_index { |args, index|
    copy = normed.merge("args" => args, "jid" => SecureRandom.hex(12), "enqueued_at" => Time.now.to_f)
    copy["at"] = (at.is_a?(Array) ? at[index] : at) if at

    result = process_single(items["class"], copy)
    result || nil
  }.compact

  raw_push(payloads) unless payloads.empty?
  payloads.collect { |payload| payload["jid"] }
end