Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Mongoid 3 support

  • Loading branch information...
commit b24f871849b34cc9e9cf86e2e26cfbeeb21a45ce 1 parent 68bcf62
@asavartsov asavartsov authored sferik committed
View
43 lib/delayed/backend/mongoid.rb
@@ -24,14 +24,6 @@ class Job
before_save :set_default_run_at
- def self.before_fork
- ::Mongoid.master.connection.close
- end
-
- def self.after_fork
- ::Mongoid.master.connection.connect
- end
-
def self.db_time_now
Time.now.utc
end
@@ -39,40 +31,31 @@ def self.db_time_now
# Reserves this job for the worker.
#
# Uses Mongo's findAndModify operation to atomically pick and lock one
- # job from from the collection. findAndModify is not yet available
- # directly thru Mongoid so go down to the Mongo Ruby driver instead.
+ # job from from the collection.
def self.reserve(worker, max_run_time = Worker.max_run_time)
right_now = db_time_now
- conditions = {:run_at => {"$lte" => right_now}, :failed_at => nil}
- (conditions[:priority] ||= {})['$gte'] = Worker.min_priority.to_i if Worker.min_priority
- (conditions[:priority] ||= {})['$lte'] = Worker.max_priority.to_i if Worker.max_priority
- (conditions[:queue] ||= {})['$in'] = Worker.queues if Worker.queues.any?
-
- conditions['$or'] = [
+ criteria = self.where(
+ :run_at => {"$lte" => right_now},
+ :failed_at => nil
+ ).any_of(
{ :locked_by => worker.name },
{ :locked_at => nil },
{ :locked_at => { '$lt' => (right_now - max_run_time) }}
- ]
+ )
- begin
- result = self.db.collection(self.collection.name).find_and_modify(
- :query => conditions,
- :sort => [['locked_by', -1], ['priority', 1], ['run_at', 1]],
- :update => {"$set" => {:locked_at => right_now, :locked_by => worker.name}}
- )
+ criteria = criteria.where(:priority => {"$gte" => Worker.min_priority.to_i}) if Worker.min_priority
+ criteria = criteria.where(:priority => {"$lte" => Worker.max_priority.to_i}) if Worker.max_priority
+ criteria = criteria.any_in(:queue => Worker.queues) if Worker.queues.any?
- # Return result as a Mongoid document.
- # When Mongoid starts supporting findAndModify, this extra step should no longer be necessary.
- self.find(result["_id"]) unless result.nil?
- rescue Mongo::OperationFailure
- nil # no jobs available
- end
+ criteria.desc(:locked_by).asc(:priority).asc(:run_at).find_and_modify(
+ "$set" => {:locked_at => right_now, :locked_by => worker.name}
+ )
end
# When a worker is exiting, make sure we don't have any locked jobs.
def self.clear_locks!(worker_name)
- self.collection.update({:locked_by => worker_name}, {"$set" => {:locked_at => nil, :locked_by => nil}}, :multi => true)
+ self.where(:locked_by => worker_name).update_all({:locked_at => nil, :locked_by => nil})
end
def reload(*args)
View
26 spec/delayed_job_mongoid_spec.rb
@@ -2,28 +2,4 @@
describe Delayed::Backend::Mongoid::Job do
it_should_behave_like 'a delayed_job backend'
-
- describe "before_fork" do
- after do
- ::Mongoid.master.connection.close
- end
-
- it "should disconnect" do
- lambda do
- Delayed::Backend::Mongoid::Job.before_fork
- end.should change { !!Mongoid.master.connection.connected? }.from(true).to(false)
- end
- end
-
- describe "after_fork" do
- before do
- ::Mongoid.master.connection.close
- end
-
- it "should call reconnect" do
- lambda do
- Delayed::Backend::Mongoid::Job.after_fork
- end.should change { !!Mongoid.master.connection.connected? }.from(false).to(true)
- end
- end
-end
+end
View
2  spec/spec_helper.rb
@@ -7,7 +7,7 @@
require 'delayed/backend/shared_spec'
Mongoid.configure do |config|
- config.master = config.master = Mongo::Connection.new.db('dl_spec')
+ config.connect_to("dl_spec")
end
class Story
Please sign in to comment.
Something went wrong with that request. Please try again.