2012-01-26 15:45:04 -05:00
|
|
|
require 'sidekiq/util'
|
2012-02-04 19:53:09 -05:00
|
|
|
require 'sidekiq/middleware'
|
2012-01-26 15:45:04 -05:00
|
|
|
require 'celluloid'
|
2012-01-25 16:32:51 -05:00
|
|
|
|
|
|
|
module Sidekiq
|
|
|
|
class Processor
|
2012-01-26 15:45:04 -05:00
|
|
|
include Util
|
2012-01-29 17:35:16 -05:00
|
|
|
include Celluloid
|
2012-01-25 16:32:51 -05:00
|
|
|
|
|
|
|
def initialize(boss)
|
|
|
|
@boss = boss
|
|
|
|
end
|
|
|
|
|
2012-02-07 20:18:28 -05:00
|
|
|
def process(msg, queue)
|
2012-02-04 19:53:09 -05:00
|
|
|
klass = constantize(msg['class'])
|
2012-02-07 20:18:28 -05:00
|
|
|
invoke_chain(klass.new, msg, queue)
|
2012-02-04 19:53:09 -05:00
|
|
|
@boss.processor_done!(current_actor)
|
|
|
|
end
|
|
|
|
|
2012-02-07 20:18:28 -05:00
|
|
|
def invoke_chain(worker, msg, queue)
|
2012-02-05 00:45:34 -05:00
|
|
|
chain = Sidekiq::Middleware::Chain.retrieve.dup
|
|
|
|
traverse_chain = lambda do
|
|
|
|
if chain.empty?
|
|
|
|
worker.perform(*msg['args'])
|
|
|
|
else
|
2012-02-07 20:18:28 -05:00
|
|
|
chain.shift.call(worker, msg, queue, &traverse_chain)
|
2012-02-04 19:53:09 -05:00
|
|
|
end
|
|
|
|
end
|
2012-02-05 00:45:34 -05:00
|
|
|
traverse_chain.call
|
2012-01-25 16:32:51 -05:00
|
|
|
end
|
2012-01-29 17:35:16 -05:00
|
|
|
|
|
|
|
# See http://github.com/tarcieri/celluloid/issues/22
|
|
|
|
def inspect
|
|
|
|
"Sidekiq::Processor<#{object_id}>"
|
|
|
|
end
|
2012-01-25 16:32:51 -05:00
|
|
|
end
|
|
|
|
end
|