.gitignore HISTORY.txt delayed_job.gemspec lib/delayed_job.rb @@ -1,5 +1 @@ -require File.dirname(__FILE__) + '/lib/delayed/message_sending' -require File.dirname(__FILE__) + '/lib/delayed/performable_method' -require File.dirname(__FILE__) + '/lib/delayed/job' - -Object.send(:include, Delayed::MessageSending) \ No newline at end of file +require File.dirname(__FILE__) + '/lib/delayed_job' init.rb @@ -15,8 +15,8 @@ module Delayed cattr_accessor :destroy_failed_jobs self.destroy_failed_jobs = true - # Every worker has a unique name which by default is the pid of the process. - # There are some advantages to overriding this with something which survives worker retarts: + # Every worker has a unique name which by default is the pid of the process. + # There are some advantages to overriding this with something which survives worker retarts: # Workers can safely resume working on tasks which are locked by themselves. The worker will assume that it crashed before. cattr_accessor :worker_name self.worker_name = "host:#{Socket.gethostname} pid:#{Process.pid}" rescue "pid:#{Process.pid}" @@ -25,10 +25,10 @@ module Delayed NextTaskOrder = 'priority DESC, run_at ASC' ParseObjectFromYaml = /\!ruby\/\w+\:([^\s]+)/ - + cattr_accessor :min_priority, :max_priority self.min_priority = nil - self.max_priority = nil + self.max_priority = nil class LockError < StandardError end @@ -45,8 +45,8 @@ module Delayed def payload_object @payload_object ||= deserialize(self['handler']) end - - def name + + def name @name ||= begin payload = payload_object if payload.respond_to?(:display_name) @@ -80,13 +80,13 @@ module Delayed if block_given? priority = args.first || 0 run_at = args.second - + Job.create(:payload_object => EvaledJob.new(&block), :priority => priority.to_i, :run_at => run_at) else object = args.first priority = args.second || 0 run_at = args.third - + unless object.respond_to?(:perform) raise ArgumentError, 'Cannot enqueue items which do not respond to perform' end @@ -96,32 +96,32 @@ module Delayed end def self.find_available(limit = 5, max_run_time = MAX_RUN_TIME) - - time_now = db_time_now - + + time_now = db_time_now + sql = NextTaskSQL.dup conditions = [time_now, time_now - max_run_time, worker_name] - + if self.min_priority sql << ' AND (priority >= ?)' conditions << min_priority end - + if self.max_priority sql << ' AND (priority <= ?)' - conditions << max_priority + conditions << max_priority end - conditions.unshift(sql) - + conditions.unshift(sql) + records = ActiveRecord::Base.silence do find(:all, :conditions => conditions, :order => NextTaskOrder, :limit => limit) end - + records.sort { rand() } - end - + end + # Get the payload of the next job we can get an exclusive lock on. # If no jobs are left we return nil def self.reserve(max_run_time = MAX_RUN_TIME, &block) @@ -142,7 +142,7 @@ module Delayed rescue LockError # We did not get the lock, some other worker process must have logger.warn "* [JOB] failed to aquire exclusive lock for #{job.name}" - rescue StandardError => e + rescue StandardError => e job.reschedule e.message, e.backtrace log_exception(job, e) return job @@ -160,16 +160,16 @@ module Delayed # We don't own this job so we will update the locked_by name and the locked_at self.class.update_all(["locked_at = ?, locked_by = ?", now, worker], ["id = ? and (locked_at is null or locked_at < ?)", id, (now - max_run_time.to_i)]) else - # We already own this job, this may happen if the job queue crashes. + # We already own this job, this may happen if the job queue crashes. # Simply resume and update the locked_at self.class.update_all(["locked_at = ?", now], ["id = ? and locked_by = ?", id, worker]) end raise LockError.new("Attempted to aquire exclusive lock failed") unless affected_rows == 1 - + self.locked_at = now self.locked_by = worker end - + def unlock self.locked_at = nil self.locked_by = nil lib/delayed/job.rb @@ -2,19 +2,21 @@ module Delayed class Worker SLEEP = 5 + cattr_accessor :logger + self.logger = RAILS_DEFAULT_LOGGER if const_defined?(:RAILS_DEFAULT_LOGGER) + def initialize(options={}) - @quiet = options[:quiet] + @quiet = options[:quiet] Delayed::Job.min_priority = options[:min_priority] if options.has_key?(:min_priority) Delayed::Job.max_priority = options[:max_priority] if options.has_key?(:max_priority) - end + end def start say "*** Starting job worker #{Delayed::Job.worker_name}" trap('TERM') { say 'Exiting...'; $exit = true } trap('INT') { say 'Exiting...'; $exit = true } - - + loop do result = nil @@ -33,15 +35,15 @@ module Delayed end break if $exit - end - + end + ensure Delayed::Job.clear_locks! end - + def say(text) puts text unless @quiet - RAILS_DEFAULT_LOGGER.info text + logger.info text if logger end end lib/delayed/worker.rb bcf8d1d2decd70b59b63640263bfde9a12959c5c Justin Knowlden jaknowlden@gmail.com http://github.com/collectiveidea/delayed_job/commit/9ba6b0872375b55eadf9979b8773efe9132bda2c 9ba6b0872375b55eadf9979b8773efe9132bda2c 2008-11-28T22:06:07-08:00 2008-11-28T22:06:07-08:00 Made a gem out of delayed_job so I can use it in my services. Had to modify Worker to not infer DJ is running in a Rails instance. 0f6c5a8fe1b4c73c84d970e1254ac437f76fd207 Justin Knowlden jaknowlden@gmail.com