diff options
Diffstat (limited to 'activejob/lib/active_job')
31 files changed, 2020 insertions, 0 deletions
diff --git a/activejob/lib/active_job/arguments.rb b/activejob/lib/active_job/arguments.rb new file mode 100644 index 0000000000..8e462bfe5d --- /dev/null +++ b/activejob/lib/active_job/arguments.rb @@ -0,0 +1,158 @@ +require 'active_support/core_ext/hash' + +module ActiveJob + # Raised when an exception is raised during job arguments deserialization. + # + # Wraps the original exception raised as +original_exception+. + class DeserializationError < StandardError + # The original exception that was raised during deserialization of job + # arguments. + attr_reader :original_exception + + def initialize(e) #:nodoc: + super("Error while trying to deserialize arguments: #{e.message}") + @original_exception = e + set_backtrace e.backtrace + end + end + + # 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). + # 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. + class SerializationError < ArgumentError; end + + module Arguments + extend self + # :nodoc: + TYPE_WHITELIST = [ NilClass, Fixnum, Float, String, TrueClass, FalseClass, Bignum ] + + # Serializes a set of arguments. Whitelisted types are returned + # as-is. Arrays/Hashes are serialized element by element. + # All other types are serialized using GlobalID. + def serialize(arguments) + arguments.map { |argument| serialize_argument(argument) } + end + + # Deserializes a set of arguments. Whitelisted types are returned + # as-is. Arrays/Hashes are deserialized element by element. + # All other types are deserialized using GlobalID. + def deserialize(arguments) + arguments.map { |argument| deserialize_argument(argument) } + rescue => e + raise DeserializationError.new(e) + end + + private + # :nodoc: + GLOBALID_KEY = '_aj_globalid'.freeze + # :nodoc: + SYMBOL_KEYS_KEY = '_aj_symbol_keys'.freeze + # :nodoc: + WITH_INDIFFERENT_ACCESS_KEY = '_aj_hash_with_indifferent_access'.freeze + private_constant :GLOBALID_KEY, :SYMBOL_KEYS_KEY, :WITH_INDIFFERENT_ACCESS_KEY + + def serialize_argument(argument) + case argument + when *TYPE_WHITELIST + argument + when GlobalID::Identification + convert_to_global_id_hash(argument) + when Array + argument.map { |arg| serialize_argument(arg) } + when ActiveSupport::HashWithIndifferentAccess + result = serialize_hash(argument) + result[WITH_INDIFFERENT_ACCESS_KEY] = serialize_argument(true) + result + when Hash + symbol_keys = argument.each_key.grep(Symbol).map(&:to_s) + result = serialize_hash(argument) + result[SYMBOL_KEYS_KEY] = symbol_keys + result + else + raise SerializationError.new("Unsupported argument type: #{argument.class.name}") + end + end + + def deserialize_argument(argument) + case argument + when String + GlobalID::Locator.locate(argument) || argument + when *TYPE_WHITELIST + argument + when Array + argument.map { |arg| deserialize_argument(arg) } + when Hash + if serialized_global_id?(argument) + deserialize_global_id argument + else + deserialize_hash(argument) + end + else + raise ArgumentError, "Can only deserialize primitive arguments: #{argument.inspect}" + end + end + + def serialized_global_id?(hash) + hash.size == 1 and hash.include?(GLOBALID_KEY) + end + + def deserialize_global_id(hash) + GlobalID::Locator.locate hash[GLOBALID_KEY] + end + + def serialize_hash(argument) + argument.each_with_object({}) do |(key, value), hash| + hash[serialize_hash_key(key)] = serialize_argument(value) + end + end + + def deserialize_hash(serialized_hash) + result = serialized_hash.transform_values { |v| deserialize_argument(v) } + if result.delete(WITH_INDIFFERENT_ACCESS_KEY) + result = result.with_indifferent_access + elsif symbol_keys = result.delete(SYMBOL_KEYS_KEY) + result = transform_symbol_keys(result, symbol_keys) + end + result + end + + # :nodoc: + RESERVED_KEYS = [ + GLOBALID_KEY, GLOBALID_KEY.to_sym, + SYMBOL_KEYS_KEY, SYMBOL_KEYS_KEY.to_sym, + WITH_INDIFFERENT_ACCESS_KEY, WITH_INDIFFERENT_ACCESS_KEY.to_sym, + ] + private_constant :RESERVED_KEYS + + def serialize_hash_key(key) + case key + when *RESERVED_KEYS + raise SerializationError.new("Can't serialize a Hash with reserved key #{key.inspect}") + when String, Symbol + key.to_s + else + raise SerializationError.new("Only string and symbol hash keys may be serialized as job arguments, but #{key.inspect} is a #{key.class}") + end + end + + def transform_symbol_keys(hash, symbol_keys) + hash.transform_keys do |key| + if symbol_keys.include?(key) + key.to_sym + else + key + end + end + end + + def convert_to_global_id_hash(argument) + { GLOBALID_KEY => argument.to_global_id.to_s } + rescue URI::GID::MissingModelIdError + raise SerializationError, "Unable to serialize #{argument.class} " \ + "without an id. (Maybe you forgot to call save?)" + end + end +end diff --git a/activejob/lib/active_job/async_job.rb b/activejob/lib/active_job/async_job.rb new file mode 100644 index 0000000000..6c1c070994 --- /dev/null +++ b/activejob/lib/active_job/async_job.rb @@ -0,0 +1,74 @@ +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 new file mode 100644 index 0000000000..e5f09f65fb --- /dev/null +++ b/activejob/lib/active_job/base.rb @@ -0,0 +1,70 @@ +require 'active_job/core' +require 'active_job/queue_adapter' +require 'active_job/queue_name' +require 'active_job/queue_priority' +require 'active_job/enqueuing' +require 'active_job/execution' +require 'active_job/callbacks' +require 'active_job/logging' +require 'active_job/translation' + +module ActiveJob #:nodoc: + # = Active Job + # + # Active Job objects can be configured to work with different backend + # queuing frameworks. To specify a queue adapter to use: + # + # ActiveJob::Base.queue_adapter = :inline + # + # A list of supported adapters can be found in QueueAdapters. + # + # Active Job objects can be defined by creating a class that inherits + # from the ActiveJob::Base class. The only necessary method to + # implement is the "perform" method. + # + # To define an Active Job object: + # + # class ProcessPhotoJob < ActiveJob::Base + # def perform(photo) + # photo.watermark!('Rails') + # photo.rotate!(90.degrees) + # photo.resize_to_fit!(300, 300) + # photo.upload! + # end + # end + # + # 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: + # + # ProcessPhotoJob.perform_later(photo) + # + # To enqueue a job to be processed at some point in the future: + # + # ProcessPhotoJob.set(wait_until: Date.tomorrow.noon).perform_later(photo) + # + # More information can be found in ActiveJob::Core::ClassMethods#set + # + # A job can also be processed immediately without sending to the queue: + # + # ProcessPhotoJob.perform_now(photo) + # + # == Exceptions + # + # * DeserializationError - Error class for deserialization errors. + # * SerializationError - Error class for serialization errors. + class Base + include Core + include QueueAdapter + include QueueName + include QueuePriority + include Enqueuing + include Execution + include Callbacks + include Logging + include Translation + + ActiveSupport.run_load_hooks(:active_job, self) + end +end diff --git a/activejob/lib/active_job/callbacks.rb b/activejob/lib/active_job/callbacks.rb new file mode 100644 index 0000000000..2b6149e84e --- /dev/null +++ b/activejob/lib/active_job/callbacks.rb @@ -0,0 +1,146 @@ +require 'active_support/callbacks' + +module ActiveJob + # = Active Job Callbacks + # + # Active Job provides hooks during the life cycle of a job. Callbacks allow you + # to trigger logic during the life cycle of a job. Available callbacks are: + # + # * <tt>before_enqueue</tt> + # * <tt>around_enqueue</tt> + # * <tt>after_enqueue</tt> + # * <tt>before_perform</tt> + # * <tt>around_perform</tt> + # * <tt>after_perform</tt> + # + module Callbacks + extend ActiveSupport::Concern + include ActiveSupport::Callbacks + + included do + define_callbacks :perform + define_callbacks :enqueue + end + + # These methods will be included into any Active Job object, adding + # callbacks for +perform+ and +enqueue+ methods. + module ClassMethods + # Defines a callback that will get called right before the + # job's perform method is executed. + # + # class VideoProcessJob < ActiveJob::Base + # queue_as :default + # + # before_perform do |job| + # UserMailer.notify_video_started_processing(job.arguments.first) + # end + # + # def perform(video_id) + # Video.find(video_id).process + # end + # end + # + def before_perform(*filters, &blk) + set_callback(:perform, :before, *filters, &blk) + end + + # Defines a callback that will get called right after the + # job's perform method has finished. + # + # class VideoProcessJob < ActiveJob::Base + # queue_as :default + # + # after_perform do |job| + # UserMailer.notify_video_processed(job.arguments.first) + # end + # + # def perform(video_id) + # Video.find(video_id).process + # end + # end + # + def after_perform(*filters, &blk) + set_callback(:perform, :after, *filters, &blk) + end + + # Defines a callback that will get called around the job's perform method. + # + # class VideoProcessJob < ActiveJob::Base + # queue_as :default + # + # around_perform do |job, block| + # UserMailer.notify_video_started_processing(job.arguments.first) + # block.call + # UserMailer.notify_video_processed(job.arguments.first) + # end + # + # def perform(video_id) + # Video.find(video_id).process + # end + # end + # + def around_perform(*filters, &blk) + set_callback(:perform, :around, *filters, &blk) + end + + # Defines a callback that will get called right before the + # job is enqueued. + # + # class VideoProcessJob < ActiveJob::Base + # queue_as :default + # + # before_enqueue do |job| + # $statsd.increment "enqueue-video-job.try" + # end + # + # def perform(video_id) + # Video.find(video_id).process + # end + # end + # + def before_enqueue(*filters, &blk) + set_callback(:enqueue, :before, *filters, &blk) + end + + # Defines a callback that will get called right after the + # job is enqueued. + # + # class VideoProcessJob < ActiveJob::Base + # queue_as :default + # + # after_enqueue do |job| + # $statsd.increment "enqueue-video-job.success" + # end + # + # def perform(video_id) + # Video.find(video_id).process + # end + # end + # + def after_enqueue(*filters, &blk) + set_callback(:enqueue, :after, *filters, &blk) + end + + # Defines a callback that will get called before and after the + # job is enqueued. + # + # class VideoProcessJob < ActiveJob::Base + # queue_as :default + # + # around_enqueue do |job, block| + # $statsd.time "video-job.process" do + # block.call + # end + # end + # + # def perform(video_id) + # Video.find(video_id).process + # end + # end + # + def around_enqueue(*filters, &blk) + set_callback(:enqueue, :around, *filters, &blk) + end + end + end +end diff --git a/activejob/lib/active_job/configured_job.rb b/activejob/lib/active_job/configured_job.rb new file mode 100644 index 0000000000..979280b910 --- /dev/null +++ b/activejob/lib/active_job/configured_job.rb @@ -0,0 +1,16 @@ +module ActiveJob + class ConfiguredJob #:nodoc: + def initialize(job_class, options={}) + @options = options + @job_class = job_class + end + + def perform_now(*args) + @job_class.new(*args).perform_now + end + + def perform_later(*args) + @job_class.new(*args).enqueue @options + end + end +end diff --git a/activejob/lib/active_job/core.rb b/activejob/lib/active_job/core.rb new file mode 100644 index 0000000000..19b900a285 --- /dev/null +++ b/activejob/lib/active_job/core.rb @@ -0,0 +1,130 @@ +module ActiveJob + # Provides general behavior that will be included into every Active Job + # object that inherits from ActiveJob::Base. + module Core + extend ActiveSupport::Concern + + included do + # Job arguments + attr_accessor :arguments + attr_writer :serialized_arguments + + # Timestamp when the job should be performed + attr_accessor :scheduled_at + + # Job Identifier + attr_accessor :job_id + + # Queue in which the job will reside. + attr_writer :queue_name + + # Priority that the job will have (lower is more priority). + attr_writer :priority + + # ID optionally provided by adapter + attr_accessor :provider_job_id + + # I18n.locale to be used during the job. + attr_accessor :locale + end + + # These methods will be included into any Active Job object, adding + # helpers for de/serialization and creation of job instances. + module ClassMethods + # Creates a new job instance from a hash created with +serialize+ + def deserialize(job_data) + job = job_data['job_class'].constantize.new + job.deserialize(job_data) + job + end + + # Creates a job preconfigured with the given options. You can call + # perform_later with the job arguments to enqueue the job with the + # preconfigured options + # + # ==== Options + # * <tt>:wait</tt> - Enqueues the job with the specified delay + # * <tt>:wait_until</tt> - Enqueues the job at the time specified + # * <tt>:queue</tt> - Enqueues the job on the specified queue + # * <tt>:priority</tt> - Enqueues the job with the specified priority + # + # ==== Examples + # + # VideoJob.set(queue: :some_queue).perform_later(Video.last) + # VideoJob.set(wait: 5.minutes).perform_later(Video.last) + # VideoJob.set(wait_until: Time.now.tomorrow).perform_later(Video.last) + # VideoJob.set(queue: :some_queue, wait: 5.minutes).perform_later(Video.last) + # VideoJob.set(queue: :some_queue, wait_until: Time.now.tomorrow).perform_later(Video.last) + # VideoJob.set(queue: :some_queue, wait: 5.minutes, priority: 10).perform_later(Video.last) + def set(options={}) + ConfiguredJob.new(self, options) + end + end + + # Creates a new job instance. Takes the arguments that will be + # passed to the perform method. + def initialize(*arguments) + @arguments = arguments + @job_id = SecureRandom.uuid + @queue_name = self.class.queue_name + @priority = self.class.priority + end + + # Returns a hash with the job data that can safely be passed to the + # queueing adapter. + def serialize + { + 'job_class' => self.class.name, + 'job_id' => job_id, + 'queue_name' => queue_name, + 'priority' => priority, + 'arguments' => serialize_arguments(arguments), + 'locale' => I18n.locale + } + end + + # Attaches the stored job data to the current instance. Receives a hash + # returned from +serialize+ + # + # ==== Examples + # + # class DeliverWebhookJob < ActiveJob::Base + # def serialize + # super.merge('attempt_number' => (@attempt_number || 0) + 1) + # end + # + # def deserialize(job_data) + # super + # @attempt_number = job_data['attempt_number'] + # end + # + # rescue_from(TimeoutError) do |exception| + # raise exception if @attempt_number > 5 + # retry_job(wait: 10) + # end + # end + def deserialize(job_data) + self.job_id = job_data['job_id'] + self.queue_name = job_data['queue_name'] + self.priority = job_data['priority'] + self.serialized_arguments = job_data['arguments'] + self.locale = job_data['locale'] || I18n.locale + end + + private + def deserialize_arguments_if_needed + if defined?(@serialized_arguments) && @serialized_arguments.present? + @arguments = deserialize_arguments(@serialized_arguments) + @serialized_arguments = nil + end + end + + def serialize_arguments(serialized_args) + Arguments.serialize(serialized_args) + end + + def deserialize_arguments(serialized_args) + Arguments.deserialize(serialized_args) + end + end +end diff --git a/activejob/lib/active_job/enqueuing.rb b/activejob/lib/active_job/enqueuing.rb new file mode 100644 index 0000000000..22154457fd --- /dev/null +++ b/activejob/lib/active_job/enqueuing.rb @@ -0,0 +1,82 @@ +require 'active_job/arguments' + +module ActiveJob + # Provides behavior for enqueuing and retrying jobs. + module Enqueuing + extend ActiveSupport::Concern + + # Includes the +perform_later+ method for job initialization. + module ClassMethods + # Push a job onto the queue. The arguments must be legal JSON types + # (string, int, float, nil, true, false, hash or array) or + # GlobalID::Identification instances. Arbitrary Ruby objects + # are not supported. + # + # Returns an instance of the job class queued with arguments available in + # Job#arguments. + def perform_later(*args) + job_or_instantiate(*args).enqueue + end + + protected + def job_or_instantiate(*args) + args.first.is_a?(self) ? args.first : new(*args) + end + end + + # Reschedules the job to be re-executed. This is useful in combination + # with the +rescue_from+ option. When you rescue an exception from your job + # you can ask Active Job to retry performing your job. + # + # ==== Options + # * <tt>:wait</tt> - Enqueues the job with the specified delay + # * <tt>:wait_until</tt> - Enqueues the job at the time specified + # * <tt>:queue</tt> - Enqueues the job on the specified queue + # * <tt>:priority</tt> - Enqueues the job with the specified priority + # + # ==== Examples + # + # class SiteScrapperJob < ActiveJob::Base + # rescue_from(ErrorLoadingSite) do + # retry_job queue: :low_priority + # end + # + # def perform(*args) + # # raise ErrorLoadingSite if cannot scrape + # end + # end + def retry_job(options={}) + enqueue options + end + + # Enqueues the job to be performed by the queue adapter. + # + # ==== Options + # * <tt>:wait</tt> - Enqueues the job with the specified delay + # * <tt>:wait_until</tt> - Enqueues the job at the time specified + # * <tt>:queue</tt> - Enqueues the job on the specified queue + # * <tt>:priority</tt> - Enqueues the job with the specified priority + # + # ==== Examples + # + # my_job_instance.enqueue + # my_job_instance.enqueue wait: 5.minutes + # my_job_instance.enqueue queue: :important + # my_job_instance.enqueue wait_until: Date.tomorrow.midnight + # my_job_instance.enqueue priority: 10 + def enqueue(options={}) + self.scheduled_at = options[:wait].seconds.from_now.to_f if options[:wait] + self.scheduled_at = options[:wait_until].to_f if options[:wait_until] + self.queue_name = self.class.queue_name_from_part(options[:queue]) if options[:queue] + self.priority = options[:priority].to_i if options[:priority] + run_callbacks :enqueue do + if self.scheduled_at + self.class.queue_adapter.enqueue_at self, self.scheduled_at + else + self.class.queue_adapter.enqueue self + end + end + self + end + end +end diff --git a/activejob/lib/active_job/execution.rb b/activejob/lib/active_job/execution.rb new file mode 100644 index 0000000000..79d232da4a --- /dev/null +++ b/activejob/lib/active_job/execution.rb @@ -0,0 +1,42 @@ +require 'active_support/rescuable' +require 'active_job/arguments' + +module ActiveJob + module Execution + extend ActiveSupport::Concern + include ActiveSupport::Rescuable + + # Includes methods for executing and performing jobs instantly. + module ClassMethods + # Performs the job immediately. + # + # MyJob.perform_now("mike") + # + def perform_now(*args) + job_or_instantiate(*args).perform_now + end + + def execute(job_data) #:nodoc: + job = deserialize(job_data) + job.perform_now + end + end + + # Performs the job immediately. The job is not sent to the queueing adapter + # but directly executed by blocking the execution of others until it's finished. + # + # MyJob.new(*args).perform_now + def perform_now + deserialize_arguments_if_needed + run_callbacks :perform do + perform(*arguments) + end + rescue => exception + rescue_with_handler(exception) || raise(exception) + end + + def perform(*) + fail NotImplementedError + end + end +end diff --git a/activejob/lib/active_job/gem_version.rb b/activejob/lib/active_job/gem_version.rb new file mode 100644 index 0000000000..27a5de93f4 --- /dev/null +++ b/activejob/lib/active_job/gem_version.rb @@ -0,0 +1,15 @@ +module ActiveJob + # Returns the version of the currently loaded Active Job as a <tt>Gem::Version</tt> + def self.gem_version + Gem::Version.new VERSION::STRING + end + + module VERSION + MAJOR = 5 + MINOR = 0 + TINY = 0 + PRE = "alpha" + + STRING = [MAJOR, MINOR, TINY, PRE].compact.join(".") + end +end diff --git a/activejob/lib/active_job/logging.rb b/activejob/lib/active_job/logging.rb new file mode 100644 index 0000000000..605057d1e8 --- /dev/null +++ b/activejob/lib/active_job/logging.rb @@ -0,0 +1,121 @@ +require 'active_support/core_ext/hash/transform_values' +require 'active_support/core_ext/string/filters' +require 'active_support/tagged_logging' +require 'active_support/logger' + +module ActiveJob + module Logging #:nodoc: + extend ActiveSupport::Concern + + included do + cattr_accessor(:logger) { ActiveSupport::TaggedLogging.new(ActiveSupport::Logger.new(STDOUT)) } + + around_enqueue do |_, block, _| + tag_logger do + block.call + end + end + + around_perform do |job, block, _| + tag_logger(job.class.name, job.job_id) do + payload = {adapter: job.class.queue_adapter, job: job} + ActiveSupport::Notifications.instrument("perform_start.active_job", payload.dup) + ActiveSupport::Notifications.instrument("perform.active_job", payload) do + block.call + end + end + end + + after_enqueue do |job| + if job.scheduled_at + ActiveSupport::Notifications.instrument "enqueue_at.active_job", + adapter: job.class.queue_adapter, job: job + else + ActiveSupport::Notifications.instrument "enqueue.active_job", + adapter: job.class.queue_adapter, job: job + end + end + end + + private + def tag_logger(*tags) + if logger.respond_to?(:tagged) + tags.unshift "ActiveJob" unless logger_tagged_by_active_job? + ActiveJob::Base.logger.tagged(*tags){ yield } + else + yield + end + end + + def logger_tagged_by_active_job? + logger.formatter.current_tags.include?("ActiveJob") + end + + class LogSubscriber < ActiveSupport::LogSubscriber #:nodoc: + def enqueue(event) + info do + job = event.payload[:job] + "Enqueued #{job.class.name} (Job ID: #{job.job_id}) to #{queue_name(event)}" + args_info(job) + end + end + + def enqueue_at(event) + info do + job = event.payload[:job] + "Enqueued #{job.class.name} (Job ID: #{job.job_id}) to #{queue_name(event)} at #{scheduled_at(event)}" + args_info(job) + end + end + + def perform_start(event) + info do + job = event.payload[:job] + "Performing #{job.class.name} from #{queue_name(event)}" + args_info(job) + end + end + + def perform(event) + info do + job = event.payload[:job] + "Performed #{job.class.name} from #{queue_name(event)} in #{event.duration.round(2)}ms" + end + end + + private + def queue_name(event) + event.payload[:adapter].class.name.demodulize.remove('Adapter') + "(#{event.payload[:job].queue_name})" + end + + def args_info(job) + if job.arguments.any? + ' with arguments: ' + + job.arguments.map { |arg| format(arg).inspect }.join(', ') + else + '' + end + end + + def format(arg) + case arg + when Hash + arg.transform_values { |value| format(value) } + when Array + arg.map { |value| format(value) } + when GlobalID::Identification + arg.to_global_id rescue arg + else + arg + end + end + + def scheduled_at(event) + Time.at(event.payload[:job].scheduled_at).utc + end + + def logger + ActiveJob::Base.logger + end + end + end +end + +ActiveJob::Logging::LogSubscriber.attach_to :active_job diff --git a/activejob/lib/active_job/queue_adapter.rb b/activejob/lib/active_job/queue_adapter.rb new file mode 100644 index 0000000000..457015b741 --- /dev/null +++ b/activejob/lib/active_job/queue_adapter.rb @@ -0,0 +1,63 @@ +require 'active_job/queue_adapters/inline_adapter' +require 'active_support/core_ext/class/attribute' +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. + 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 + # Returns the backend queue provider. The default queue adapter + # is the +:inline+ 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 + # information. + def queue_adapter=(name_or_adapter_or_class) + self._queue_adapter = interpret_adapter(name_or_adapter_or_class) + end + + private + + 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 new file mode 100644 index 0000000000..aeb1fe1e73 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters.rb @@ -0,0 +1,136 @@ +module ActiveJob + # == Active Job adapters + # + # Active Job has adapters for the following queueing backends: + # + # * {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 1.x}[https://github.com/resque/resque/tree/1-x-stable] + # * {Sidekiq}[http://sidekiq.org] + # * {Sneakers}[https://github.com/jondot/sneakers] + # * {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] + # + # === Backends Features + # + # | | Async | Queues | Delayed | Priorities | Timeout | Retries | + # |-------------------|-------|--------|------------|------------|---------|---------| + # | 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 | + # | Sidekiq | Yes | Yes | Yes | Queue | No | Job | + # | Sneakers | Yes | Yes | No | Queue | Queue | No | + # | Sucker Punch | Yes | Yes | No | 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 | + # + # ==== Async + # + # Yes: The Queue Adapter runs the jobs in a separate or forked process. + # + # No: The job is run in the same process. + # + # ==== Queues + # + # Yes: Jobs may set which queue they are run in with queue_as or by using the set + # method. + # + # ==== Delayed + # + # Yes: The adapter will run the job in the future through perform_later. + # + # (Gem): An additional gem is required to use perform_later with this adapter. + # + # No: The adapter will run jobs at the next opportunity and cannot use perform_later. + # + # N/A: The adapter does not support queueing. + # + # NOTE: + # queue_classic supports job scheduling since version 3.1. + # For older versions you can use the queue_classic-later gem. + # + # ==== Priorities + # + # The order in which jobs are processed can be configured differently depending + # on the adapter. + # + # Job: Any class inheriting from the adapter may set the priority on the job + # object relative to other jobs. + # + # Queue: The adapter can set the priority for job queues, when setting a queue + # with Active Job this will be respected. + # + # Yes: Allows the priority to be set on the job object, at the queue level or + # as default configuration option. + # + # No: Does not allow the priority of jobs to be configured. + # + # N/A: The adapter does not support queueing, and therefore sorting them. + # + # ==== Timeout + # + # When a job will stop after the allotted time. + # + # Job: The timeout can be set for each instance of the job class. + # + # Queue: The timeout is set for all jobs on the queue. + # + # Global: The adapter is configured that all jobs have a maximum run time. + # + # N/A: This adapter does not run in a separate process, and therefore timeout + # is unsupported. + # + # ==== Retries + # + # Job: The number of retries can be set per instance of the job class. + # + # Yes: The Number of retries can be configured globally, for each instance or + # on the queue. This adapter may also present failed instances of the job class + # that can be restarted. + # + # Global: The adapter has a global number of retries. + # + # N/A: The adapter does not run in a separate process, and therefore doesn't + # support retries. + # + # === Async and Inline Queue Adapters + # + # Active Job has two built-in queue adapters intended for development and + # testing: +:async+ and +:inline+. + module QueueAdapters + extend ActiveSupport::Autoload + + autoload :AsyncAdapter + autoload :InlineAdapter + autoload :BackburnerAdapter + autoload :DelayedJobAdapter + autoload :QuAdapter + autoload :QueAdapter + autoload :QueueClassicAdapter + autoload :ResqueAdapter + autoload :SidekiqAdapter + autoload :SneakersAdapter + autoload :SuckerPunchAdapter + autoload :TestAdapter + + ADAPTER = 'Adapter'.freeze + private_constant :ADAPTER + + class << self + # Returns adapter for specified name. + # + # ActiveJob::QueueAdapters.lookup(:sidekiq) + # # => ActiveJob::QueueAdapters::SidekiqAdapter + def lookup(name) + const_get(name.to_s.camelize << ADAPTER) + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/async_adapter.rb b/activejob/lib/active_job/queue_adapters/async_adapter.rb new file mode 100644 index 0000000000..3fc27f56e7 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/async_adapter.rb @@ -0,0 +1,23 @@ +require 'active_job/async_job' + +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]. + # + # To use +AsyncJob+ set the queue_adapter config to +:async+. + # + # Rails.application.config.active_job.queue_adapter = :async + class AsyncAdapter + def enqueue(job) #:nodoc: + ActiveJob::AsyncJob.enqueue(job.serialize, queue: job.queue_name) + end + + def enqueue_at(job, timestamp) #:nodoc: + ActiveJob::AsyncJob.enqueue_at(job.serialize, timestamp, queue: job.queue_name) + 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 new file mode 100644 index 0000000000..17703e3e41 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/backburner_adapter.rb @@ -0,0 +1,34 @@ +require 'backburner' + +module ActiveJob + module QueueAdapters + # == Backburner adapter for Active Job + # + # Backburner is a beanstalkd-powered job queue that can handle a very + # high volume of jobs. You create background jobs and place them on + # multiple work queues to be processed later. Read more about + # Backburner {here}[https://github.com/nesquena/backburner]. + # + # To use Backburner set the queue_adapter config to +:backburner+. + # + # Rails.application.config.active_job.queue_adapter = :backburner + class BackburnerAdapter + 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 + + class JobWrapper #:nodoc: + class << self + def perform(job_data) + Base.execute job_data + end + end + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb b/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb new file mode 100644 index 0000000000..0a785fad3b --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb @@ -0,0 +1,41 @@ +require 'delayed_job' + +module ActiveJob + module QueueAdapters + # == Delayed Job adapter for Active Job + # + # Delayed::Job (or DJ) encapsulates the common pattern of asynchronously + # executing longer tasks in the background. Although DJ can have many + # storage backends, one of the most used is based on Active Record. + # Read more about Delayed Job {here}[https://github.com/collectiveidea/delayed_job]. + # + # To use Delayed Job, set the queue_adapter config to +:delayed_job+. + # + # Rails.application.config.active_job.queue_adapter = :delayed_job + class DelayedJobAdapter + def enqueue(job) #:nodoc: + delayed_job = Delayed::Job.enqueue(JobWrapper.new(job.serialize), queue: job.queue_name, priority: job.priority) + job.provider_job_id = delayed_job.id + delayed_job + end + + def enqueue_at(job, timestamp) #:nodoc: + delayed_job = Delayed::Job.enqueue(JobWrapper.new(job.serialize), queue: job.queue_name, priority: job.priority, run_at: Time.at(timestamp)) + job.provider_job_id = delayed_job.id + delayed_job + end + + class JobWrapper #:nodoc: + attr_accessor :job_data + + 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/queue_adapters/inline_adapter.rb b/activejob/lib/active_job/queue_adapters/inline_adapter.rb new file mode 100644 index 0000000000..8ad5f4de07 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/inline_adapter.rb @@ -0,0 +1,21 @@ +module ActiveJob + module QueueAdapters + # == Active Job Inline adapter + # + # When enqueueing jobs with the Inline adapter the job will be executed + # immediately. + # + # To use the Inline set the queue_adapter config to +:inline+. + # + # Rails.application.config.active_job.queue_adapter = :inline + class InlineAdapter + def enqueue(job) #:nodoc: + Base.execute(job.serialize) + 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" + end + 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 new file mode 100644 index 0000000000..0e198922fc --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/qu_adapter.rb @@ -0,0 +1,44 @@ +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/queue_adapters/que_adapter.rb b/activejob/lib/active_job/queue_adapters/que_adapter.rb new file mode 100644 index 0000000000..ab13689747 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/que_adapter.rb @@ -0,0 +1,37 @@ +require 'que' + +module ActiveJob + module QueueAdapters + # == Que adapter for Active Job + # + # Que is a high-performance alternative to DelayedJob or QueueClassic that + # improves the reliability of your application by protecting your jobs with + # the same ACID guarantees as the rest of your data. Que is a queue for + # Ruby and PostgreSQL that manages jobs using advisory locks. + # + # Read more about Que {here}[https://github.com/chanks/que]. + # + # To use Que set the queue_adapter config to +:que+. + # + # Rails.application.config.active_job.queue_adapter = :que + class QueAdapter + def enqueue(job) #:nodoc: + que_job = JobWrapper.enqueue job.serialize, priority: job.priority + job.provider_job_id = que_job.attrs["job_id"] + que_job + end + + def enqueue_at(job, timestamp) #:nodoc: + que_job = JobWrapper.enqueue job.serialize, priority: job.priority, run_at: Time.at(timestamp) + job.provider_job_id = que_job.attrs["job_id"] + que_job + end + + class JobWrapper < Que::Job #:nodoc: + def run(job_data) + Base.execute job_data + end + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb b/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb new file mode 100644 index 0000000000..0ee41407d8 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb @@ -0,0 +1,56 @@ +require 'queue_classic' + +module ActiveJob + module QueueAdapters + # == queue_classic adapter for Active Job + # + # queue_classic provides a simple interface to a PostgreSQL-backed message + # queue. queue_classic specializes in concurrent locking and minimizing + # database load while providing a simple, intuitive developer experience. + # queue_classic assumes that you are already using PostgreSQL in your + # production environment and that adding another dependency (e.g. redis, + # beanstalkd, 0mq) is undesirable. + # + # Read more about queue_classic {here}[https://github.com/QueueClassic/queue_classic]. + # + # To use queue_classic set the queue_adapter config to +:queue_classic+. + # + # Rails.application.config.active_job.queue_adapter = :queue_classic + class QueueClassicAdapter + def enqueue(job) #:nodoc: + qc_job = build_queue(job.queue_name).enqueue("#{JobWrapper.name}.perform", job.serialize) + job.provider_job_id = qc_job["id"] if qc_job.is_a?(Hash) + qc_job + 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 + qc_job = queue.enqueue_at(timestamp, "#{JobWrapper.name}.perform", job.serialize) + job.provider_job_id = qc_job["id"] if qc_job.is_a?(Hash) + qc_job + 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: + class << self + def perform(job_data) + Base.execute job_data + end + end + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/resque_adapter.rb b/activejob/lib/active_job/queue_adapters/resque_adapter.rb new file mode 100644 index 0000000000..417854afd8 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/resque_adapter.rb @@ -0,0 +1,50 @@ +require 'resque' +require 'active_support/core_ext/enumerable' +require 'active_support/core_ext/array/access' + +begin + require 'resque-scheduler' +rescue LoadError + begin + require 'resque_scheduler' + rescue LoadError + false + end +end + +module ActiveJob + module QueueAdapters + # == Resque adapter for Active Job + # + # Resque (pronounced like "rescue") is a Redis-backed library for creating + # background jobs, placing those jobs on multiple queues, and processing + # them later. + # + # Read more about Resque {here}[https://github.com/resque/resque]. + # + # To use Resque set the queue_adapter config to +:resque+. + # + # Rails.application.config.active_job.queue_adapter = :resque + class ResqueAdapter + 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 + end + + class JobWrapper #:nodoc: + class << self + def perform(job_data) + Base.execute job_data + end + end + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb b/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb new file mode 100644 index 0000000000..c321776bf5 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb @@ -0,0 +1,45 @@ +require 'sidekiq' + +module ActiveJob + module QueueAdapters + # == Sidekiq adapter for Active Job + # + # Simple, efficient background processing for Ruby. Sidekiq uses threads to + # handle many jobs at the same time in the same process. It does not + # require Rails but will integrate tightly with it to make background + # processing dead simple. + # + # Read more about Sidekiq {here}[http://sidekiq.org]. + # + # To use Sidekiq set the queue_adapter config to +:sidekiq+. + # + # Rails.application.config.active_job.queue_adapter = :sidekiq + class SidekiqAdapter + def enqueue(job) #:nodoc: + #Sidekiq::Client does not support symbols as keys + job.provider_job_id = Sidekiq::Client.push \ + 'class' => JobWrapper, + 'wrapped' => job.class.to_s, + 'queue' => job.queue_name, + 'args' => [ job.serialize ] + end + + def enqueue_at(job, timestamp) #:nodoc: + job.provider_job_id = Sidekiq::Client.push \ + 'class' => JobWrapper, + 'wrapped' => job.class.to_s, + 'queue' => job.queue_name, + 'args' => [ job.serialize ], + 'at' => timestamp + end + + class JobWrapper #:nodoc: + include Sidekiq::Worker + + def perform(job_data) + Base.execute job_data + end + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb b/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb new file mode 100644 index 0000000000..d78bdecdcb --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb @@ -0,0 +1,46 @@ +require 'sneakers' +require 'monitor' + +module ActiveJob + module QueueAdapters + # == Sneakers adapter for Active Job + # + # A high-performance RabbitMQ background processing framework for Ruby. + # Sneakers is being used in production for both I/O and CPU intensive + # workloads, and have achieved the goals of high-performance and + # 0-maintenance, as designed. + # + # Read more about Sneakers {here}[https://github.com/jondot/sneakers]. + # + # To use Sneakers set the queue_adapter config to +:sneakers+. + # + # Rails.application.config.active_job.queue_adapter = :sneakers + class SneakersAdapter + def initialize + @monitor = Monitor.new + 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, "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 #:nodoc: + include Sneakers::Worker + from_queue 'default' + + def work(msg) + job_data = ActiveSupport::JSON.decode(msg) + Base.execute job_data + ack! + end + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb b/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb new file mode 100644 index 0000000000..c6c35f8ab4 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb @@ -0,0 +1,38 @@ +require 'sucker_punch' + +module ActiveJob + module QueueAdapters + # == 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. + # + # Read more about Sucker Punch {here}[https://github.com/brandonhilkert/sucker_punch]. + # + # To use Sucker Punch set the queue_adapter config to +:sucker_punch+. + # + # Rails.application.config.active_job.queue_adapter = :sucker_punch + class SuckerPunchAdapter + def enqueue(job) #:nodoc: + JobWrapper.new.async.perform job.serialize + 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" + end + + class JobWrapper #:nodoc: + include SuckerPunch::Job + + def perform(job_data) + Base.execute job_data + end + end + end + end +end diff --git a/activejob/lib/active_job/queue_adapters/test_adapter.rb b/activejob/lib/active_job/queue_adapters/test_adapter.rb new file mode 100644 index 0000000000..9b7b7139f4 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/test_adapter.rb @@ -0,0 +1,60 @@ +module ActiveJob + module QueueAdapters + # == Test adapter for Active Job + # + # The test adapter should be used only in testing. Along with + # <tt>ActiveJob::TestCase</tt> and <tt>ActiveJob::TestHelper</tt> + # it makes a great tool to test your Rails application. + # + # To use the test adapter set queue_adapter config to +:test+. + # + # Rails.application.config.active_job.queue_adapter = :test + class TestAdapter + 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 performed jobs with the TestAdapter so you can check them. + def performed_jobs + @performed_jobs ||= [] + end + + def enqueue(job) #:nodoc: + return if filtered?(job) + + 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) + + job_data = job_to_hash(job, at: timestamp) + enqueue_or_perform(perform_enqueued_at_jobs, job, job_data) + end + + private + + 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 + end + + def filtered?(job) + filter && !Array(filter).include?(job.class) + end + end + end +end diff --git a/activejob/lib/active_job/queue_name.rb b/activejob/lib/active_job/queue_name.rb new file mode 100644 index 0000000000..65786a49ff --- /dev/null +++ b/activejob/lib/active_job/queue_name.rb @@ -0,0 +1,51 @@ +module ActiveJob + module QueueName + extend ActiveSupport::Concern + + # Includes the ability to override the default queue name and prefix. + module ClassMethods + mattr_accessor(:queue_name_prefix) + mattr_accessor(:default_queue_name) { "default" } + + # Specifies the name of the queue to process the job on. + # + # class PublishToFeedJob < ActiveJob::Base + # queue_as :feeds + # + # def perform(post) + # post.to_feed! + # end + # end + def queue_as(part_name=nil, &block) + if block_given? + self.queue_name = block + else + self.queue_name = queue_name_from_part(part_name) + end + end + + def queue_name_from_part(part_name) #:nodoc: + queue_name = part_name || default_queue_name + name_parts = [queue_name_prefix.presence, queue_name] + name_parts.compact.join(queue_name_delimiter) + end + end + + included do + class_attribute :queue_name, instance_accessor: false + class_attribute :queue_name_delimiter, instance_accessor: false + + self.queue_name = default_queue_name + self.queue_name_delimiter = '_' # set default delimiter to '_' + end + + # Returns the name of the queue the job will be run on. + def queue_name + if @queue_name.is_a?(Proc) + @queue_name = self.class.queue_name_from_part(instance_exec(&@queue_name)) + end + @queue_name + end + + end +end diff --git a/activejob/lib/active_job/queue_priority.rb b/activejob/lib/active_job/queue_priority.rb new file mode 100644 index 0000000000..01d84910ff --- /dev/null +++ b/activejob/lib/active_job/queue_priority.rb @@ -0,0 +1,44 @@ +module ActiveJob + module QueuePriority + extend ActiveSupport::Concern + + # Includes the ability to override the default queue priority. + module ClassMethods + mattr_accessor(:default_priority) + + # Specifies the priority of the queue to create the job with. + # + # class PublishToFeedJob < ActiveJob::Base + # queue_with_priority 50 + # + # def perform(post) + # post.to_feed! + # end + # end + # + # Specify either an argument or a block. + def queue_with_priority(priority=nil, &block) + if block_given? + self.priority = block + else + self.priority = priority + end + end + end + + included do + class_attribute :priority, instance_accessor: false + + self.priority = default_priority + end + + # Returns the priority that the job will be created with + def priority + if @priority.is_a?(Proc) + @priority = instance_exec(&@priority) + end + @priority + end + + end +end diff --git a/activejob/lib/active_job/railtie.rb b/activejob/lib/active_job/railtie.rb new file mode 100644 index 0000000000..6538ac1b30 --- /dev/null +++ b/activejob/lib/active_job/railtie.rb @@ -0,0 +1,23 @@ +require 'global_id/railtie' +require 'active_job' + +module ActiveJob + # = Active Job Railtie + class Railtie < Rails::Railtie # :nodoc: + config.active_job = ActiveSupport::OrderedOptions.new + + initializer 'active_job.logger' do + ActiveSupport.on_load(:active_job) { self.logger = ::Rails.logger } + end + + initializer "active_job.set_configs" do |app| + options = app.config.active_job + options.queue_adapter ||= :inline + + ActiveSupport.on_load(:active_job) do + options.each { |k,v| send("#{k}=", v) } + end + end + + end +end diff --git a/activejob/lib/active_job/test_case.rb b/activejob/lib/active_job/test_case.rb new file mode 100644 index 0000000000..d894a7b5cd --- /dev/null +++ b/activejob/lib/active_job/test_case.rb @@ -0,0 +1,7 @@ +require 'active_support/test_case' + +module ActiveJob + class TestCase < ActiveSupport::TestCase + include ActiveJob::TestHelper + end +end diff --git a/activejob/lib/active_job/test_helper.rb b/activejob/lib/active_job/test_helper.rb new file mode 100644 index 0000000000..44ddfa5f69 --- /dev/null +++ b/activejob/lib/active_job/test_helper.rb @@ -0,0 +1,328 @@ +require 'active_support/core_ext/class/subclasses' +require 'active_support/core_ext/hash/keys' + +module ActiveJob + # Provides helper methods for testing Active Job + module TestHelper + extend ActiveSupport::Concern + + included do + 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 + 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 + + # 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 + 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 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 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 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 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 + + 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 + ActiveJob::Base.queue_adapter + end + + delegate :enqueued_jobs, :enqueued_jobs=, + :performed_jobs, :performed_jobs=, + to: :queue_adapter + + private + def clear_enqueued_jobs # :nodoc: + enqueued_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 + 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 + end +end diff --git a/activejob/lib/active_job/translation.rb b/activejob/lib/active_job/translation.rb new file mode 100644 index 0000000000..67e4cf4ab9 --- /dev/null +++ b/activejob/lib/active_job/translation.rb @@ -0,0 +1,11 @@ +module ActiveJob + module Translation #:nodoc: + extend ActiveSupport::Concern + + included do + around_perform do |job, block, _| + I18n.with_locale(job.locale, &block) + end + end + end +end diff --git a/activejob/lib/active_job/version.rb b/activejob/lib/active_job/version.rb new file mode 100644 index 0000000000..971ba9fe0c --- /dev/null +++ b/activejob/lib/active_job/version.rb @@ -0,0 +1,8 @@ +require_relative 'gem_version' + +module ActiveJob + # Returns the version of the currently loaded Active Job as a <tt>Gem::Version</tt> + def self.version + gem_version + end +end |