From 3648838173a1d7217560ed1cf8ab2a217ccbc6d1 Mon Sep 17 00:00:00 2001 From: Mike Perham Date: Mon, 19 May 2014 15:27:28 -0700 Subject: [PATCH] Implement enqueue_at/enqueue_in Delayed jobs are supported by all systems except QueueClassic. For it I decided to raise NotImplementedError. The inline implementation is a bit rough. --- Gemfile | 1 + Gemfile.lock | 7 +++++++ lib/active_job/enqueuing.rb | 20 +++++++++++++++++++ .../queue_adapters/delayed_job_adapter.rb | 6 +++++- .../queue_adapters/inline_adapter.rb | 15 +++++++++++++- .../queue_adapters/queue_classic_adapter.rb | 4 ++++ .../queue_adapters/resque_adapter.rb | 6 ++++++ .../queue_adapters/sidekiq_adapter.rb | 7 +++++++ .../queue_adapters/sucker_punch_adapter.rb | 14 +++++++++++++ test/cases/queuing_test.rb | 9 +++++++++ 10 files changed, 87 insertions(+), 2 deletions(-) diff --git a/Gemfile b/Gemfile index 550da15615..cc2d17b5b7 100644 --- a/Gemfile +++ b/Gemfile @@ -4,6 +4,7 @@ gemspec gem 'rake' gem 'resque' +gem 'resque-scheduler' gem 'sidekiq' gem 'sucker_punch' gem 'delayed_job' diff --git a/Gemfile.lock b/Gemfile.lock index c1df0ebc36..17f0d032e3 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -47,6 +47,12 @@ GEM redis-namespace (~> 1.2) sinatra (>= 0.9.2) vegas (~> 0.1.2) + resque-scheduler (2.2.0) + redis (>= 3.0.0) + resque (>= 1.20.0, < 1.25) + rufus-scheduler (~> 2.0) + rufus-scheduler (2.0.24) + tzinfo (>= 0.3.22) sidekiq (3.0.2) celluloid (>= 0.15.2) connection_pool (>= 2.0.0) @@ -76,5 +82,6 @@ DEPENDENCIES queue_classic rake resque + resque-scheduler sidekiq sucker_punch diff --git a/lib/active_job/enqueuing.rb b/lib/active_job/enqueuing.rb index 324385216e..a5a50d69db 100644 --- a/lib/active_job/enqueuing.rb +++ b/lib/active_job/enqueuing.rb @@ -13,5 +13,25 @@ module ActiveJob def enqueue(*args) queue_adapter.queue self, *Parameters.serialize(args) end + + ## + # Enqueue a job to be performed at +interval+ from now. + # + # enqueue_in(1.week, "mike") + # + # Returns truthy if a job was scheduled. + def enqueue_in(interval, *args) + enqueue_at(interval.from_now, *args) + end + + ## + # Enqueue a job to be performed at an explicit point in time. + # + # enqueue_at(Date.tomorrow.midnight, "mike") + # + # Returns truthy if a job was scheduled. + def enqueue_at(timestamp, *args) + queue_adapter.queue_at self, timestamp.to_f, *Parameters.serialize(args) + end end end diff --git a/lib/active_job/queue_adapters/delayed_job_adapter.rb b/lib/active_job/queue_adapters/delayed_job_adapter.rb index 14072e2801..214733e3a6 100644 --- a/lib/active_job/queue_adapters/delayed_job_adapter.rb +++ b/lib/active_job/queue_adapters/delayed_job_adapter.rb @@ -7,8 +7,12 @@ module ActiveJob def queue(job, *args) JobWrapper.new.delay(queue: job.queue_name).perform(job, *args) end + + def queue_at(job, timestamp, *args) + JobWrapper.new.delay(queue: job.queue_name, run_at: timestamp).perform(job, *args) + end end - + class JobWrapper def perform(job, *args) job.new.perform *Parameters.deserialize(args) diff --git a/lib/active_job/queue_adapters/inline_adapter.rb b/lib/active_job/queue_adapters/inline_adapter.rb index cffa55af82..dd4b3f4fc0 100644 --- a/lib/active_job/queue_adapters/inline_adapter.rb +++ b/lib/active_job/queue_adapters/inline_adapter.rb @@ -5,7 +5,20 @@ module ActiveJob def queue(job, *args) job.new.perform *Parameters.deserialize(args) end + + def queue_at(job, ts, *args) + # TODO better error handling? + Thread.new do + begin + interval = Time.now.to_f - ts + sleep(interval) if interval > 0 + job.new.perform *Parameters.deserialize(args) + rescue => ex + puts ex.message + end + end + end end end end -end \ No newline at end of file +end diff --git a/lib/active_job/queue_adapters/queue_classic_adapter.rb b/lib/active_job/queue_adapters/queue_classic_adapter.rb index e3392a646e..38c04ca5c9 100644 --- a/lib/active_job/queue_adapters/queue_classic_adapter.rb +++ b/lib/active_job/queue_adapters/queue_classic_adapter.rb @@ -8,6 +8,10 @@ module ActiveJob qc_queue = QC::Queue.new(job.queue_name) qc_queue.enqueue("ActiveJob::QueueAdapters::QueueClassicAdapter::JobWrapper.perform", job, *args) end + + def queue_at(job, timestamp, *args) + raise NotImplementedError + end end class JobWrapper diff --git a/lib/active_job/queue_adapters/resque_adapter.rb b/lib/active_job/queue_adapters/resque_adapter.rb index 6686f10593..8fa8dddd11 100644 --- a/lib/active_job/queue_adapters/resque_adapter.rb +++ b/lib/active_job/queue_adapters/resque_adapter.rb @@ -1,6 +1,7 @@ require 'resque' require 'active_support/core_ext/enumerable' require 'active_support/core_ext/array/access' +require 'resque_scheduler' module ActiveJob module QueueAdapters @@ -9,6 +10,11 @@ module ActiveJob def queue(job, *args) Resque.enqueue JobWrapper.new(job), job, *args end + + def queue_at(job, timestamp, *args) + # requires resque-scheduler + Resque.enqueue_at timestamp, JobWrapper.new(job), job, *args + end end class JobWrapper diff --git a/lib/active_job/queue_adapters/sidekiq_adapter.rb b/lib/active_job/queue_adapters/sidekiq_adapter.rb index c8fac32963..41f58f554a 100644 --- a/lib/active_job/queue_adapters/sidekiq_adapter.rb +++ b/lib/active_job/queue_adapters/sidekiq_adapter.rb @@ -7,6 +7,13 @@ module ActiveJob def queue(job, *args) JobWrapper.client_push class: JobWrapper, queue: job.queue_name, args: [ job, *args ] end + + def queue_at(job, timestamp, *args) + job = { class: JobWrapper, queue: job.queue_name, args: [ job, *args ], at: timestamp } + # Optimization to enqueue something now that is scheduled to go out now or in the past + job.delete(:at) if timestamp <= Time.now.to_f + JobWrapper.client_push(job) + end end class JobWrapper diff --git a/lib/active_job/queue_adapters/sucker_punch_adapter.rb b/lib/active_job/queue_adapters/sucker_punch_adapter.rb index 182b5ce018..713da08359 100644 --- a/lib/active_job/queue_adapters/sucker_punch_adapter.rb +++ b/lib/active_job/queue_adapters/sucker_punch_adapter.rb @@ -7,6 +7,16 @@ module ActiveJob def queue(job, *args) JobWrapper.new.async.perform(job, *args) end + + def queue_at(job, timestamp, *args) + secs = Time.now.to_f - timestamp + if secs < 1 + # Optimization to enqueue something now that is scheduled to go out now or in the past + JobWrapper.new.async.perform(job, *args) + else + JobWrapper.new.async.later(secs, job, *args) + end + end end class JobWrapper @@ -15,6 +25,10 @@ module ActiveJob def perform(job_name, *args) job_name.new.perform *Parameters.deserialize(args) end + + def later(sec, job_name, *args) + after(sec) { p args; perform(job_name, *args) } + end end end end diff --git a/test/cases/queuing_test.rb b/test/cases/queuing_test.rb index b6180a23dd..f29f627da1 100644 --- a/test/cases/queuing_test.rb +++ b/test/cases/queuing_test.rb @@ -1,5 +1,6 @@ require 'helper' require 'jobs/hello_job' +require 'active_support/core_ext/numeric/time' class QueuingTest < ActiveSupport::TestCase @@ -16,4 +17,12 @@ class QueuingTest < ActiveSupport::TestCase HelloJob.enqueue "Jamie" assert_equal "Jamie says hello", $BUFFER.pop end + + test 'run queued job later' do + begin + result = HelloJob.enqueue_at 1.second.ago, "Jamie" + assert_not_nil result + rescue NotImplementedError + end + end end