diff options
Diffstat (limited to 'lib/active_job/queue_adapters')
-rw-r--r-- | lib/active_job/queue_adapters/backburner_adapter.rb | 25 | ||||
-rw-r--r-- | lib/active_job/queue_adapters/inline_adapter.rb | 2 | ||||
-rw-r--r-- | lib/active_job/queue_adapters/sidekiq_adapter.rb | 14 | ||||
-rw-r--r-- | lib/active_job/queue_adapters/sneakers_adapter.rb | 10 | ||||
-rw-r--r-- | lib/active_job/queue_adapters/sucker_punch_adapter.rb | 7 |
5 files changed, 50 insertions, 8 deletions
diff --git a/lib/active_job/queue_adapters/backburner_adapter.rb b/lib/active_job/queue_adapters/backburner_adapter.rb new file mode 100644 index 0000000000..0d023d9ee7 --- /dev/null +++ b/lib/active_job/queue_adapters/backburner_adapter.rb @@ -0,0 +1,25 @@ +require 'backburner' + +module ActiveJob + module QueueAdapters + class BackburnerAdapter + class << self + def queue(job, *args) + Backburner::Worker.enqueue JobWrapper, [ job.name, *args ], queue: job.queue_name + end + + def queue_at(job, timestamp, *args) + raise NotImplementedError + end + end + + class JobWrapper + class << self + def perform(job_name, *args) + job_name.constantize.new.perform *Parameters.deserialize(args) + end + end + end + end + end +end diff --git a/lib/active_job/queue_adapters/inline_adapter.rb b/lib/active_job/queue_adapters/inline_adapter.rb index 4afd4f708e..8a2c7d9d92 100644 --- a/lib/active_job/queue_adapters/inline_adapter.rb +++ b/lib/active_job/queue_adapters/inline_adapter.rb @@ -14,7 +14,7 @@ module ActiveJob sleep(interval) if interval > 0 job.new.perform *Parameters.deserialize(args) rescue => ex - ActiveSupport::Notifications.instrument "perform_error.active_job", adapter: self, job: job, params: args, error: ex + ActiveJob::Base.logger "Error performing #{job}: #{ex.message}" end end end diff --git a/lib/active_job/queue_adapters/sidekiq_adapter.rb b/lib/active_job/queue_adapters/sidekiq_adapter.rb index c6faf9d054..be6bd4ee01 100644 --- a/lib/active_job/queue_adapters/sidekiq_adapter.rb +++ b/lib/active_job/queue_adapters/sidekiq_adapter.rb @@ -5,12 +5,20 @@ module ActiveJob class SidekiqAdapter class << self def queue(job, *args) - JobWrapper.client_push class: JobWrapper, queue: job.queue_name, args: [ job, *args ] + Sidekiq::Client.push \ + 'class' => JobWrapper, + 'queue' => job.queue_name, + 'args' => [ job, *args ], + 'retry' => true end def queue_at(job, timestamp, *args) - job = { class: JobWrapper, queue: job.queue_name, args: [ job, *args ], at: timestamp } - JobWrapper.client_push(job) + Sidekiq::Client.push \ + 'class' => JobWrapper, + 'queue' => job.queue_name, + 'args' => [ job, *args ], + 'at' => timestamp, + 'retry' => true end end diff --git a/lib/active_job/queue_adapters/sneakers_adapter.rb b/lib/active_job/queue_adapters/sneakers_adapter.rb index ae9e49a0bf..b299b25a96 100644 --- a/lib/active_job/queue_adapters/sneakers_adapter.rb +++ b/lib/active_job/queue_adapters/sneakers_adapter.rb @@ -1,11 +1,17 @@ require 'sneakers' +require 'thread' module ActiveJob module QueueAdapters class SneakersAdapter + @mutex = Mutex.new + class << self def queue(job, *args) - JobWrapper.enqueue([job, *args]) + @mutex.synchronize do + JobWrapper.from_queue job.queue_name + JobWrapper.enqueue [ job, *args ] + end end def queue_at(job, timestamp, *args) @@ -16,8 +22,6 @@ module ActiveJob class JobWrapper include Sneakers::Worker - self.from_queue("queue", {}) - def work(job, *args) job.new.perform *Parameters.deserialize(args) end diff --git a/lib/active_job/queue_adapters/sucker_punch_adapter.rb b/lib/active_job/queue_adapters/sucker_punch_adapter.rb index 8125499695..79043b06e0 100644 --- a/lib/active_job/queue_adapters/sucker_punch_adapter.rb +++ b/lib/active_job/queue_adapters/sucker_punch_adapter.rb @@ -9,7 +9,12 @@ module ActiveJob end def queue_at(job, timestamp, *args) - JobWrapper.new.async.later(secs, job, *args) + delay = Time.now.to_f - timestamp + if delay > 0 + JobWrapper.new.async.later(delay, job, *args) + else + JobWrapper.new.async.perform(job, *args) + end end end |