kotovalexarian-likes-github/mperham--sidekiq
1
0
Fork 0
mirror of https://github.com/mperham/sidekiq.git synced 2022-11-09 13:52:34 -05:00
Code Releases Activity

Rejigger Sidekiq::Client to be instance-based, #1279

Browse source
This commit is contained in:
Mike Perham 2013-10-23 21:47:57 -07:00
parent 3ef6309b0d
commit b76ac9464c
5 changed files with 145 additions and 114 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Changes.md
View file

@ -1,6 +1,7 @@
2.16.0
-----------
- Refactor Sidekiq::Client to be instance-based [#1279]
- Pass all Redis options to the Redis driver so Unix sockets
can be fully configured. [#1270, salimane]
- Allow sidekiq-web extensions to add locale paths so extensions

4
lib/sidekiq.rb
View file

@ -77,8 +77,8 @@ module Sidekiq
end
end
def self.client_middleware
@client_chain ||= Client.default_middleware
def self.client_middleware(&block)
@client_chain ||= Middleware::Chain.new
yield @client_chain if block_given?
@client_chain
end

220
lib/sidekiq/client.rb
View file

@ -4,70 +4,102 @@ require 'sidekiq/middleware/chain'
module Sidekiq
class Client
class << self
def default_middleware
Middleware::Chain.new do
end
##
# 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.
#
def middleware(&block)
@chain ||= Sidekiq.client_middleware
if block_given?
@chain = @chain.dup
yield @chain
end
@chain
end
##
# 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:
# Sidekiq::Client.push('queue' => 'my_queue', 'class' => MyWorker, 'args' => ['foo', 1, :bat => 'bar'])
#
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 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 Client.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.
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
class << self
def default
@default ||= new
end
# deprecated
def registered_workers
puts "Deprecated, please use Sidekiq::Workers.new"
Sidekiq.redis { |x| x.smembers('workers') }
end
# deprecated
def registered_queues
puts "Deprecated, please use Sidekiq::Queue.all"
Sidekiq.redis { |x| x.smembers('queues') }
end
##
# 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:
# Sidekiq::Client.push('queue' => 'my_queue', 'class' => MyWorker, 'args' => ['foo', 1, :bat => 'bar'])
#
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
default.push(item)
end
##
# 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 Client.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.
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
default.push_bulk(items)
end
# Resque compatibility helpers. Note all helpers
@ -109,56 +141,56 @@ module Sidekiq
def enqueue_in(interval, klass, *args)
klass.perform_in(interval, *args)
end
end
private
private
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
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
pushed
end
def process_single(worker_class, item)
queue = item['queue']
Sidekiq.client_middleware.invoke(worker_class, item, queue) do
item
end
end
def normalize_item(item)
raise(ArgumentError, "Message must be a Hash of the form: { 'class' => SomeWorker, 'args' => ['bob', 1, :foo => 'bar'] }") unless item.is_a?(Hash)
raise(ArgumentError, "Message must include a class and set of arguments: #{item.inspect}") if !item['class'] || !item['args']
raise(ArgumentError, "Message args must be an Array") unless item['args'].is_a?(Array)
raise(ArgumentError, "Message class must be either a Class or String representation of the class name") unless item['class'].is_a?(Class) || item['class'].is_a?(String)
if item['class'].is_a?(Class)
raise(ArgumentError, "Message must include a Sidekiq::Worker class, not class name: #{item['class'].ancestors.inspect}") if !item['class'].respond_to?('get_sidekiq_options')
normalized_item = item['class'].get_sidekiq_options.merge(item)
normalized_item['class'] = normalized_item['class'].to_s
else
normalized_item = Sidekiq.default_worker_options.merge(item)
end
normalized_item['jid'] ||= SecureRandom.hex(12)
normalized_item['enqueued_at'] ||= Time.now.to_f
normalized_item
end
pushed
end
def process_single(worker_class, item)
queue = item['queue']
middleware.invoke(worker_class, item, queue) do
item
end
end
def normalize_item(item)
raise(ArgumentError, "Message must be a Hash of the form: { 'class' => SomeWorker, 'args' => ['bob', 1, :foo => 'bar'] }") unless item.is_a?(Hash)
raise(ArgumentError, "Message must include a class and set of arguments: #{item.inspect}") if !item['class'] || !item['args']
raise(ArgumentError, "Message args must be an Array") unless item['args'].is_a?(Array)
raise(ArgumentError, "Message class must be either a Class or String representation of the class name") unless item['class'].is_a?(Class) || item['class'].is_a?(String)
if item['class'].is_a?(Class)
raise(ArgumentError, "Message must include a Sidekiq::Worker class, not class name: #{item['class'].ancestors.inspect}") if !item['class'].respond_to?('get_sidekiq_options')
normalized_item = item['class'].get_sidekiq_options.merge(item)
normalized_item['class'] = normalized_item['class'].to_s
else
normalized_item = Sidekiq.default_worker_options.merge(item)
end
normalized_item['jid'] ||= SecureRandom.hex(12)
normalized_item['enqueued_at'] ||= Time.now.to_f
normalized_item
end
end
end

30
lib/sidekiq/testing.rb
View file

@ -54,24 +54,22 @@ module Sidekiq
class EmptyQueueError < RuntimeError; end
class Client
class << self
alias_method :raw_push_real, :raw_push
alias_method :raw_push_real, :raw_push
def raw_push(payloads)
if Sidekiq::Testing.fake?
payloads.each do |job|
job['class'].constantize.jobs << Sidekiq.load_json(Sidekiq.dump_json(job))
end
true
elsif Sidekiq::Testing.inline?
payloads.each do |item|
marshalled = Sidekiq.load_json(Sidekiq.dump_json(item))
marshalled['class'].constantize.new.perform(*marshalled['args'])
end
true
else
raw_push_real(payloads)
def raw_push(payloads)
if Sidekiq::Testing.fake?
payloads.each do |job|
job['class'].constantize.jobs << Sidekiq.load_json(Sidekiq.dump_json(job))
end
true
elsif Sidekiq::Testing.inline?
payloads.each do |item|
marshalled = Sidekiq.load_json(Sidekiq.dump_json(item))
marshalled['class'].constantize.new.perform(*marshalled['args'])
end
true
else
raw_push_real(payloads)
end
end
end

4
test/test_client.rb
View file

@ -201,11 +201,11 @@ class TestClient < Sidekiq::Test
describe 'item normalization' do
it 'defaults retry to true' do
assert_equal true, Sidekiq::Client.send(:normalize_item, 'class' => QueuedWorker, 'args' => [])['retry']
assert_equal true, Sidekiq::Client.new.send(:normalize_item, 'class' => QueuedWorker, 'args' => [])['retry']
end
it "does not normalize numeric retry's" do
assert_equal 2, Sidekiq::Client.send(:normalize_item, 'class' => CWorker, 'args' => [])['retry']
assert_equal 2, Sidekiq::Client.new.send(:normalize_item, 'class' => CWorker, 'args' => [])['retry']
end
end
end