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

Refactor to use Sidekiq.dump_json and Sidekiq.load_json

These methods perform MultiJson feature detection and can be removed
after this library's MultiJson dependency is upgraded to ~> 2.0.
This commit is contained in:
Erik Michaels-Ober 2012-04-22 14:02:35 -07:00
parent 68c725e47b
commit 5eb8d397f0
11 changed files with 32 additions and 72 deletions

View file

@ -82,4 +82,22 @@ module Sidekiq
@server_chain @server_chain
end end
def self.load_json(string, options={})
# Can't reliably detect whether MultiJson responds to load, since it's
# a reserved word. Use adapter as a proxy for new features.
if MultiJson.respond_to?(:adapter)
MultiJson.load(string, options)
else
MultiJson.decode(string, options)
end
end
def self.dump_json(object, options={})
if MultiJson.respond_to?(:dump)
MultiJson.dump(object, options)
else
MultiJson.encode(object, options)
end
end
end end

View file

@ -45,11 +45,7 @@ module Sidekiq
pushed = false pushed = false
Sidekiq.client_middleware.invoke(worker_class, item, queue) do Sidekiq.client_middleware.invoke(worker_class, item, queue) do
payload = if MultiJson.respond_to?(:dump) payload = Sidekiq.dump_json(item)
MultiJson.dump(item)
else
MultiJson.encode(item)
end
Sidekiq.redis do |conn| Sidekiq.redis do |conn|
_, pushed = conn.multi do _, pushed = conn.multi do
conn.sadd('queues', queue) conn.sadd('queues', queue)

View file

@ -110,11 +110,7 @@ module Sidekiq
processor = @ready.pop processor = @ready.pop
@in_progress[processor.object_id] = [msg, queue] @in_progress[processor.object_id] = [msg, queue]
@busy << processor @busy << processor
if MultiJson.respond_to?(:adapter) processor.process!(Sidekiq.load_json(msg), queue)
processor.process!(MultiJson.load(msg), queue)
else
processor.process!(MultiJson.decode(msg), queue)
end
end end
end end
end end

View file

@ -10,11 +10,7 @@ module Sidekiq
def call(worker_class, item, queue) def call(worker_class, item, queue)
enabled = worker_class.get_sidekiq_options['unique'] enabled = worker_class.get_sidekiq_options['unique']
if enabled if enabled
payload_hash = if MultiJson.respond_to?(:dump) payload_hash = Digest::MD5.hexdigest(Sidekiq.dump_json(item))
Digest::MD5.hexdigest(MultiJson.dump(item))
else
Digest::MD5.hexdigest(MultiJson.encode(item))
end
unique = false unique = false
Sidekiq.redis do |conn| Sidekiq.redis do |conn|

View file

@ -16,11 +16,7 @@ module Sidekiq
:worker => args[1]['class'], :worker => args[1]['class'],
:queue => args[2] :queue => args[2]
} }
if MultiJson.respond_to?(:dump) Sidekiq.redis {|conn| conn.rpush(:failed, Sidekiq.dump_json(data)) }
Sidekiq.redis {|conn| conn.rpush(:failed, MultiJson.dump(data)) }
else
Sidekiq.redis {|conn| conn.rpush(:failed, MultiJson.encode(data)) }
end
raise raise
end end
end end

View file

@ -46,11 +46,7 @@ module Sidekiq
delay = DELAY.call(count) delay = DELAY.call(count)
logger.debug { "Failure! Retry #{count} in #{delay} seconds" } logger.debug { "Failure! Retry #{count} in #{delay} seconds" }
retry_at = Time.now.to_f + delay retry_at = Time.now.to_f + delay
payload = if MultiJson.respond_to?(:dump) payload = Sidekiq.dump_json(msg)
MultiJson.dump(msg)
else
MultiJson.encode(msg)
end
Sidekiq.redis do |conn| Sidekiq.redis do |conn|
conn.zadd('retry', retry_at.to_s, payload) conn.zadd('retry', retry_at.to_s, payload)
end end

View file

@ -7,11 +7,7 @@ module Sidekiq
def call(*args) def call(*args)
yield yield
ensure ensure
json = if MultiJson.respond_to?(:dump) json = Sidekiq.dump_json(args[1])
MultiJson.dump(args[1])
else
MultiJson.encode(args[1])
end
hash = Digest::MD5.hexdigest(json) hash = Digest::MD5.hexdigest(json)
Sidekiq.redis {|conn| conn.del(hash) } Sidekiq.redis {|conn| conn.del(hash) }
end end

View file

@ -55,11 +55,7 @@ module Sidekiq
conn.multi do conn.multi do
conn.set("worker:#{self}:started", Time.now.to_s) conn.set("worker:#{self}:started", Time.now.to_s)
hash = {:queue => queue, :payload => msg, :run_at => Time.now.strftime("%Y/%m/%d %H:%M:%S %Z")} hash = {:queue => queue, :payload => msg, :run_at => Time.now.strftime("%Y/%m/%d %H:%M:%S %Z")}
if MultiJson.respond_to?(:dump) conn.set("worker:#{self}", Sidekiq.dump_json(hash))
conn.set("worker:#{self}", MultiJson.dump(hash))
else
conn.set("worker:#{self}", MultiJson.encode(hash))
end
end end
end end

