diff options
Diffstat (limited to 'activejob/lib/active_job')
-rw-r--r-- | activejob/lib/active_job/base.rb | 1 | ||||
-rw-r--r-- | activejob/lib/active_job/core.rb | 20 | ||||
-rw-r--r-- | activejob/lib/active_job/exceptions.rb | 24 | ||||
-rw-r--r-- | activejob/lib/active_job/logging.rb | 10 | ||||
-rw-r--r-- | activejob/lib/active_job/queue_adapters.rb | 4 | ||||
-rw-r--r-- | activejob/lib/active_job/queue_adapters/inline_adapter.rb | 2 | ||||
-rw-r--r-- | activejob/lib/active_job/queue_adapters/qu_adapter.rb | 46 | ||||
-rw-r--r-- | activejob/lib/active_job/serializers.rb | 1 | ||||
-rw-r--r-- | activejob/lib/active_job/serializers/object_serializer.rb | 2 | ||||
-rw-r--r-- | activejob/lib/active_job/test_helper.rb | 44 |
10 files changed, 72 insertions, 82 deletions
diff --git a/activejob/lib/active_job/base.rb b/activejob/lib/active_job/base.rb index 2b2a59e969..95b1062701 100644 --- a/activejob/lib/active_job/base.rb +++ b/activejob/lib/active_job/base.rb @@ -60,7 +60,6 @@ module ActiveJob #:nodoc: # * SerializationError - Error class for serialization errors. class Base include Core - include Serializers include QueueAdapter include QueueName include QueuePriority diff --git a/activejob/lib/active_job/core.rb b/activejob/lib/active_job/core.rb index da841ae45b..61d402cfca 100644 --- a/activejob/lib/active_job/core.rb +++ b/activejob/lib/active_job/core.rb @@ -88,7 +88,7 @@ module ActiveJob "provider_job_id" => provider_job_id, "queue_name" => queue_name, "priority" => priority, - "arguments" => serialize_arguments(arguments), + "arguments" => serialize_arguments_if_needed(arguments), "executions" => executions, "locale" => I18n.locale.to_s, "timezone" => Time.zone.try(:name) @@ -133,19 +133,31 @@ module ActiveJob end private + def serialize_arguments_if_needed(arguments) + if arguments_serialized? + @serialized_arguments + else + serialize_arguments(arguments) + end + end + def deserialize_arguments_if_needed - if defined?(@serialized_arguments) && @serialized_arguments.present? + if arguments_serialized? @arguments = deserialize_arguments(@serialized_arguments) @serialized_arguments = nil end end - def serialize_arguments(serialized_args) - Arguments.serialize(serialized_args) + def serialize_arguments(arguments) + Arguments.serialize(arguments) end def deserialize_arguments(serialized_args) Arguments.deserialize(serialized_args) end + + def arguments_serialized? + defined?(@serialized_arguments) && @serialized_arguments + end end end diff --git a/activejob/lib/active_job/exceptions.rb b/activejob/lib/active_job/exceptions.rb index ae700848d0..1e57dbcb1c 100644 --- a/activejob/lib/active_job/exceptions.rb +++ b/activejob/lib/active_job/exceptions.rb @@ -30,8 +30,8 @@ module ActiveJob # class RemoteServiceJob < ActiveJob::Base # retry_on CustomAppException # defaults to 3s wait, 5 attempts # retry_on AnotherCustomAppException, wait: ->(executions) { executions * 2 } - # retry_on(YetAnotherCustomAppException) do |job, exception| - # ExceptionNotifier.caught(exception) + # retry_on(YetAnotherCustomAppException) do |job, error| + # ExceptionNotifier.caught(error) # end # retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3 # retry_on Net::OpenTimeout, wait: :exponentially_longer, attempts: 10 @@ -42,16 +42,16 @@ module ActiveJob # # Might raise Net::OpenTimeout when the remote service is down # end # end - def retry_on(exception, wait: 3.seconds, attempts: 5, queue: nil, priority: nil) - rescue_from exception do |error| + def retry_on(*exceptions, wait: 3.seconds, attempts: 5, queue: nil, priority: nil) + rescue_from(*exceptions) do |error| if executions < attempts - logger.error "Retrying #{self.class} in #{wait} seconds, due to a #{exception}. The original exception was #{error.cause.inspect}." + logger.error "Retrying #{self.class} in #{wait} seconds, due to a #{error.class}. The original exception was #{error.cause.inspect}." retry_job wait: determine_delay(wait), queue: queue, priority: priority else if block_given? yield self, error else - logger.error "Stopped retrying #{self.class} due to a #{exception}, which reoccurred on #{executions} attempts. The original exception was #{error.cause.inspect}." + logger.error "Stopped retrying #{self.class} due to a #{error.class}, which reoccurred on #{executions} attempts. The original exception was #{error.cause.inspect}." raise error end end @@ -67,8 +67,8 @@ module ActiveJob # # class SearchIndexingJob < ActiveJob::Base # discard_on ActiveJob::DeserializationError - # discard_on(CustomAppException) do |job, exception| - # ExceptionNotifier.caught(exception) + # discard_on(CustomAppException) do |job, error| + # ExceptionNotifier.caught(error) # end # # def perform(record) @@ -76,12 +76,12 @@ module ActiveJob # # Might raise CustomAppException for something domain specific # end # end - def discard_on(exception) - rescue_from exception do |error| + def discard_on(*exceptions) + rescue_from(*exceptions) do |error| if block_given? - yield self, exception + yield self, error else - logger.error "Discarded #{self.class} due to a #{exception}. The original exception was #{error.cause.inspect}." + logger.error "Discarded #{self.class} due to a #{error.class}. The original exception was #{error.cause.inspect}." end end end diff --git a/activejob/lib/active_job/logging.rb b/activejob/lib/active_job/logging.rb index 3312857ac7..9ffd60ad53 100644 --- a/activejob/lib/active_job/logging.rb +++ b/activejob/lib/active_job/logging.rb @@ -27,13 +27,13 @@ module ActiveJob end end - after_enqueue do |job| + around_enqueue do |job, block| if job.scheduled_at - ActiveSupport::Notifications.instrument "enqueue_at.active_job", - adapter: job.class.queue_adapter, job: job + ActiveSupport::Notifications.instrument("enqueue_at.active_job", + adapter: job.class.queue_adapter, job: job, &block) else - ActiveSupport::Notifications.instrument "enqueue.active_job", - adapter: job.class.queue_adapter, job: job + ActiveSupport::Notifications.instrument("enqueue.active_job", + adapter: job.class.queue_adapter, job: job, &block) end end end diff --git a/activejob/lib/active_job/queue_adapters.rb b/activejob/lib/active_job/queue_adapters.rb index c1a1d3c510..00c7b407b1 100644 --- a/activejob/lib/active_job/queue_adapters.rb +++ b/activejob/lib/active_job/queue_adapters.rb @@ -7,7 +7,6 @@ module ActiveJob # # * {Backburner}[https://github.com/nesquena/backburner] # * {Delayed Job}[https://github.com/collectiveidea/delayed_job] - # * {Qu}[https://github.com/bkeepers/qu] # * {Que}[https://github.com/chanks/que] # * {queue_classic}[https://github.com/QueueClassic/queue_classic] # * {Resque}[https://github.com/resque/resque] @@ -16,6 +15,7 @@ module ActiveJob # * {Sucker Punch}[https://github.com/brandonhilkert/sucker_punch] # * {Active Job Async Job}[http://api.rubyonrails.org/classes/ActiveJob/QueueAdapters/AsyncAdapter.html] # * {Active Job Inline}[http://api.rubyonrails.org/classes/ActiveJob/QueueAdapters/InlineAdapter.html] + # * Please Note: We are not accepting pull requests for new adapters. See the {README}[link:files/activejob/README_md.html] for more details. # # === Backends Features # @@ -23,7 +23,6 @@ module ActiveJob # |-------------------|-------|--------|------------|------------|---------|---------| # | Backburner | Yes | Yes | Yes | Yes | Job | Global | # | Delayed Job | Yes | Yes | Yes | Job | Global | Global | - # | Qu | Yes | Yes | No | No | No | Global | # | Que | Yes | Yes | Yes | Job | No | Job | # | queue_classic | Yes | Yes | Yes* | No | No | No | # | Resque | Yes | Yes | Yes (Gem) | Queue | Global | Yes | @@ -114,7 +113,6 @@ module ActiveJob autoload :InlineAdapter autoload :BackburnerAdapter autoload :DelayedJobAdapter - autoload :QuAdapter autoload :QueAdapter autoload :QueueClassicAdapter autoload :ResqueAdapter diff --git a/activejob/lib/active_job/queue_adapters/inline_adapter.rb b/activejob/lib/active_job/queue_adapters/inline_adapter.rb index 3d0b590212..ca04dc943c 100644 --- a/activejob/lib/active_job/queue_adapters/inline_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/inline_adapter.rb @@ -16,7 +16,7 @@ module ActiveJob end def enqueue_at(*) #:nodoc: - raise NotImplementedError, "Use a queueing backend to enqueue jobs in the future. Read more at http://guides.rubyonrails.org/active_job_basics.html" + raise NotImplementedError, "Use a queueing backend to enqueue jobs in the future. Read more at https://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 deleted file mode 100644 index bd7003e177..0000000000 --- a/activejob/lib/active_job/queue_adapters/qu_adapter.rb +++ /dev/null @@ -1,46 +0,0 @@ -# frozen_string_literal: true - -require "qu" - -module ActiveJob - module QueueAdapters - # == Qu adapter for Active Job - # - # Qu is a Ruby library for queuing and processing background jobs. It is - # heavily inspired by delayed_job and Resque. Qu was created to overcome - # some shortcomings in the existing queuing libraries. - # The advantages of Qu are: Multiple backends (redis, mongo), jobs are - # requeued when worker is killed, resque-like API. - # - # Read more about Qu {here}[https://github.com/bkeepers/qu]. - # - # To use Qu set the queue_adapter config to +:qu+. - # - # Rails.application.config.active_job.queue_adapter = :qu - class QuAdapter - def enqueue(job, *args) #:nodoc: - qu_job = Qu::Payload.new(klass: JobWrapper, args: [job.serialize]).tap do |payload| - payload.instance_variable_set(:@queue, job.queue_name) - end.push - - # qu_job can be nil depending on the configured backend - job.provider_job_id = qu_job.id unless qu_job.nil? - qu_job - end - - def enqueue_at(job, timestamp, *args) #: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" - end - - class JobWrapper < Qu::Job #:nodoc: - def initialize(job_data) - @job_data = job_data - end - - def perform - Base.execute @job_data - end - end - end - end -end diff --git a/activejob/lib/active_job/serializers.rb b/activejob/lib/active_job/serializers.rb index df66e66659..a5d90f48b8 100644 --- a/activejob/lib/active_job/serializers.rb +++ b/activejob/lib/active_job/serializers.rb @@ -7,7 +7,6 @@ module ActiveJob # and to add new ones. It also has helpers to serialize/deserialize objects. module Serializers # :nodoc: extend ActiveSupport::Autoload - extend ActiveSupport::Concern autoload :ObjectSerializer autoload :SymbolSerializer diff --git a/activejob/lib/active_job/serializers/object_serializer.rb b/activejob/lib/active_job/serializers/object_serializer.rb index 1dfd1e44be..6d280969be 100644 --- a/activejob/lib/active_job/serializers/object_serializer.rb +++ b/activejob/lib/active_job/serializers/object_serializer.rb @@ -38,7 +38,7 @@ module ActiveJob { Arguments::OBJECT_SERIALIZER_KEY => self.class.name }.merge!(hash) end - # Deserilizes an argument form a JSON primiteve type. + # Deserializes an argument from a JSON primitive type. def deserialize(_argument) raise NotImplementedError end diff --git a/activejob/lib/active_job/test_helper.rb b/activejob/lib/active_job/test_helper.rb index 1cd2c40c15..04cde28a96 100644 --- a/activejob/lib/active_job/test_helper.rb +++ b/activejob/lib/active_job/test_helper.rb @@ -159,11 +159,19 @@ module ActiveJob # end # end # + # It can be asserted that no jobs are enqueued to a specific queue: + # + # def test_no_logging + # assert_no_enqueued_jobs queue: 'default' do + # LoggingJob.set(queue: :some_queue).perform_later + # end + # end + # # Note: This assertion is simply a shortcut for: # # assert_enqueued_jobs 0, &block - def assert_no_enqueued_jobs(only: nil, except: nil, &block) - assert_enqueued_jobs 0, only: only, except: except, &block + def assert_no_enqueued_jobs(only: nil, except: nil, queue: nil, &block) + assert_enqueued_jobs 0, only: only, except: except, queue: queue, &block end # Asserts that the number of performed jobs matches the given number. @@ -286,7 +294,18 @@ module ActiveJob assert_performed_jobs 0, only: only, except: except, &block end - # Asserts that the job passed in the block has been enqueued with the given arguments. + # Asserts that the job has been enqueued with the given arguments. + # + # def test_assert_enqueued_with + # MyJob.perform_later(1,2,3) + # assert_enqueued_with(job: MyJob, args: [1,2,3], queue: 'low') + # + # MyJob.set(wait_until: Date.tomorrow.noon).perform_later + # assert_enqueued_with(job: MyJob, at: Date.tomorrow.noon) + # end + # + # If a block is passed, that block should cause the job to be + # enqueued with the given arguments. # # def test_assert_enqueued_with # assert_enqueued_with(job: MyJob, args: [1,2,3], queue: 'low') do @@ -298,14 +317,23 @@ module ActiveJob # end # end def assert_enqueued_with(job: nil, args: nil, at: nil, queue: nil) - original_enqueued_jobs_count = enqueued_jobs.count expected = { job: job, args: args, at: at, queue: queue }.compact serialized_args = serialize_args_for_assertion(expected) - yield - in_block_jobs = enqueued_jobs.drop(original_enqueued_jobs_count) - matching_job = in_block_jobs.find do |in_block_job| - serialized_args.all? { |key, value| value == in_block_job[key] } + + if block_given? + original_enqueued_jobs_count = enqueued_jobs.count + + yield + + jobs = enqueued_jobs.drop(original_enqueued_jobs_count) + else + jobs = enqueued_jobs end + + matching_job = jobs.find do |enqueued_job| + serialized_args.all? { |key, value| value == enqueued_job[key] } + end + assert matching_job, "No enqueued job found with #{expected}" instantiate_job(matching_job) end |