diff options
Diffstat (limited to 'activejob/lib/active_job')
15 files changed, 195 insertions, 161 deletions
diff --git a/activejob/lib/active_job/logging.rb b/activejob/lib/active_job/logging.rb index cd29e6908e..54774db601 100644 --- a/activejob/lib/active_job/logging.rb +++ b/activejob/lib/active_job/logging.rb @@ -81,7 +81,7 @@ module ActiveJob private def queue_name(event) - event.payload[:adapter].name.demodulize.remove('Adapter') + "(#{event.payload[:job].queue_name})" + event.payload[:adapter].class.name.demodulize.remove('Adapter') + "(#{event.payload[:job].queue_name})" end def args_info(job) diff --git a/activejob/lib/active_job/queue_adapter.rb b/activejob/lib/active_job/queue_adapter.rb index aa3ebdbc7b..9c4519432d 100644 --- a/activejob/lib/active_job/queue_adapter.rb +++ b/activejob/lib/active_job/queue_adapter.rb @@ -1,4 +1,5 @@ require 'active_job/queue_adapters/inline_adapter' +require 'active_support/core_ext/class/attribute' require 'active_support/core_ext/string/inflections' module ActiveJob @@ -7,27 +8,54 @@ module ActiveJob module QueueAdapter #:nodoc: extend ActiveSupport::Concern + included do + class_attribute :_queue_adapter, instance_accessor: false, instance_predicate: false + self.queue_adapter = :inline + end + # Includes the setter method for changing the active queue adapter. module ClassMethods - mattr_reader(:queue_adapter) { ActiveJob::QueueAdapters::InlineAdapter } + def queue_adapter + _queue_adapter + end # Specify the backend queue provider. The default queue adapter # is the :inline queue. See QueueAdapters for more # information. - def queue_adapter=(name_or_adapter) - @@queue_adapter = \ - case name_or_adapter - when Symbol, String - load_adapter(name_or_adapter) - else - name_or_adapter if name_or_adapter.respond_to?(:enqueue) - end + def queue_adapter=(name_or_adapter_or_class) + self._queue_adapter = interpret_adapter(name_or_adapter_or_class) end private - def load_adapter(name) - "ActiveJob::QueueAdapters::#{name.to_s.camelize}Adapter".constantize + + def interpret_adapter(name_or_adapter_or_class) + case name_or_adapter_or_class + when Symbol, String + ActiveJob::QueueAdapters.lookup(name_or_adapter_or_class).new + else + if queue_adapter?(name_or_adapter_or_class) + name_or_adapter_or_class + elsif queue_adapter_class?(name_or_adapter_or_class) + ActiveSupport::Deprecation.warn "Passing an adapter class is deprecated " \ + "and will be removed in Rails 5.1. Please pass an adapter name " \ + "(.queue_adapter = :#{name_or_adapter_or_class.name.demodulize.remove('Adapter').underscore}) " \ + "or an instance (.queue_adapter = #{name_or_adapter_or_class.name}.new) instead." + name_or_adapter_or_class.new + else + raise ArgumentError + end end + end + + QUEUE_ADAPTER_METHODS = [:enqueue, :enqueue_at].freeze + + def queue_adapter?(object) + QUEUE_ADAPTER_METHODS.all? { |meth| object.respond_to?(meth) } + end + + def queue_adapter_class?(object) + object.is_a?(Class) && QUEUE_ADAPTER_METHODS.all? { |meth| object.public_method_defined?(meth) } + end end end end diff --git a/activejob/lib/active_job/queue_adapters.rb b/activejob/lib/active_job/queue_adapters.rb index 4b91c93dbe..b3d91dc562 100644 --- a/activejob/lib/active_job/queue_adapters.rb +++ b/activejob/lib/active_job/queue_adapters.rb @@ -48,5 +48,14 @@ module ActiveJob autoload :SneakersAdapter autoload :SuckerPunchAdapter autoload :TestAdapter + + ADAPTER = 'Adapter'.freeze + private_constant :ADAPTER + + class << self + def lookup(name) + const_get(name.to_s.camelize << ADAPTER) + end + end end end diff --git a/activejob/lib/active_job/queue_adapters/backburner_adapter.rb b/activejob/lib/active_job/queue_adapters/backburner_adapter.rb index 2453d065de..17703e3e41 100644 --- a/activejob/lib/active_job/queue_adapters/backburner_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/backburner_adapter.rb @@ -13,15 +13,13 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :backburner class BackburnerAdapter - class << self - def enqueue(job) #:nodoc: - Backburner::Worker.enqueue JobWrapper, [ job.serialize ], queue: job.queue_name - end + def enqueue(job) #:nodoc: + Backburner::Worker.enqueue JobWrapper, [ job.serialize ], queue: job.queue_name + end - def enqueue_at(job, timestamp) #:nodoc: - delay = timestamp - Time.current.to_f - Backburner::Worker.enqueue JobWrapper, [ job.serialize ], queue: job.queue_name, delay: delay - end + def enqueue_at(job, timestamp) #:nodoc: + delay = timestamp - Time.current.to_f + Backburner::Worker.enqueue JobWrapper, [ job.serialize ], queue: job.queue_name, delay: delay end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb b/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb index 69d9e70de3..852a6ee326 100644 --- a/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb @@ -13,14 +13,12 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :delayed_job class DelayedJobAdapter - class << self - def enqueue(job) #:nodoc: - Delayed::Job.enqueue(JobWrapper.new(job.serialize), queue: job.queue_name) - end + def enqueue(job) #:nodoc: + Delayed::Job.enqueue(JobWrapper.new(job.serialize), queue: job.queue_name) + end - def enqueue_at(job, timestamp) #:nodoc: - Delayed::Job.enqueue(JobWrapper.new(job.serialize), queue: job.queue_name, run_at: Time.at(timestamp)) - end + def enqueue_at(job, timestamp) #:nodoc: + Delayed::Job.enqueue(JobWrapper.new(job.serialize), queue: job.queue_name, run_at: Time.at(timestamp)) end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/inline_adapter.rb b/activejob/lib/active_job/queue_adapters/inline_adapter.rb index e25d88e723..1d06324c18 100644 --- a/activejob/lib/active_job/queue_adapters/inline_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/inline_adapter.rb @@ -9,14 +9,12 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :inline class InlineAdapter - class << self - def enqueue(job) #:nodoc: - Base.execute(job.serialize) - end + def enqueue(job) #:nodoc: + Base.execute(job.serialize) + end - def enqueue_at(*) #:nodoc: - raise NotImplementedError.new("Use a queueing backend to enqueue jobs in the future. Read more at http://guides.rubyonrails.org/active_job_basics.html") - end + def enqueue_at(*) #:nodoc: + raise NotImplementedError.new("Use a queueing backend to enqueue jobs in the future. Read more at http://guides.rubyonrails.org/active_job_basics.html") end end end diff --git a/activejob/lib/active_job/queue_adapters/qu_adapter.rb b/activejob/lib/active_job/queue_adapters/qu_adapter.rb index 30aa5a4670..94584ef9d8 100644 --- a/activejob/lib/active_job/queue_adapters/qu_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/qu_adapter.rb @@ -16,16 +16,14 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :qu class QuAdapter - class << self - def enqueue(job, *args) #:nodoc: - Qu::Payload.new(klass: JobWrapper, args: [job.serialize]).tap do |payload| - payload.instance_variable_set(:@queue, job.queue_name) - end.push - end + def enqueue(job, *args) #:nodoc: + Qu::Payload.new(klass: JobWrapper, args: [job.serialize]).tap do |payload| + payload.instance_variable_set(:@queue, job.queue_name) + end.push + end - def enqueue_at(job, timestamp, *args) #:nodoc: - raise NotImplementedError - end + def enqueue_at(job, timestamp, *args) #:nodoc: + raise NotImplementedError end class JobWrapper < Qu::Job #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/que_adapter.rb b/activejob/lib/active_job/queue_adapters/que_adapter.rb index e501fe0368..84cc2845b0 100644 --- a/activejob/lib/active_job/queue_adapters/que_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/que_adapter.rb @@ -15,14 +15,12 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :que class QueAdapter - class << self - def enqueue(job) #:nodoc: - JobWrapper.enqueue job.serialize, queue: job.queue_name - end + def enqueue(job) #:nodoc: + JobWrapper.enqueue job.serialize, queue: job.queue_name + end - def enqueue_at(job, timestamp) #:nodoc: - JobWrapper.enqueue job.serialize, queue: job.queue_name, run_at: Time.at(timestamp) - end + def enqueue_at(job, timestamp) #:nodoc: + JobWrapper.enqueue job.serialize, queue: job.queue_name, run_at: Time.at(timestamp) end class JobWrapper < Que::Job #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb b/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb index 34c11a68b2..059754a87f 100644 --- a/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb @@ -17,29 +17,27 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :queue_classic class QueueClassicAdapter - class << self - def enqueue(job) #:nodoc: - build_queue(job.queue_name).enqueue("#{JobWrapper.name}.perform", job.serialize) - end + def enqueue(job) #:nodoc: + build_queue(job.queue_name).enqueue("#{JobWrapper.name}.perform", job.serialize) + end - def enqueue_at(job, timestamp) #:nodoc: - queue = build_queue(job.queue_name) - unless queue.respond_to?(:enqueue_at) - raise NotImplementedError, 'To be able to schedule jobs with queue_classic ' \ - 'the QC::Queue needs to respond to `enqueue_at(timestamp, method, *args)`. ' \ - 'You can implement this yourself or you can use the queue_classic-later gem.' - end - queue.enqueue_at(timestamp, "#{JobWrapper.name}.perform", job.serialize) + def enqueue_at(job, timestamp) #:nodoc: + queue = build_queue(job.queue_name) + unless queue.respond_to?(:enqueue_at) + raise NotImplementedError, 'To be able to schedule jobs with queue_classic ' \ + 'the QC::Queue needs to respond to `enqueue_at(timestamp, method, *args)`. ' \ + 'You can implement this yourself or you can use the queue_classic-later gem.' end + queue.enqueue_at(timestamp, "#{JobWrapper.name}.perform", job.serialize) + end - # Builds a <tt>QC::Queue</tt> object to schedule jobs on. - # - # If you have a custom <tt>QC::Queue</tt> subclass you'll need to subclass - # <tt>ActiveJob::QueueAdapters::QueueClassicAdapter</tt> and override the - # <tt>build_queue</tt> method. - def build_queue(queue_name) - QC::Queue.new(queue_name) - end + # Builds a <tt>QC::Queue</tt> object to schedule jobs on. + # + # If you have a custom <tt>QC::Queue</tt> subclass you'll need to subclass + # <tt>ActiveJob::QueueAdapters::QueueClassicAdapter</tt> and override the + # <tt>build_queue</tt> method. + def build_queue(queue_name) + QC::Queue.new(queue_name) end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/resque_adapter.rb b/activejob/lib/active_job/queue_adapters/resque_adapter.rb index 88c6b48fef..417854afd8 100644 --- a/activejob/lib/active_job/queue_adapters/resque_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/resque_adapter.rb @@ -26,18 +26,16 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :resque class ResqueAdapter - class << self - def enqueue(job) #:nodoc: - Resque.enqueue_to job.queue_name, JobWrapper, job.serialize - end + def enqueue(job) #:nodoc: + Resque.enqueue_to job.queue_name, JobWrapper, job.serialize + end - def enqueue_at(job, timestamp) #:nodoc: - unless Resque.respond_to?(:enqueue_at_with_queue) - raise NotImplementedError, "To be able to schedule jobs with Resque you need the " \ - "resque-scheduler gem. Please add it to your Gemfile and run bundle install" - end - Resque.enqueue_at_with_queue job.queue_name, timestamp, JobWrapper, job.serialize + def enqueue_at(job, timestamp) #:nodoc: + unless Resque.respond_to?(:enqueue_at_with_queue) + raise NotImplementedError, "To be able to schedule jobs with Resque you need the " \ + "resque-scheduler gem. Please add it to your Gemfile and run bundle install" end + Resque.enqueue_at_with_queue job.queue_name, timestamp, JobWrapper, job.serialize end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb b/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb index 21005fc728..743d5ea333 100644 --- a/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb @@ -15,22 +15,22 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :sidekiq class SidekiqAdapter - class << self - def enqueue(job) #:nodoc: - #Sidekiq::Client does not support symbols as keys - Sidekiq::Client.push \ - 'class' => JobWrapper, - 'queue' => job.queue_name, - 'args' => [ job.serialize ] - end + def enqueue(job) #:nodoc: + #Sidekiq::Client does not support symbols as keys + Sidekiq::Client.push \ + 'class' => JobWrapper, + 'wrapped' => job.class.to_s, + 'queue' => job.queue_name, + 'args' => [ job.serialize ] + end - def enqueue_at(job, timestamp) #:nodoc: - Sidekiq::Client.push \ - 'class' => JobWrapper, - 'queue' => job.queue_name, - 'args' => [ job.serialize ], - 'at' => timestamp - end + def enqueue_at(job, timestamp) #:nodoc: + Sidekiq::Client.push \ + 'class' => JobWrapper, + 'wrapped' => job.class.to_s, + 'queue' => job.queue_name, + 'args' => [ job.serialize ], + 'at' => timestamp end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb b/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb index 6d60a2f303..f5737487ca 100644 --- a/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb @@ -16,19 +16,19 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :sneakers class SneakersAdapter - @monitor = Monitor.new + def initialize + @monitor = Monitor.new + end - class << self - def enqueue(job) #:nodoc: - @monitor.synchronize do - JobWrapper.from_queue job.queue_name - JobWrapper.enqueue ActiveSupport::JSON.encode(job.serialize) - end + def enqueue(job) #:nodoc: + @monitor.synchronize do + JobWrapper.from_queue job.queue_name + JobWrapper.enqueue ActiveSupport::JSON.encode(job.serialize) end + end - def enqueue_at(job, timestamp) #:nodoc: - raise NotImplementedError - end + def enqueue_at(job, timestamp) #:nodoc: + raise NotImplementedError end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb b/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb index be9e7fd03a..64c93e8198 100644 --- a/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb @@ -18,14 +18,12 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :sucker_punch class SuckerPunchAdapter - class << self - def enqueue(job) #:nodoc: - JobWrapper.new.async.perform job.serialize - end + def enqueue(job) #:nodoc: + JobWrapper.new.async.perform job.serialize + end - def enqueue_at(job, timestamp) #:nodoc: - raise NotImplementedError - end + def enqueue_at(job, timestamp) #:nodoc: + raise NotImplementedError end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/queue_adapters/test_adapter.rb b/activejob/lib/active_job/queue_adapters/test_adapter.rb index fd7c0b207a..9b7b7139f4 100644 --- a/activejob/lib/active_job/queue_adapters/test_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/test_adapter.rb @@ -10,52 +10,50 @@ module ActiveJob # # Rails.application.config.active_job.queue_adapter = :test class TestAdapter - class << self - attr_accessor(:perform_enqueued_jobs, :perform_enqueued_at_jobs, :filter) - attr_writer(:enqueued_jobs, :performed_jobs) + attr_accessor(:perform_enqueued_jobs, :perform_enqueued_at_jobs, :filter) + attr_writer(:enqueued_jobs, :performed_jobs) - # Provides a store of all the enqueued jobs with the TestAdapter so you can check them. - def enqueued_jobs - @enqueued_jobs ||= [] - end + # Provides a store of all the enqueued jobs with the TestAdapter so you can check them. + def enqueued_jobs + @enqueued_jobs ||= [] + end - # Provides a store of all the performed jobs with the TestAdapter so you can check them. - def performed_jobs - @performed_jobs ||= [] - end + # Provides a store of all the performed jobs with the TestAdapter so you can check them. + def performed_jobs + @performed_jobs ||= [] + end - def enqueue(job) #:nodoc: - return if filtered?(job) + def enqueue(job) #:nodoc: + return if filtered?(job) - job_data = job_to_hash(job) - enqueue_or_perform(perform_enqueued_jobs, job, job_data) - end + job_data = job_to_hash(job) + enqueue_or_perform(perform_enqueued_jobs, job, job_data) + end - def enqueue_at(job, timestamp) #:nodoc: - return if filtered?(job) + def enqueue_at(job, timestamp) #:nodoc: + return if filtered?(job) - job_data = job_to_hash(job, at: timestamp) - enqueue_or_perform(perform_enqueued_at_jobs, job, job_data) - end + job_data = job_to_hash(job, at: timestamp) + enqueue_or_perform(perform_enqueued_at_jobs, job, job_data) + end - private + private - def job_to_hash(job, extras = {}) - { job: job.class, args: job.serialize.fetch('arguments'), queue: job.queue_name }.merge!(extras) - end + def job_to_hash(job, extras = {}) + { job: job.class, args: job.serialize.fetch('arguments'), queue: job.queue_name }.merge!(extras) + end - def enqueue_or_perform(perform, job, job_data) - if perform - performed_jobs << job_data - Base.execute job.serialize - else - enqueued_jobs << job_data - end + def enqueue_or_perform(perform, job, job_data) + if perform + performed_jobs << job_data + Base.execute job.serialize + else + enqueued_jobs << job_data end + end - def filtered?(job) - filter && !Array(filter).include?(job.class) - end + def filtered?(job) + filter && !Array(filter).include?(job.class) end end end diff --git a/activejob/lib/active_job/test_helper.rb b/activejob/lib/active_job/test_helper.rb index 66508114d1..4efb4b72d2 100644 --- a/activejob/lib/active_job/test_helper.rb +++ b/activejob/lib/active_job/test_helper.rb @@ -1,3 +1,4 @@ +require 'active_support/core_ext/class/subclasses' require 'active_support/core_ext/hash/keys' module ActiveJob @@ -7,18 +8,27 @@ module ActiveJob included do def before_setup - @old_queue_adapter = queue_adapter - ActiveJob::Base.queue_adapter = :test + test_adapter = ActiveJob::QueueAdapters::TestAdapter.new + + @old_queue_adapters = (ActiveJob::Base.subclasses << ActiveJob::Base).select do |klass| + # only override explicitly set adapters, a quirk of `class_attribute` + klass.singleton_class.public_instance_methods(false).include?(:_queue_adapter) + end.map do |klass| + [klass, klass.queue_adapter].tap do + klass.queue_adapter = test_adapter + end + end + clear_enqueued_jobs clear_performed_jobs - queue_adapter.perform_enqueued_jobs = false - queue_adapter.perform_enqueued_at_jobs = false super end def after_teardown super - ActiveJob::Base.queue_adapter = @old_queue_adapter + @old_queue_adapters.each do |(klass, adapter)| + klass.queue_adapter = adapter + end end # Asserts that the number of enqueued jobs matches the given number. @@ -253,15 +263,20 @@ module ActiveJob end def perform_enqueued_jobs(only: nil) - @old_perform_enqueued_jobs = queue_adapter.perform_enqueued_jobs - @old_perform_enqueued_at_jobs = queue_adapter.perform_enqueued_at_jobs - queue_adapter.perform_enqueued_jobs = true - queue_adapter.perform_enqueued_at_jobs = true - queue_adapter.filter = only - yield - ensure - queue_adapter.perform_enqueued_jobs = @old_perform_enqueued_jobs - queue_adapter.perform_enqueued_at_jobs = @old_perform_enqueued_at_jobs + old_perform_enqueued_jobs = queue_adapter.perform_enqueued_jobs + old_perform_enqueued_at_jobs = queue_adapter.perform_enqueued_at_jobs + old_filter = queue_adapter.filter + + begin + queue_adapter.perform_enqueued_jobs = true + queue_adapter.perform_enqueued_at_jobs = true + queue_adapter.filter = only + yield + ensure + queue_adapter.perform_enqueued_jobs = old_perform_enqueued_jobs + queue_adapter.perform_enqueued_at_jobs = old_perform_enqueued_at_jobs + queue_adapter.filter = old_filter + end end def queue_adapter |