@@ -6,7 +6,8 @@ class WorkerQueue STATUS_RUNNING = 1 STATUS_ERROR = 2 STATUS_COMPLETED = 3 - + STATUS_SKIPPED = 4 + named_scope :waiting, lambda {{ :conditions => {:status => STATUS_WAITING} }} named_scope :running, @@ -24,11 +25,12 @@ class WorkerQueue # Execute ourselves # Note that the task executed expects Class.method(args_hash, binary_blob) to return true or false. # Options - # *<tt>:keep_binary_data</tt> Do not empty the binary data on completion. + # *<tt>:keep_data</tt> Do not empty the data on completion. + # *<tt>:skip_on_error</tt> Do not stop the execution of a group when an error was encountered. def execute(options = {}) - ah = self.argument_hash.clone - ah.store(:data, self.data) if self.data + ah = argument_hash.clone + ah.store(:data, data) if data begin unless class_name.classify.constantize.send(method_name.to_sym, ah) @@ -41,7 +43,9 @@ class WorkerQueue end # If we have an error, do not run anything in this group (halt the chain) - unless self.status == STATUS_ERROR + if error? + self.status = STATUS_SKIPPED if options[:skip_on_error] + else self.status = STATUS_COMPLETED self.data = nil unless !!options[:keep_data] end @@ -61,19 +65,19 @@ class WorkerQueue def executable? # Return false if picked up by another WQ instance - if self.id - old_lock_version = self.lock_version + if id + old_lock_version = lock_version self.reload - return false if old_lock_version != self.lock_version + return false if old_lock_version != lock_version end # Return true we can sill be executed - return WorkerQueue.available_tasks.include?(self) && !self.completed? && !self.running? + return WorkerQueue.available_tasks.include?(self) && !completed? && !running? end # Validates hash in the argument_hash attribute. If none found, a hash is inserted. def hash_in_argument_hash - self.argument_hash = {} if self.argument_hash.nil? + argument_hash = {} if argument_hash.nil? return true end @@ -86,18 +90,12 @@ class WorkerQueue # Find tasks with a certain flag uncompleted tasks in the database def self.waiting_tasks - waiting( - :order => 'id', - :select => WorkerQueue::WorkerQueueItem.partial_select_attributes - ) + waiting(:order => 'id', :select => partial_select_attributes) end # Find all tasks being worked on at the moment. def self.busy_tasks - busy( - :order => 'id', - :select => WorkerQueue::WorkerQueueItem.partial_select_attributes - ) + busy(:order => 'id', :select => partial_select_attributes) end end lib/worker_queue/worker_queue_item.rb @@ -1,10 +1,6 @@ require 'rake' require 'rubygems' -def worker_queue_loaders - -end - namespace :worker_queue do desc 'Load worker items' @@ -13,7 +9,9 @@ namespace :worker_queue do end desc 'Load worker items and start the worker' - task :load_and_work => [ :load, :work ] do ; end + task :load_and_work => [ :load, :work ] do + + end desc 'Start the worker' task :work => :environment do tasks/worker_queue.rake 9fc3909edf8e54ddf0bb61cc2991004598bda0b3 Bart ten Brinke info@retrosync.com http://github.com/barttenbrinke/worker_queue/commit/5eb8fb7c35b10dbc6a588ebad6b6b7391a52ef14 5eb8fb7c35b10dbc6a588ebad6b6b7391a52ef14 2008-09-16T13:27:39-07:00 2008-09-16T13:27:39-07:00 Cleaned up self and fixed comments a38feba68b698fb2ec3cad0981f1eb73d36b830f Bart ten Brinke info@retrosync.com