Class: QueueingProxy::Worker
- Inherits:
-
Object
- Object
- QueueingProxy::Worker
- Defined in:
- lib/queueing_proxy/worker.rb
Defined Under Namespace
Modules: Priority Classes: Upstream
Instance Attribute Summary collapse
-
#logger ⇒ Object
readonly
Returns the value of attribute logger.
Instance Method Summary collapse
-
#beanstalk ⇒ Object
Setup a beanstalk connection.
-
#initialize(logger, to_host, to_port, beanstalk_host, tube, retries = 3) ⇒ Worker
constructor
A new instance of Worker.
-
#run ⇒ Object
Run the beanstalk consumer that pops the job off the queue and passes it to a connection object that makes an upstream connection.
Constructor Details
#initialize(logger, to_host, to_port, beanstalk_host, tube, retries = 3) ⇒ Worker
Returns a new instance of Worker.
14 15 16 |
# File 'lib/queueing_proxy/worker.rb', line 14 def initialize(logger, to_host, to_port, beanstalk_host, tube, retries=3) @logger, @to_host, @to_port, @beanstalk_host, @tube, @retries = logger, to_host, to_port, beanstalk_host, tube, retries end |
Instance Attribute Details
#logger ⇒ Object (readonly)
Returns the value of attribute logger.
12 13 14 |
# File 'lib/queueing_proxy/worker.rb', line 12 def logger @logger end |
Instance Method Details
#beanstalk ⇒ Object
Setup a beanstalk connection
19 20 21 |
# File 'lib/queueing_proxy/worker.rb', line 19 def beanstalk @beanstalk ||= EMJack::Connection.new(:host => @beanstalk_host, :tube => @tube) end |
#run ⇒ Object
Run the beanstalk consumer that pops the job off the queue and passes it to a connection object that makes an upstream connection
25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
# File 'lib/queueing_proxy/worker.rb', line 25 def run logger.debug "Worker #{@tube}@#{@beanstalk_host} #{object_id} reporting again for duty" beanstalk.reserve {|job| logger.info "Worker #{object_id} reserved #{job}" upstream = Upstream.new(job.body, @to_host, @to_port, 15, logger).request upstream.errback{ logger.info "Worker #{object_id} upstream connection timed-out with #{job}." # If there's an upstream problem, try this a fe more times, then bury it job.stats{|stats| if stats['reserves'] < @retries logger.info "Worker #{object_id} delayed for 5 seconds." job.release(:delay => 5){ run } else logger.info "Worker #{object_id} max #{@retries} retries. Burying." job.bury(Priority::Lowest) { run } end } } upstream.callback { case status = Integer(upstream.response.status_code) when 200..299 logger.info "Upstream HTTP #{status} response - Worker #{object_id} deleting #{job.jobid}." job.delete { run } else logger.info "Upstream HTTP #{status} response - Worker #{object_id}. Burying #{job} for inspection." job.bury(Priority::Lowest) { run } end } }.errback{|status| case status when :disconnected logger.error "Worker #{object_id} beanstalk reservation error." else logger.error "Worker #{object_id} unhandled beanstalk error #{status}." end run # Keep on chuggin partner! } end |