View file

@ -46,11 +46,7 @@ module Sidekiq
messages.each do |message| messages.each do |message|
logger.debug { "Retrying #{message}" } logger.debug { "Retrying #{message}" }
msg = if MultiJson.respond_to?(:adapter) msg = Sidekiq.load_json(message)
MultiJson.load(message)
else
MultiJson.decode(message)
end
conn.rpush("queue:#{msg['queue']}", message) conn.rpush("queue:#{msg['queue']}", message)
end end
end end

View file

@ -52,13 +52,7 @@ module Sidekiq
Sidekiq.redis do |conn| Sidekiq.redis do |conn|
conn.smembers('workers').map do |w| conn.smembers('workers').map do |w|
msg = conn.get("worker:#{w}") msg = conn.get("worker:#{w}")
if msg msg = Sidekiq.load_json(msg) if msg
msg = if MultiJson.respond_to?(:adapter)
MultiJson.load(msg)
else
MultiJson.decode(msg)
end
end
[w, msg] [w, msg]
end.sort { |x| x[1] ? -1 : 1 } end.sort { |x| x[1] ? -1 : 1 }
end end
@ -80,11 +74,7 @@ module Sidekiq
def retries def retries
Sidekiq.redis do |conn| Sidekiq.redis do |conn|
results = conn.zrange('retry', 0, 25, :withscores => true) results = conn.zrange('retry', 0, 25, :withscores => true)
if MultiJson.respond_to?(:adapter) results.each_slice(2).map { |msg, score| [Sidekiq.load_json(msg), Float(score)] }
results.each_slice(2).map { |msg, score| [MultiJson.load(msg), Float(score)] }
else
results.each_slice(2).map { |msg, score| [MultiJson.decode(msg), Float(score)] }
end
end end
end end
@ -99,11 +89,7 @@ module Sidekiq
def retries_with_score(score) def retries_with_score(score)
Sidekiq.redis do |conn| Sidekiq.redis do |conn|
results = conn.zrangebyscore('retry', score, score) results = conn.zrangebyscore('retry', score, score)
if MultiJson.respond_to?(:adapter) results.map { |msg| Sidekiq.load_json(msg) }
results.map { |msg| MultiJson.load(msg) }
else
results.map { |msg| MultiJson.decode(msg) }
end
end end
end end
@ -138,11 +124,7 @@ module Sidekiq
get "/queues/:name" do get "/queues/:name" do
halt 404 unless params[:name] halt 404 unless params[:name]
@name = params[:name] @name = params[:name]
if MultiJson.respond_to?(:adapter) @messages = Sidekiq.redis {|conn| conn.lrange("queue:#{@name}", 0, 10) }.map { |str| Sidekiq.load_json(str) }
@messages = Sidekiq.redis {|conn| conn.lrange("queue:#{@name}", 0, 10) }.map { |str| MultiJson.load(str) }
else
@messages = Sidekiq.redis {|conn| conn.lrange("queue:#{@name}", 0, 10) }.map { |str| MultiJson.decode(str) }
end
slim :queue slim :queue
end end
@ -160,11 +142,7 @@ module Sidekiq
results = conn.zrangebyscore('retry', score, score) results = conn.zrangebyscore('retry', score, score)
conn.zremrangebyscore('retry', score, score) conn.zremrangebyscore('retry', score, score)
results.map do |message| results.map do |message|
msg = if MultiJson.respond_to?(:adapter) msg = Sidekiq.load_json(message)
MultiJson.load(message)
else
MultiJson.decode(message)
end
conn.rpush("queue:#{msg['queue']}", message) conn.rpush("queue:#{msg['queue']}", message)
end end
end end

View file

@ -82,11 +82,7 @@ class TestRetry < MiniTest::Unit::TestCase
end end
it 'should poll like a bad mother...SHUT YO MOUTH' do it 'should poll like a bad mother...SHUT YO MOUTH' do
fake_msg = if MultiJson.respond_to?(:dump) fake_msg = Sidekiq.dump_json({ 'class' => 'Bob', 'args' => [1,2], 'queue' => 'someq' })
MultiJson.dump({ 'class' => 'Bob', 'args' => [1,2], 'queue' => 'someq' })
else
MultiJson.encode({ 'class' => 'Bob', 'args' => [1,2], 'queue' => 'someq' })
end
@redis.expect :multi, [[fake_msg], 1], [] @redis.expect :multi, [[fake_msg], 1], []
@redis.expect :rpush, 1, ['queue:someq', fake_msg] @redis.expect :rpush, 1, ['queue:someq', fake_msg]