From 3648838173a1d7217560ed1cf8ab2a217ccbc6d1 Mon Sep 17 00:00:00 2001 From: Mike Perham Date: Mon, 19 May 2014 15:27:28 -0700 Subject: 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. --- lib/active_job/enqueuing.rb | 20 ++++++++++++++++++++ lib/active_job/queue_adapters/delayed_job_adapter.rb | 6 +++++- lib/active_job/queue_adapters/inline_adapter.rb | 15 ++++++++++++++- .../queue_adapters/queue_classic_adapter.rb | 4 ++++ lib/active_job/queue_adapters/resque_adapter.rb | 6 ++++++ lib/active_job/queue_adapters/sidekiq_adapter.rb | 7 +++++++ .../queue_adapters/sucker_punch_adapter.rb | 14 ++++++++++++++ 7 files changed, 70 insertions(+), 2 deletions(-) (limited to 'lib/active_job') 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 -- cgit v1.2.3