Reschedule jobs using a time provided by the handler, or fall back to…

… default

lib/delayed/backend/base.rb

@@ -96,12 +96,17 @@ def hook(name, *args)
         end
       end
 
+      def reschedule_at
+        payload_object.respond_to?(:reschedule_at) ? 
+          payload_object.reschedule_at(self.class.db_time_now, attempts) :
+          self.class.db_time_now + (attempts ** 4) + 5
+      end
+
     protected
 
       def set_default_run_at
         self.run_at ||= self.class.db_time_now
       end
-
     end
   end
 end

lib/delayed/backend/shared_spec.rb

@@ -294,7 +294,7 @@ def create_job(opts = {})
       @job.locked_at.should be_nil
     end
   end
-
+  
   context "large handler" do
     before do
       text = "Lorem ipsum dolor sit amet. " * 1000
@@ -402,8 +402,8 @@ def create_job(opts = {})
         # reset defaults
         Delayed::Worker.destroy_failed_jobs = true
         Delayed::Worker.max_attempts = 25
-
-        @job = Delayed::Job.enqueue ErrorJob.new
+        
+        @job = Delayed::Job.enqueue(ErrorJob.new)
       end
 
       it "should record last_error when destroy_failed_jobs = false, max_attempts = 1" do
@@ -425,6 +425,14 @@ def create_job(opts = {})
         @job.run_at.should > Delayed::Job.db_time_now - 10.minutes
         @job.run_at.should < Delayed::Job.db_time_now + 10.minutes
       end
+
+      it 'should re-schedule with handler provided time if present' do
+        @job = Delayed::Job.enqueue(CustomRescheduleJob.new(99.minutes))
+        @worker.run(@job)
+        @job.reload
+
+        (Delayed::Job.db_time_now + 99.minutes - @job.run_at).abs.should < 1
+      end
     end
 
     context "reschedule" do

lib/delayed/worker.rb

@@ -128,7 +128,7 @@ def run(job)
     # Uses an exponential scale depending on the number of failed attempts.
     def reschedule(job, time = nil)
       if (job.attempts += 1) < self.class.max_attempts
-        time ||= Job.db_time_now + (job.attempts ** 4) + 5
+        time ||= job.reschedule_at
         job.run_at = time
         job.unlock
         job.save!

spec/sample_jobs.rb

@@ -14,6 +14,12 @@ class ErrorJob
   def perform; raise 'did not work'; end
 end
 
+class CustomRescheduleJob < Struct.new(:offset)
+  cattr_accessor :runs; self.runs = 0
+  def perform; raise 'did not work'; end
+  def reschedule_at(time, attempts); time + offset; end
+end
+
 class LongRunningJob
   def perform; sleep 250; end
 end

spec/worker_spec.rb

@@ -34,4 +34,55 @@
       lambda { Delayed::Worker.guess_backend }.should_not change { Delayed::Worker.backend }
     end
   end
+
+  describe "running a job" do
+    before(:each) do
+      @worker = Delayed::Worker.new
+    end
+
+    after(:each) do 
+      Delayed::Job.delete_all
+    end
+
+    describe 'that fails' do
+      before(:each) do
+        @handler = ErrorJob.new
+        @job = Delayed::Job.enqueue(@handler)
+      end
+
+      it 'should increase the attempts' do
+        @worker.run(@job)
+        @job.attempts.should == 1
+      end
+
+      it 'should reschedule the job in the future' do 
+        @worker.run(@job) 
+        @job.run_at.should > Job.db_time_now + 5
+      end
+
+      describe 'with custom rescheduling strategy' do
+        before(:each) do
+          @reschedule_at = Time.current + 7.hours
+          @handler.stub!(:reschedule_at).and_return(@reschedule_at)
+        end
+
+        it 'should invoke the strategy' do
+          @handler.should_receive(:reschedule_at) do |time, attempts|
+            (Job.db_time_now - time).should < 2
+            attempts.should == 1
+
+            Job.db_time.now + 5
+          end
+
+          @worker.run(@job)
+        end
+
+      end
+
+      it 'should reschedule at the specified time' do        
+        @worker.run(@job)
+        @job.run_at.should == @reschedule_at
+      end
+    end
+  end
 end