diff options
Diffstat (limited to 'activejob/lib/active_job')
| -rw-r--r-- | activejob/lib/active_job/arguments.rb | 4 | ||||
| -rw-r--r-- | activejob/lib/active_job/async_job.rb | 74 | ||||
| -rw-r--r-- | activejob/lib/active_job/base.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/gem_version.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapter.rb | 8 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapters.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapters/async_adapter.rb | 105 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapters/inline_adapter.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb | 25 | ||||
| -rw-r--r-- | activejob/lib/active_job/railtie.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/test_helper.rb | 582 |
11 files changed, 415 insertions, 393 deletions
diff --git a/activejob/lib/active_job/arguments.rb b/activejob/lib/active_job/arguments.rb index e56bc79328..33bd5b4eb3 100644 --- a/activejob/lib/active_job/arguments.rb +++ b/activejob/lib/active_job/arguments.rb @@ -25,7 +25,7 @@ module ActiveJob # Raised when an unsupported argument type is set as a job argument. We # currently support NilClass, Fixnum, Float, String, TrueClass, FalseClass, - # Bignum and objects that can be represented as GlobalIDs (ex: Active Record). + # Bignum, BigDecimal, and objects that can be represented as GlobalIDs (ex: Active Record). # Raised if you set the key for a Hash something else than a string or # a symbol. Also raised when trying to serialize an object which can't be # identified with a Global ID - such as an unpersisted Active Record model. @@ -34,7 +34,7 @@ module ActiveJob module Arguments extend self # :nodoc: - TYPE_WHITELIST = [ NilClass, Fixnum, Float, String, TrueClass, FalseClass, Bignum ] + TYPE_WHITELIST = [ NilClass, Fixnum, Float, String, TrueClass, FalseClass, Bignum, BigDecimal ] # Serializes a set of arguments. Whitelisted types are returned # as-is. Arrays/Hashes are serialized element by element. diff --git a/activejob/lib/active_job/async_job.rb b/activejob/lib/active_job/async_job.rb deleted file mode 100644 index 6c1c070994..0000000000 --- a/activejob/lib/active_job/async_job.rb +++ /dev/null @@ -1,74 +0,0 @@ -require 'concurrent' - -module ActiveJob - # == Active Job Async Job - # - # When enqueueing jobs with Async Job each job will be executed asynchronously - # on a +concurrent-ruby+ thread pool. All job data is retained in memory. - # Because job data is not saved to a persistent datastore there is no - # additional infrastructure needed and jobs process quickly. The lack of - # persistence, however, means that all unprocessed jobs will be lost on - # application restart. Therefore in-memory queue adapters are unsuitable for - # most production environments but are excellent for development and testing. - # - # Read more about Concurrent Ruby {here}[https://github.com/ruby-concurrency/concurrent-ruby]. - # - # To use Async Job set the queue_adapter config to +:async+. - # - # Rails.application.config.active_job.queue_adapter = :async - # - # Async Job supports job queues specified with +queue_as+. Queues are created - # automatically as needed and each has its own thread pool. - class AsyncJob - - DEFAULT_EXECUTOR_OPTIONS = { - min_threads: [2, Concurrent.processor_count].max, - max_threads: Concurrent.processor_count * 10, - auto_terminate: true, - idletime: 60, # 1 minute - max_queue: 0, # unlimited - fallback_policy: :caller_runs # shouldn't matter -- 0 max queue - }.freeze - - QUEUES = Concurrent::Map.new do |hash, queue_name| #:nodoc: - hash.compute_if_absent(queue_name) { ActiveJob::AsyncJob.create_thread_pool } - end - - class << self - # Forces jobs to process immediately when testing the Active Job gem. - # This should only be called from within unit tests. - def perform_immediately! #:nodoc: - @perform_immediately = true - end - - # Allows jobs to run asynchronously when testing the Active Job gem. - # This should only be called from within unit tests. - def perform_asynchronously! #:nodoc: - @perform_immediately = false - end - - def create_thread_pool #:nodoc: - if @perform_immediately - Concurrent::ImmediateExecutor.new - else - Concurrent::ThreadPoolExecutor.new(DEFAULT_EXECUTOR_OPTIONS) - end - end - - def enqueue(job_data, queue: 'default') #:nodoc: - QUEUES[queue].post(job_data) { |job| ActiveJob::Base.execute(job) } - end - - def enqueue_at(job_data, timestamp, queue: 'default') #:nodoc: - delay = timestamp - Time.current.to_f - if delay > 0 - Concurrent::ScheduledTask.execute(delay, args: [job_data], executor: QUEUES[queue]) do |job| - ActiveJob::Base.execute(job) - end - else - enqueue(job_data, queue: queue) - end - end - end - end -end diff --git a/activejob/lib/active_job/base.rb b/activejob/lib/active_job/base.rb index e5f09f65fb..ff5c69ddc6 100644 --- a/activejob/lib/active_job/base.rb +++ b/activejob/lib/active_job/base.rb @@ -36,7 +36,7 @@ module ActiveJob #:nodoc: # Records that are passed in are serialized/deserialized using Global # ID. More information can be found in Arguments. # - # To enqueue a job to be performed as soon the queueing system is free: + # To enqueue a job to be performed as soon as the queueing system is free: # # ProcessPhotoJob.perform_later(photo) # diff --git a/activejob/lib/active_job/gem_version.rb b/activejob/lib/active_job/gem_version.rb index 27a5de93f4..be4fabf545 100644 --- a/activejob/lib/active_job/gem_version.rb +++ b/activejob/lib/active_job/gem_version.rb @@ -8,7 +8,7 @@ module ActiveJob MAJOR = 5 MINOR = 0 TINY = 0 - PRE = "alpha" + PRE = "beta3" STRING = [MAJOR, MINOR, TINY, PRE].compact.join(".") end diff --git a/activejob/lib/active_job/queue_adapter.rb b/activejob/lib/active_job/queue_adapter.rb index 457015b741..72e4ebf935 100644 --- a/activejob/lib/active_job/queue_adapter.rb +++ b/activejob/lib/active_job/queue_adapter.rb @@ -4,25 +4,25 @@ require 'active_support/core_ext/string/inflections' module ActiveJob # The <tt>ActiveJob::QueueAdapter</tt> module is used to load the - # correct adapter. The default queue adapter is the +:inline+ queue. + # correct adapter. The default queue adapter is the +:async+ queue. module QueueAdapter #:nodoc: extend ActiveSupport::Concern included do class_attribute :_queue_adapter, instance_accessor: false, instance_predicate: false - self.queue_adapter = :inline + self.queue_adapter = :async end # Includes the setter method for changing the active queue adapter. module ClassMethods # Returns the backend queue provider. The default queue adapter - # is the +:inline+ queue. See QueueAdapters for more information. + # is the +:async+ queue. See QueueAdapters for more information. def queue_adapter _queue_adapter end # Specify the backend queue provider. The default queue adapter - # is the +:inline+ queue. See QueueAdapters for more + # is the +:async+ queue. See QueueAdapters for more # information. def queue_adapter=(name_or_adapter_or_class) self._queue_adapter = interpret_adapter(name_or_adapter_or_class) diff --git a/activejob/lib/active_job/queue_adapters.rb b/activejob/lib/active_job/queue_adapters.rb index aeb1fe1e73..2c5039ef4d 100644 --- a/activejob/lib/active_job/queue_adapters.rb +++ b/activejob/lib/active_job/queue_adapters.rb @@ -27,7 +27,7 @@ module ActiveJob # | Resque | Yes | Yes | Yes (Gem) | Queue | Global | Yes | # | Sidekiq | Yes | Yes | Yes | Queue | No | Job | # | Sneakers | Yes | Yes | No | Queue | Queue | No | - # | Sucker Punch | Yes | Yes | No | No | No | No | + # | Sucker Punch | Yes | Yes | Yes | No | No | No | # | Active Job Async | Yes | Yes | Yes | No | No | No | # | Active Job Inline | No | Yes | N/A | N/A | N/A | N/A | # diff --git a/activejob/lib/active_job/queue_adapters/async_adapter.rb b/activejob/lib/active_job/queue_adapters/async_adapter.rb index 3fc27f56e7..922bc4afce 100644 --- a/activejob/lib/active_job/queue_adapters/async_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/async_adapter.rb @@ -1,22 +1,113 @@ -require 'active_job/async_job' +require 'securerandom' +require 'concurrent/scheduled_task' +require 'concurrent/executor/thread_pool_executor' +require 'concurrent/utility/processor_counter' module ActiveJob module QueueAdapters # == Active Job Async adapter # - # When enqueueing jobs with the Async adapter the job will be executed - # asynchronously using {AsyncJob}[http://api.rubyonrails.org/classes/ActiveJob/AsyncJob.html]. + # The Async adapter runs jobs with an in-process thread pool. # - # To use +AsyncJob+ set the queue_adapter config to +:async+. + # This is the default queue adapter. It's well-suited for dev/test since + # it doesn't need an external infrastructure, but it's a poor fit for + # production since it drops pending jobs on restart. # - # Rails.application.config.active_job.queue_adapter = :async + # To use this adapter, set queue adapter to +:async+: + # + # config.active_job.queue_adapter = :async + # + # To configure the adapter's thread pool, instantiate the adapter and + # pass your own config: + # + # config.active_job.queue_adapter = ActiveJob::QueueAdapters::AsyncAdapter.new \ + # min_threads: 1, + # max_threads: 2 * Concurrent.processor_count, + # idletime: 600.seconds + # + # The adapter uses a {Concurrent Ruby}[https://github.com/ruby-concurrency/concurrent-ruby] thread pool to schedule and execute + # jobs. Since jobs share a single thread pool, long-running jobs will block + # short-lived jobs. Fine for dev/test; bad for production. class AsyncAdapter + # See {Concurrent::ThreadPoolExecutor}[http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/ThreadPoolExecutor.html] for executor options. + def initialize(**executor_options) + @scheduler = Scheduler.new(**executor_options) + end + def enqueue(job) #:nodoc: - ActiveJob::AsyncJob.enqueue(job.serialize, queue: job.queue_name) + @scheduler.enqueue JobWrapper.new(job), queue_name: job.queue_name end def enqueue_at(job, timestamp) #:nodoc: - ActiveJob::AsyncJob.enqueue_at(job.serialize, timestamp, queue: job.queue_name) + @scheduler.enqueue_at JobWrapper.new(job), timestamp, queue_name: job.queue_name + end + + # Gracefully stop processing jobs. Finishes in-progress work and handles + # any new jobs following the executor's fallback policy (`caller_runs`). + # Waits for termination by default. Pass `wait: false` to continue. + def shutdown(wait: true) #:nodoc: + @scheduler.shutdown wait: wait + end + + # Used for our test suite. + def immediate=(immediate) #:nodoc: + @scheduler.immediate = immediate + end + + # Note that we don't actually need to serialize the jobs since we're + # performing them in-process, but we do so anyway for parity with other + # adapters and deployment environments. Otherwise, serialization bugs + # may creep in undetected. + class JobWrapper #:nodoc: + def initialize(job) + job.provider_job_id = SecureRandom.uuid + @job_data = job.serialize + end + + def perform + Base.execute @job_data + end + end + + class Scheduler #:nodoc: + DEFAULT_EXECUTOR_OPTIONS = { + min_threads: 0, + max_threads: Concurrent.processor_count, + auto_terminate: true, + idletime: 60, # 1 minute + max_queue: 0, # unlimited + fallback_policy: :caller_runs # shouldn't matter -- 0 max queue + }.freeze + + attr_accessor :immediate + + def initialize(**options) + self.immediate = false + @immediate_executor = Concurrent::ImmediateExecutor.new + @async_executor = Concurrent::ThreadPoolExecutor.new(DEFAULT_EXECUTOR_OPTIONS.merge(options)) + end + + def enqueue(job, queue_name:) + executor.post(job, &:perform) + end + + def enqueue_at(job, timestamp, queue_name:) + delay = timestamp - Time.current.to_f + if delay > 0 + Concurrent::ScheduledTask.execute(delay, args: [job], executor: executor, &:perform) + else + enqueue(job, queue_name: queue_name) + end + end + + def shutdown(wait: true) + @async_executor.shutdown + @async_executor.wait_for_termination if wait + end + + def executor + immediate ? @immediate_executor : @async_executor + end end end end diff --git a/activejob/lib/active_job/queue_adapters/inline_adapter.rb b/activejob/lib/active_job/queue_adapters/inline_adapter.rb index 8ad5f4de07..0496f8449e 100644 --- a/activejob/lib/active_job/queue_adapters/inline_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/inline_adapter.rb @@ -2,7 +2,7 @@ module ActiveJob module QueueAdapters # == Active Job Inline adapter # - # When enqueueing jobs with the Inline adapter the job will be executed + # When enqueuing jobs with the Inline adapter the job will be executed # immediately. # # To use the Inline set the queue_adapter config to +:inline+. 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 c6c35f8ab4..311109e958 100644 --- a/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb @@ -5,12 +5,10 @@ module ActiveJob # == Sucker Punch adapter for Active Job # # Sucker Punch is a single-process Ruby asynchronous processing library. - # It's girl_friday and DSL sugar on top of Celluloid. With Celluloid's - # actor pattern, we can do asynchronous processing within a single process. - # This reduces costs of hosting on a service like Heroku along with the - # memory footprint of having to maintain additional jobs if hosting on - # a dedicated server. All queues can run within a single Rails/Sinatra - # process. + # This reduces the cost of of hosting on a service like Heroku along + # with the memory footprint of having to maintain additional jobs if + # hosting on a dedicated server. All queues can run within a + # single application (eg. Rails, Sinatra, etc.) process. # # Read more about Sucker Punch {here}[https://github.com/brandonhilkert/sucker_punch]. # @@ -19,11 +17,22 @@ module ActiveJob # Rails.application.config.active_job.queue_adapter = :sucker_punch class SuckerPunchAdapter def enqueue(job) #:nodoc: - JobWrapper.new.async.perform job.serialize + if JobWrapper.respond_to?(:perform_async) + # sucker_punch 2.0 API + JobWrapper.perform_async job.serialize + else + # sucker_punch 1.0 API + JobWrapper.new.async.perform job.serialize + end end def enqueue_at(job, timestamp) #:nodoc: - raise NotImplementedError, "This queueing backend does not support scheduling jobs. To see what features are supported go to http://api.rubyonrails.org/classes/ActiveJob/QueueAdapters.html" + if JobWrapper.respond_to?(:perform_in) + delay = timestamp - Time.current.to_f + JobWrapper.perform_in delay, job.serialize + else + raise NotImplementedError, 'sucker_punch 1.0 does not support `enqueued_at`. Please upgrade to version ~> 2.0.0 to enable this behavior.' + end end class JobWrapper #:nodoc: diff --git a/activejob/lib/active_job/railtie.rb b/activejob/lib/active_job/railtie.rb index 6538ac1b30..b249ec09dd 100644 --- a/activejob/lib/active_job/railtie.rb +++ b/activejob/lib/active_job/railtie.rb @@ -12,7 +12,7 @@ module ActiveJob initializer "active_job.set_configs" do |app| options = app.config.active_job - options.queue_adapter ||= :inline + options.queue_adapter ||= :async ActiveSupport.on_load(:active_job) do options.each { |k,v| send("#{k}=", v) } diff --git a/activejob/lib/active_job/test_helper.rb b/activejob/lib/active_job/test_helper.rb index 44ddfa5f69..e06b98736d 100644 --- a/activejob/lib/active_job/test_helper.rb +++ b/activejob/lib/active_job/test_helper.rb @@ -4,325 +4,321 @@ require 'active_support/core_ext/hash/keys' module ActiveJob # Provides helper methods for testing Active Job module TestHelper - extend ActiveSupport::Concern + delegate :enqueued_jobs, :enqueued_jobs=, + :performed_jobs, :performed_jobs=, + to: :queue_adapter - included do - def before_setup # :nodoc: - test_adapter = ActiveJob::QueueAdapters::TestAdapter.new + def before_setup # :nodoc: + 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 + @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 - - clear_enqueued_jobs - clear_performed_jobs - super end - def after_teardown # :nodoc: - super - @old_queue_adapters.each do |(klass, adapter)| - klass.queue_adapter = adapter - end - end + clear_enqueued_jobs + clear_performed_jobs + super + end - # Asserts that the number of enqueued jobs matches the given number. - # - # def test_jobs - # assert_enqueued_jobs 0 - # HelloJob.perform_later('david') - # assert_enqueued_jobs 1 - # HelloJob.perform_later('abdelkader') - # assert_enqueued_jobs 2 - # end - # - # If a block is passed, that block should cause the specified number of - # jobs to be enqueued. - # - # def test_jobs_again - # assert_enqueued_jobs 1 do - # HelloJob.perform_later('cristian') - # end - # - # assert_enqueued_jobs 2 do - # HelloJob.perform_later('aaron') - # HelloJob.perform_later('rafael') - # end - # end - # - # The number of times a specific job is enqueued can be asserted. - # - # def test_logging_job - # assert_enqueued_jobs 2, only: LoggingJob do - # LoggingJob.perform_later - # HelloJob.perform_later('jeremy') - # end - # end - def assert_enqueued_jobs(number, only: nil) - if block_given? - original_count = enqueued_jobs_size(only: only) - yield - new_count = enqueued_jobs_size(only: only) - assert_equal number, new_count - original_count, "#{number} jobs expected, but #{new_count - original_count} were enqueued" - else - actual_count = enqueued_jobs_size(only: only) - assert_equal number, actual_count, "#{number} jobs expected, but #{actual_count} were enqueued" - end + def after_teardown # :nodoc: + super + @old_queue_adapters.each do |(klass, adapter)| + klass.queue_adapter = adapter end + end - # Asserts that no jobs have been enqueued. - # - # def test_jobs - # assert_no_enqueued_jobs - # HelloJob.perform_later('jeremy') - # assert_enqueued_jobs 1 - # end - # - # If a block is passed, that block should not cause any job to be enqueued. - # - # def test_jobs_again - # assert_no_enqueued_jobs do - # # No job should be enqueued from this block - # end - # end - # - # It can be asserted that no jobs of a specific kind are enqueued: - # - # def test_no_logging - # assert_no_enqueued_jobs only: LoggingJob do - # HelloJob.perform_later('jeremy') - # end - # end - # - # Note: This assertion is simply a shortcut for: - # - # assert_enqueued_jobs 0, &block - def assert_no_enqueued_jobs(only: nil, &block) - assert_enqueued_jobs 0, only: only, &block + # Asserts that the number of enqueued jobs matches the given number. + # + # def test_jobs + # assert_enqueued_jobs 0 + # HelloJob.perform_later('david') + # assert_enqueued_jobs 1 + # HelloJob.perform_later('abdelkader') + # assert_enqueued_jobs 2 + # end + # + # If a block is passed, that block should cause the specified number of + # jobs to be enqueued. + # + # def test_jobs_again + # assert_enqueued_jobs 1 do + # HelloJob.perform_later('cristian') + # end + # + # assert_enqueued_jobs 2 do + # HelloJob.perform_later('aaron') + # HelloJob.perform_later('rafael') + # end + # end + # + # The number of times a specific job is enqueued can be asserted. + # + # def test_logging_job + # assert_enqueued_jobs 1, only: LoggingJob do + # LoggingJob.perform_later + # HelloJob.perform_later('jeremy') + # end + # end + def assert_enqueued_jobs(number, only: nil) + if block_given? + original_count = enqueued_jobs_size(only: only) + yield + new_count = enqueued_jobs_size(only: only) + assert_equal number, new_count - original_count, "#{number} jobs expected, but #{new_count - original_count} were enqueued" + else + actual_count = enqueued_jobs_size(only: only) + assert_equal number, actual_count, "#{number} jobs expected, but #{actual_count} were enqueued" end + end - # Asserts that the number of performed jobs matches the given number. - # If no block is passed, <tt>perform_enqueued_jobs</tt> - # must be called around the job call. - # - # def test_jobs - # assert_performed_jobs 0 - # - # perform_enqueued_jobs do - # HelloJob.perform_later('xavier') - # end - # assert_performed_jobs 1 - # - # perform_enqueued_jobs do - # HelloJob.perform_later('yves') - # assert_performed_jobs 2 - # end - # end - # - # If a block is passed, that block should cause the specified number of - # jobs to be performed. - # - # def test_jobs_again - # assert_performed_jobs 1 do - # HelloJob.perform_later('robin') - # end - # - # assert_performed_jobs 2 do - # HelloJob.perform_later('carlos') - # HelloJob.perform_later('sean') - # end - # end - # - # The block form supports filtering. If the :only option is specified, - # then only the listed job(s) will be performed. - # - # def test_hello_job - # assert_performed_jobs 1, only: HelloJob do - # HelloJob.perform_later('jeremy') - # LoggingJob.perform_later - # end - # end - # - # An array may also be specified, to support testing multiple jobs. - # - # def test_hello_and_logging_jobs - # assert_nothing_raised do - # assert_performed_jobs 2, only: [HelloJob, LoggingJob] do - # HelloJob.perform_later('jeremy') - # LoggingJob.perform_later('stewie') - # RescueJob.perform_later('david') - # end - # end - # end - def assert_performed_jobs(number, only: nil) - if block_given? - original_count = performed_jobs.size - perform_enqueued_jobs(only: only) { yield } - new_count = performed_jobs.size - assert_equal number, new_count - original_count, - "#{number} jobs expected, but #{new_count - original_count} were performed" - else - performed_jobs_size = performed_jobs.size - assert_equal number, performed_jobs_size, "#{number} jobs expected, but #{performed_jobs_size} were performed" - end - end + # Asserts that no jobs have been enqueued. + # + # def test_jobs + # assert_no_enqueued_jobs + # HelloJob.perform_later('jeremy') + # assert_enqueued_jobs 1 + # end + # + # If a block is passed, that block should not cause any job to be enqueued. + # + # def test_jobs_again + # assert_no_enqueued_jobs do + # # No job should be enqueued from this block + # end + # end + # + # It can be asserted that no jobs of a specific kind are enqueued: + # + # def test_no_logging + # assert_no_enqueued_jobs only: LoggingJob do + # HelloJob.perform_later('jeremy') + # end + # end + # + # Note: This assertion is simply a shortcut for: + # + # assert_enqueued_jobs 0, &block + def assert_no_enqueued_jobs(only: nil, &block) + assert_enqueued_jobs 0, only: only, &block + end - # Asserts that no jobs have been performed. - # - # def test_jobs - # assert_no_performed_jobs - # - # perform_enqueued_jobs do - # HelloJob.perform_later('matthew') - # assert_performed_jobs 1 - # end - # end - # - # If a block is passed, that block should not cause any job to be performed. - # - # def test_jobs_again - # assert_no_performed_jobs do - # # No job should be performed from this block - # end - # end - # - # The block form supports filtering. If the :only option is specified, - # then only the listed job(s) will be performed. - # - # def test_hello_job - # assert_performed_jobs 1, only: HelloJob do - # HelloJob.perform_later('jeremy') - # LoggingJob.perform_later - # end - # end - # - # An array may also be specified, to support testing multiple jobs. - # - # def test_hello_and_logging_jobs - # assert_nothing_raised do - # assert_performed_jobs 2, only: [HelloJob, LoggingJob] do - # HelloJob.perform_later('jeremy') - # LoggingJob.perform_later('stewie') - # RescueJob.perform_later('david') - # end - # end - # end - # - # Note: This assertion is simply a shortcut for: - # - # assert_performed_jobs 0, &block - def assert_no_performed_jobs(only: nil, &block) - assert_performed_jobs 0, only: only, &block + # Asserts that the number of performed jobs matches the given number. + # If no block is passed, <tt>perform_enqueued_jobs</tt> + # must be called around the job call. + # + # def test_jobs + # assert_performed_jobs 0 + # + # perform_enqueued_jobs do + # HelloJob.perform_later('xavier') + # end + # assert_performed_jobs 1 + # + # perform_enqueued_jobs do + # HelloJob.perform_later('yves') + # assert_performed_jobs 2 + # end + # end + # + # If a block is passed, that block should cause the specified number of + # jobs to be performed. + # + # def test_jobs_again + # assert_performed_jobs 1 do + # HelloJob.perform_later('robin') + # end + # + # assert_performed_jobs 2 do + # HelloJob.perform_later('carlos') + # HelloJob.perform_later('sean') + # end + # end + # + # The block form supports filtering. If the :only option is specified, + # then only the listed job(s) will be performed. + # + # def test_hello_job + # assert_performed_jobs 1, only: HelloJob do + # HelloJob.perform_later('jeremy') + # LoggingJob.perform_later + # end + # end + # + # An array may also be specified, to support testing multiple jobs. + # + # def test_hello_and_logging_jobs + # assert_nothing_raised do + # assert_performed_jobs 2, only: [HelloJob, LoggingJob] do + # HelloJob.perform_later('jeremy') + # LoggingJob.perform_later('stewie') + # RescueJob.perform_later('david') + # end + # end + # end + def assert_performed_jobs(number, only: nil) + if block_given? + original_count = performed_jobs.size + perform_enqueued_jobs(only: only) { yield } + new_count = performed_jobs.size + assert_equal number, new_count - original_count, + "#{number} jobs expected, but #{new_count - original_count} were performed" + else + performed_jobs_size = performed_jobs.size + assert_equal number, performed_jobs_size, "#{number} jobs expected, but #{performed_jobs_size} were performed" end + end - # Asserts that the job passed in the block has been enqueued with the given arguments. - # - # def test_assert_enqueued_with - # assert_enqueued_with(job: MyJob, args: [1,2,3], queue: 'low') do - # MyJob.perform_later(1,2,3) - # end - # - # assert_enqueued_with(job: MyJob, at: Date.tomorrow.noon) do - # MyJob.set(wait_until: Date.tomorrow.noon).perform_later - # end - # end - def assert_enqueued_with(args = {}) - original_enqueued_jobs_count = enqueued_jobs.count - args.assert_valid_keys(:job, :args, :at, :queue) - serialized_args = serialize_args_for_assertion(args) - yield - in_block_jobs = enqueued_jobs.drop(original_enqueued_jobs_count) - matching_job = in_block_jobs.find do |job| - serialized_args.all? { |key, value| value == job[key] } - end - assert matching_job, "No enqueued job found with #{args}" - instantiate_job(matching_job) - end + # Asserts that no jobs have been performed. + # + # def test_jobs + # assert_no_performed_jobs + # + # perform_enqueued_jobs do + # HelloJob.perform_later('matthew') + # assert_performed_jobs 1 + # end + # end + # + # If a block is passed, that block should not cause any job to be performed. + # + # def test_jobs_again + # assert_no_performed_jobs do + # # No job should be performed from this block + # end + # end + # + # The block form supports filtering. If the :only option is specified, + # then only the listed job(s) will be performed. + # + # def test_hello_job + # assert_performed_jobs 1, only: HelloJob do + # HelloJob.perform_later('jeremy') + # LoggingJob.perform_later + # end + # end + # + # An array may also be specified, to support testing multiple jobs. + # + # def test_hello_and_logging_jobs + # assert_nothing_raised do + # assert_performed_jobs 2, only: [HelloJob, LoggingJob] do + # HelloJob.perform_later('jeremy') + # LoggingJob.perform_later('stewie') + # RescueJob.perform_later('david') + # end + # end + # end + # + # Note: This assertion is simply a shortcut for: + # + # assert_performed_jobs 0, &block + def assert_no_performed_jobs(only: nil, &block) + assert_performed_jobs 0, only: only, &block + end - # Asserts that the job passed in the block has been performed with the given arguments. - # - # def test_assert_performed_with - # assert_performed_with(job: MyJob, args: [1,2,3], queue: 'high') do - # MyJob.perform_later(1,2,3) - # end - # - # assert_performed_with(job: MyJob, at: Date.tomorrow.noon) do - # MyJob.set(wait_until: Date.tomorrow.noon).perform_later - # end - # end - def assert_performed_with(args = {}) - original_performed_jobs_count = performed_jobs.count - args.assert_valid_keys(:job, :args, :at, :queue) - serialized_args = serialize_args_for_assertion(args) - perform_enqueued_jobs { yield } - in_block_jobs = performed_jobs.drop(original_performed_jobs_count) - matching_job = in_block_jobs.find do |job| - serialized_args.all? { |key, value| value == job[key] } - end - assert matching_job, "No performed job found with #{args}" - instantiate_job(matching_job) + # Asserts that the job passed in the block has been enqueued with the given arguments. + # + # def test_assert_enqueued_with + # assert_enqueued_with(job: MyJob, args: [1,2,3], queue: 'low') do + # MyJob.perform_later(1,2,3) + # end + # + # assert_enqueued_with(job: MyJob, at: Date.tomorrow.noon) do + # MyJob.set(wait_until: Date.tomorrow.noon).perform_later + # end + # end + def assert_enqueued_with(args = {}) + original_enqueued_jobs_count = enqueued_jobs.count + args.assert_valid_keys(:job, :args, :at, :queue) + serialized_args = serialize_args_for_assertion(args) + yield + in_block_jobs = enqueued_jobs.drop(original_enqueued_jobs_count) + matching_job = in_block_jobs.find do |job| + serialized_args.all? { |key, value| value == job[key] } end + assert matching_job, "No enqueued job found with #{args}" + instantiate_job(matching_job) + 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 - 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 + # Asserts that the job passed in the block has been performed with the given arguments. + # + # def test_assert_performed_with + # assert_performed_with(job: MyJob, args: [1,2,3], queue: 'high') do + # MyJob.perform_later(1,2,3) + # end + # + # assert_performed_with(job: MyJob, at: Date.tomorrow.noon) do + # MyJob.set(wait_until: Date.tomorrow.noon).perform_later + # end + # end + def assert_performed_with(args = {}) + original_performed_jobs_count = performed_jobs.count + args.assert_valid_keys(:job, :args, :at, :queue) + serialized_args = serialize_args_for_assertion(args) + perform_enqueued_jobs { yield } + in_block_jobs = performed_jobs.drop(original_performed_jobs_count) + matching_job = in_block_jobs.find do |job| + serialized_args.all? { |key, value| value == job[key] } end + assert matching_job, "No performed job found with #{args}" + instantiate_job(matching_job) + 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 + old_filter = queue_adapter.filter - def queue_adapter - ActiveJob::Base.queue_adapter + 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 - delegate :enqueued_jobs, :enqueued_jobs=, - :performed_jobs, :performed_jobs=, - to: :queue_adapter + def queue_adapter + ActiveJob::Base.queue_adapter + end - private - def clear_enqueued_jobs # :nodoc: - enqueued_jobs.clear - end + private + def clear_enqueued_jobs # :nodoc: + enqueued_jobs.clear + end - def clear_performed_jobs # :nodoc: - performed_jobs.clear - end + def clear_performed_jobs # :nodoc: + performed_jobs.clear + end - def enqueued_jobs_size(only: nil) # :nodoc: - if only - enqueued_jobs.count { |job| Array(only).include?(job.fetch(:job)) } - else - enqueued_jobs.count - end + def enqueued_jobs_size(only: nil) # :nodoc: + if only + enqueued_jobs.count { |job| Array(only).include?(job.fetch(:job)) } + else + enqueued_jobs.count end + end - def serialize_args_for_assertion(args) # :nodoc: - args.dup.tap do |serialized_args| - serialized_args[:args] = ActiveJob::Arguments.serialize(serialized_args[:args]) if serialized_args[:args] - serialized_args[:at] = serialized_args[:at].to_f if serialized_args[:at] - end + def serialize_args_for_assertion(args) # :nodoc: + args.dup.tap do |serialized_args| + serialized_args[:args] = ActiveJob::Arguments.serialize(serialized_args[:args]) if serialized_args[:args] + serialized_args[:at] = serialized_args[:at].to_f if serialized_args[:at] end + end - def instantiate_job(payload) # :nodoc: - job = payload[:job].new(*payload[:args]) - job.scheduled_at = Time.at(payload[:at]) if payload.key?(:at) - job.queue_name = payload[:queue] - job - end - end + def instantiate_job(payload) # :nodoc: + job = payload[:job].new(*payload[:args]) + job.scheduled_at = Time.at(payload[:at]) if payload.key?(:at) + job.queue_name = payload[:queue] + job + end end end |