Class: Sidekiq::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq/client.rb,
lib/sidekiq/testing.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.default ⇒ Object 



80
81
82
 
# File 'lib/sidekiq/client.rb', line 80

def default
  @default ||= new
end

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



112
113
114
 
# File 'lib/sidekiq/client.rb', line 112

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



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

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



119
120
121
 
# File 'lib/sidekiq/client.rb', line 119

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



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

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 



96
97
98
 
# File 'lib/sidekiq/client.rb', line 96

def push(item)
  default.push(item)
end

.push_bulk(items) ⇒ Object 



100
101
102
 
# File 'lib/sidekiq/client.rb', line 100

def push_bulk(items)
  default.push_bulk(items)
end

.registered_queues ⇒ Object

deprecated



91
92
93
94
 
# File 'lib/sidekiq/client.rb', line 91

def registered_queues
  puts "registered_queues is deprecated, please use Sidekiq::Queue.all"
  Sidekiq::Queue.all.map(&:name)
end

.registered_workers ⇒ Object

deprecated



85
86
87
88
 
# File 'lib/sidekiq/client.rb', line 85

def registered_workers
  puts "registered_workers is deprecated, please use Sidekiq::Workers.new"
  Sidekiq.redis { |x| x.smembers('workers') }
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.



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

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
retry - whether to retry this job if it fails, true or false, default true
backtrace - whether to save any error backtrace, default false

All options must be strings, not symbols. NB: because we are serializing to JSON, all symbols in ‘args’ will be converted to strings.

Returns nil if not pushed to Redis or a unique Job ID if pushed.

Example:

push('queue' => 'my_queue', 'class' => MyWorker, 'args' => ['foo', 1, :bat => 'bar'])



45
46
47
48
49
50
51
52
 
# File 'lib/sidekiq/client.rb', line 45

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

  pushed = false
  pushed = raw_push([payload]) if payload
  pushed ? payload['jid'] : nil
end

#push_bulk(items) ⇒ Object

Push a large number of jobs to Redis. In practice this method is only useful if you are pushing tens of thousands of jobs or more. This method basically cuts down on the redis round trip 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 the number of jobs pushed or nil if the pushed failed. The number of jobs pushed can be less than the number given if the middleware stopped processing for one or more jobs.



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

def push_bulk(items)
  normed = normalize_item(items)
  payloads = items['args'].map do |args|
    raise ArgumentError, "Bulk arguments must be an Array of Arrays: [[1], [2]]" if !args.is_a?(Array)
    process_single(items['class'], normed.merge('args' => args, 'jid' => SecureRandom.hex(12), 'enqueued_at' => Time.now.to_f))
  end.compact

  pushed = false
  pushed = raw_push(payloads) if !payloads.empty?
  pushed ? payloads.size : nil
end

#raw_push(payloads) ⇒ Object 



147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/sidekiq/client.rb', line 147

def raw_push(payloads)
  pushed = false
  Sidekiq.redis do |conn|
    if payloads.first['at']
      pushed = conn.zadd('schedule', payloads.map do |hash|
        at = hash.delete('at').to_s
        [at, Sidekiq.dump_json(hash)]
      end)
    else
      q = payloads.first['queue']
      to_push = payloads.map { |entry| Sidekiq.dump_json(entry) }
      _, pushed = conn.multi do
        conn.sadd('queues', q)
        conn.lpush("queue:#{q}", to_push)
      end
    end
  end
  pushed
end

#raw_push_real ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/sidekiq/testing.rb', line 57

def raw_push(payloads)
  pushed = false
  Sidekiq.redis do |conn|
    if payloads.first['at']
      pushed = conn.zadd('schedule', payloads.map do |hash|
        at = hash.delete('at').to_s
        [at, Sidekiq.dump_json(hash)]
      end)
    else
      q = payloads.first['queue']
      to_push = payloads.map { |entry| Sidekiq.dump_json(entry) }
      _, pushed = conn.multi do
        conn.sadd('queues', q)
        conn.lpush("queue:#{q}", to_push)
      end
    end
  end
  pushed
end