diff options
Diffstat (limited to 'activejob')
100 files changed, 4702 insertions, 0 deletions
diff --git a/activejob/.gitignore b/activejob/.gitignore new file mode 100644 index 0000000000..b3aaf55871 --- /dev/null +++ b/activejob/.gitignore @@ -0,0 +1 @@ +test/dummy diff --git a/activejob/CHANGELOG.md b/activejob/CHANGELOG.md new file mode 100644 index 0000000000..79235019fe --- /dev/null +++ b/activejob/CHANGELOG.md @@ -0,0 +1,132 @@ +* Fixed serializing `:at` option for `assert_enqueued_with` + and `assert_performed_with`. + + *Wojciech Wnętrzak* + +* Support passing array to `assert_enqueued_jobs` in `:only` option. + + *Wojciech Wnętrzak* + +* Add job priorities to Active Job. + + *wvengen* + +* Implement a simple `AsyncJob` processor and associated `AsyncAdapter` that + queue jobs to a `concurrent-ruby` thread pool. + + *Jerry D'Antonio* + +* Implement `provider_job_id` for `queue_classic` adapter. This requires the + latest, currently unreleased, version of queue_classic. + + *Yves Senn* + +* `assert_enqueued_with` and `assert_performed_with` now returns the matched + job instance for further assertions. + + *Jean Boussier* + +* Include I18n.locale into job serialization/deserialization and use it around + `perform`. + + Fixes #20799. + + *Johannes Opper* + +* Allow `DelayedJob`, `Sidekiq`, `qu`, and `que` to report the job id back to + `ActiveJob::Base` as `provider_job_id`. + + Fixes #18821. + + *Kevin Deisz*, *Jeroen van Baarsen* + +* `assert_enqueued_jobs` and `assert_performed_jobs` in block form use the + given number as expected value. This makes the error message much easier to + understand. + + *y-yagi* + +* A generated job now inherits from `app/jobs/application_job.rb` by default. + + *Jeroen van Baarsen* + +* Add an `:only` option to `perform_enqueued_jobs` to filter jobs based on + type. + + This allows specific jobs to be tested, while preventing others from + being performed unnecessarily. + + Example: + + 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. + + Example: + + 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 + + Fixes #18802. + + *Michael Ryan* + +* Allow keyword arguments to be used with Active Job. + + Fixes #18741. + + *Sean Griffin* + +* Add `:only` option to `assert_enqueued_jobs`, to check the number of times + a specific kind of job is enqueued. + + Example: + + def test_logging_job + assert_enqueued_jobs 1, only: LoggingJob do + LoggingJob.perform_later + HelloJob.perform_later('jeremy') + end + end + + *George Claghorn* + +* `ActiveJob::Base.deserialize` delegates to the job class. + + Since `ActiveJob::Base#deserialize` can be overridden by subclasses (like + `ActiveJob::Base#serialize`) this allows jobs to attach arbitrary metadata + when they get serialized and read it back when they get performed. + + Example: + + 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 + + *Isaac Seymour* + +Please check [4-2-stable](https://github.com/rails/rails/blob/4-2-stable/activejob/CHANGELOG.md) for previous changes. diff --git a/activejob/MIT-LICENSE b/activejob/MIT-LICENSE new file mode 100644 index 0000000000..0cef8cdda0 --- /dev/null +++ b/activejob/MIT-LICENSE @@ -0,0 +1,21 @@ +Copyright (c) 2014-2015 David Heinemeier Hansson + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be +included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + diff --git a/activejob/README.md b/activejob/README.md new file mode 100644 index 0000000000..f9a3183b1a --- /dev/null +++ b/activejob/README.md @@ -0,0 +1,131 @@ +# Active Job -- Make work happen later + +Active Job is a framework for declaring jobs and making them run on a variety +of queueing backends. These jobs can be everything from regularly scheduled +clean-ups, to billing charges, to mailings. Anything that can be chopped up into +small units of work and run in parallel, really. + +It also serves as the backend for Action Mailer's #deliver_later functionality +that makes it easy to turn any mailing into a job for running later. That's +one of the most common jobs in a modern web application: sending emails outside +of the request-response cycle, so the user doesn't have to wait on it. + +The main point is to ensure that all Rails apps will have a job infrastructure +in place, even if it's in the form of an "immediate runner". We can then have +framework features and other gems build on top of that, without having to worry +about API differences between Delayed Job and Resque. Picking your queuing +backend becomes more of an operational concern, then. And you'll be able to +switch between them without having to rewrite your jobs. + + +## Usage + +Set the queue adapter for Active Job: + +``` ruby +ActiveJob::Base.queue_adapter = :inline # default queue adapter +``` +Note: To learn how to use your preferred queueing backend see its adapter +documentation at +[ActiveJob::QueueAdapters](http://api.rubyonrails.org/classes/ActiveJob/QueueAdapters.html). + +Declare a job like so: + +```ruby +class MyJob < ActiveJob::Base + queue_as :my_jobs + + def perform(record) + record.do_work + end +end +``` + +Enqueue a job like so: + +```ruby +MyJob.perform_later record # Enqueue a job to be performed as soon the queueing system is free. +``` + +```ruby +MyJob.set(wait_until: Date.tomorrow.noon).perform_later(record) # Enqueue a job to be performed tomorrow at noon. +``` + +```ruby +MyJob.set(wait: 1.week).perform_later(record) # Enqueue a job to be performed 1 week from now. +``` + +That's it! + + +## GlobalID support + +Active Job supports [GlobalID serialization](https://github.com/rails/globalid/) for parameters. This makes it possible +to pass live Active Record objects to your job instead of class/id pairs, which +you then have to manually deserialize. Before, jobs would look like this: + +```ruby +class TrashableCleanupJob + def perform(trashable_class, trashable_id, depth) + trashable = trashable_class.constantize.find(trashable_id) + trashable.cleanup(depth) + end +end +``` + +Now you can simply do: + +```ruby +class TrashableCleanupJob + def perform(trashable, depth) + trashable.cleanup(depth) + end +end +``` + +This works with any class that mixes in GlobalID::Identification, which +by default has been mixed into Active Record classes. + + +## Supported queueing systems + +Active Job has built-in adapters for multiple queueing backends (Sidekiq, +Resque, Delayed Job and others). To get an up-to-date list of the adapters +see the API Documentation for [ActiveJob::QueueAdapters](http://api.rubyonrails.org/classes/ActiveJob/QueueAdapters.html). + +## Auxiliary gems + +* [activejob-stats](https://github.com/seuros/activejob-stats) + +## Download and installation + +The latest version of Active Job can be installed with RubyGems: + +``` + % gem install activejob +``` + +Source code can be downloaded as part of the Rails project on GitHub + +* https://github.com/rails/rails/tree/master/activejob + +## License + +Active Job is released under the MIT license: + +* http://www.opensource.org/licenses/MIT + + +## Support + +API documentation is at: + +* http://api.rubyonrails.org + +Bug reports can be filed for the Ruby on Rails project here: + +* https://github.com/rails/rails/issues + +Feature requests should be discussed on the rails-core mailing list here: + +* https://groups.google.com/forum/?fromgroups#!forum/rubyonrails-core diff --git a/activejob/Rakefile b/activejob/Rakefile new file mode 100644 index 0000000000..d9648a7f16 --- /dev/null +++ b/activejob/Rakefile @@ -0,0 +1,75 @@ +require 'rake/testtask' + +ACTIVEJOB_ADAPTERS = %w(async inline delayed_job qu que queue_classic resque sidekiq sneakers sucker_punch backburner test) +ACTIVEJOB_ADAPTERS -= %w(queue_classic) if defined?(JRUBY_VERSION) + +task default: :test +task test: 'test:default' + +namespace :test do + desc 'Run all adapter tests' + task :default do + run_without_aborting ACTIVEJOB_ADAPTERS.map { |a| "test:#{a}" } + end + + desc 'Run all adapter tests in isolation' + task :isolated do + run_without_aborting ACTIVEJOB_ADAPTERS.map { |a| "test:isolated:#{a}" } + end + + desc 'Run integration tests for all adapters' + task :integration do + run_without_aborting (ACTIVEJOB_ADAPTERS - ['test']).map { |a| "test:integration:#{a}" } + end + + task 'env:integration' do + ENV['AJ_INTEGRATION_TESTS'] = "1" + end + + ACTIVEJOB_ADAPTERS.each do |adapter| + task("env:#{adapter}") { ENV['AJ_ADAPTER'] = adapter } + + Rake::TestTask.new(adapter => "test:env:#{adapter}") do |t| + t.description = "Run adapter tests for #{adapter}" + t.libs << 'test' + t.test_files = FileList['test/cases/**/*_test.rb'] + t.verbose = true + t.warning = false + t.ruby_opts = ["--dev"] if defined?(JRUBY_VERSION) + end + + namespace :isolated do + task adapter => "test:env:#{adapter}" do + dir = File.dirname(__FILE__) + Dir.glob("#{dir}/test/cases/**/*_test.rb").all? do |file| + sh(Gem.ruby, '-w', "-I#{dir}/lib", "-I#{dir}/test", file) + end or raise 'Failures' + end + end + + namespace :integration do + Rake::TestTask.new(adapter => ["test:env:#{adapter}", 'test:env:integration']) do |t| + t.description = "Run integration tests for #{adapter}" + t.libs << 'test' + t.test_files = FileList['test/integration/**/*_test.rb'] + t.verbose = true + t.warning = false + t.ruby_opts = ["--dev"] if defined?(JRUBY_VERSION) + end + end + end +end + +def run_without_aborting(tasks) + errors = [] + + tasks.each do |task| + begin + Rake::Task[task].invoke + rescue Exception + errors << task + end + end + + abort "Errors running #{errors.join(', ')}" if errors.any? +end diff --git a/activejob/activejob.gemspec b/activejob/activejob.gemspec new file mode 100644 index 0000000000..24e38e495f --- /dev/null +++ b/activejob/activejob.gemspec @@ -0,0 +1,23 @@ +version = File.read(File.expand_path('../../RAILS_VERSION', __FILE__)).strip + +Gem::Specification.new do |s| + s.platform = Gem::Platform::RUBY + s.name = 'activejob' + s.version = version + s.summary = 'Job framework with pluggable queues.' + s.description = 'Declare job classes that can be run by a variety of queueing backends.' + + s.required_ruby_version = '>= 2.2.2' + + s.license = 'MIT' + + s.author = 'David Heinemeier Hansson' + s.email = 'david@loudthinking.com' + s.homepage = 'http://www.rubyonrails.org' + + s.files = Dir['CHANGELOG.md', 'MIT-LICENSE', 'README.md', 'lib/**/*'] + s.require_path = 'lib' + + s.add_dependency 'activesupport', version + s.add_dependency 'globalid', '>= 0.3.0' +end diff --git a/activejob/lib/active_job.rb b/activejob/lib/active_job.rb new file mode 100644 index 0000000000..eb8091a805 --- /dev/null +++ b/activejob/lib/active_job.rb @@ -0,0 +1,38 @@ +#-- +# Copyright (c) 2014-2015 David Heinemeier Hansson +# +# Permission is hereby granted, free of charge, to any person obtaining +# a copy of this software and associated documentation files (the +# "Software"), to deal in the Software without restriction, including +# without limitation the rights to use, copy, modify, merge, publish, +# distribute, sublicense, and/or sell copies of the Software, and to +# permit persons to whom the Software is furnished to do so, subject to +# the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +#++ + +require 'active_support' +require 'active_support/rails' +require 'active_job/version' +require 'global_id' + +module ActiveJob + extend ActiveSupport::Autoload + + autoload :Base + autoload :QueueAdapters + autoload :ConfiguredJob + autoload :AsyncJob + autoload :TestCase + autoload :TestHelper +end 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 diff --git a/activejob/lib/rails/generators/job/job_generator.rb b/activejob/lib/rails/generators/job/job_generator.rb new file mode 100644 index 0000000000..2115fb9f71 --- /dev/null +++ b/activejob/lib/rails/generators/job/job_generator.rb @@ -0,0 +1,23 @@ +require 'rails/generators/named_base' + +module Rails # :nodoc: + module Generators # :nodoc: + class JobGenerator < Rails::Generators::NamedBase # :nodoc: + desc 'This generator creates an active job file at app/jobs' + + class_option :queue, type: :string, default: 'default', desc: 'The queue name for the generated job' + + check_class_collision suffix: 'Job' + + hook_for :test_framework + + def self.default_generator_root + File.dirname(__FILE__) + end + + def create_job_file + template 'job.rb', File.join('app/jobs', class_path, "#{file_name}_job.rb") + end + end + end +end diff --git a/activejob/lib/rails/generators/job/templates/job.rb b/activejob/lib/rails/generators/job/templates/job.rb new file mode 100644 index 0000000000..4ad2914a45 --- /dev/null +++ b/activejob/lib/rails/generators/job/templates/job.rb @@ -0,0 +1,9 @@ +<% module_namespacing do -%> +class <%= class_name %>Job < ApplicationJob + queue_as :<%= options[:queue] %> + + def perform(*args) + # Do something later + end +end +<% end -%> diff --git a/activejob/test/adapters/async.rb b/activejob/test/adapters/async.rb new file mode 100644 index 0000000000..df58027599 --- /dev/null +++ b/activejob/test/adapters/async.rb @@ -0,0 +1,5 @@ +require 'concurrent' +require 'active_job/async_job' + +ActiveJob::Base.queue_adapter = :async +ActiveJob::AsyncJob.perform_immediately! diff --git a/activejob/test/adapters/backburner.rb b/activejob/test/adapters/backburner.rb new file mode 100644 index 0000000000..65d05f850b --- /dev/null +++ b/activejob/test/adapters/backburner.rb @@ -0,0 +1,3 @@ +require 'support/backburner/inline' + +ActiveJob::Base.queue_adapter = :backburner
\ No newline at end of file diff --git a/activejob/test/adapters/delayed_job.rb b/activejob/test/adapters/delayed_job.rb new file mode 100644 index 0000000000..afd9c9deb7 --- /dev/null +++ b/activejob/test/adapters/delayed_job.rb @@ -0,0 +1,7 @@ +ActiveJob::Base.queue_adapter = :delayed_job + +$LOAD_PATH << File.dirname(__FILE__) + "/../support/delayed_job" + +Delayed::Worker.delay_jobs = false +Delayed::Worker.backend = :test + diff --git a/activejob/test/adapters/inline.rb b/activejob/test/adapters/inline.rb new file mode 100644 index 0000000000..e0092552c4 --- /dev/null +++ b/activejob/test/adapters/inline.rb @@ -0,0 +1 @@ +ActiveJob::Base.queue_adapter = :inline
\ No newline at end of file diff --git a/activejob/test/adapters/qu.rb b/activejob/test/adapters/qu.rb new file mode 100644 index 0000000000..7728c843b4 --- /dev/null +++ b/activejob/test/adapters/qu.rb @@ -0,0 +1,3 @@ +require 'qu-immediate' + +ActiveJob::Base.queue_adapter = :qu diff --git a/activejob/test/adapters/que.rb b/activejob/test/adapters/que.rb new file mode 100644 index 0000000000..e6abc57457 --- /dev/null +++ b/activejob/test/adapters/que.rb @@ -0,0 +1,4 @@ +require 'support/que/inline' + +ActiveJob::Base.queue_adapter = :que +Que.mode = :sync diff --git a/activejob/test/adapters/queue_classic.rb b/activejob/test/adapters/queue_classic.rb new file mode 100644 index 0000000000..ad5ced3cc2 --- /dev/null +++ b/activejob/test/adapters/queue_classic.rb @@ -0,0 +1,2 @@ +require 'support/queue_classic/inline' +ActiveJob::Base.queue_adapter = :queue_classic diff --git a/activejob/test/adapters/resque.rb b/activejob/test/adapters/resque.rb new file mode 100644 index 0000000000..af7080358d --- /dev/null +++ b/activejob/test/adapters/resque.rb @@ -0,0 +1,2 @@ +ActiveJob::Base.queue_adapter = :resque +Resque.inline = true diff --git a/activejob/test/adapters/sidekiq.rb b/activejob/test/adapters/sidekiq.rb new file mode 100644 index 0000000000..cd9d2034de --- /dev/null +++ b/activejob/test/adapters/sidekiq.rb @@ -0,0 +1,2 @@ +require 'sidekiq/testing/inline' +ActiveJob::Base.queue_adapter = :sidekiq diff --git a/activejob/test/adapters/sneakers.rb b/activejob/test/adapters/sneakers.rb new file mode 100644 index 0000000000..204166a700 --- /dev/null +++ b/activejob/test/adapters/sneakers.rb @@ -0,0 +1,2 @@ +require 'support/sneakers/inline' +ActiveJob::Base.queue_adapter = :sneakers diff --git a/activejob/test/adapters/sucker_punch.rb b/activejob/test/adapters/sucker_punch.rb new file mode 100644 index 0000000000..d2d1712946 --- /dev/null +++ b/activejob/test/adapters/sucker_punch.rb @@ -0,0 +1,2 @@ +require 'sucker_punch/testing/inline' +ActiveJob::Base.queue_adapter = :sucker_punch diff --git a/activejob/test/adapters/test.rb b/activejob/test/adapters/test.rb new file mode 100644 index 0000000000..7180b38a57 --- /dev/null +++ b/activejob/test/adapters/test.rb @@ -0,0 +1,3 @@ +ActiveJob::Base.queue_adapter = :test +ActiveJob::Base.queue_adapter.perform_enqueued_jobs = true +ActiveJob::Base.queue_adapter.perform_enqueued_at_jobs = true diff --git a/activejob/test/cases/adapter_test.rb b/activejob/test/cases/adapter_test.rb new file mode 100644 index 0000000000..6d75ae9a7c --- /dev/null +++ b/activejob/test/cases/adapter_test.rb @@ -0,0 +1,7 @@ +require 'helper' + +class AdapterTest < ActiveSupport::TestCase + test "should load #{ENV['AJ_ADAPTER']} adapter" do + assert_equal "active_job/queue_adapters/#{ENV['AJ_ADAPTER']}_adapter".classify, ActiveJob::Base.queue_adapter.class.name + end +end diff --git a/activejob/test/cases/argument_serialization_test.rb b/activejob/test/cases/argument_serialization_test.rb new file mode 100644 index 0000000000..933972a52b --- /dev/null +++ b/activejob/test/cases/argument_serialization_test.rb @@ -0,0 +1,110 @@ +require 'helper' +require 'active_job/arguments' +require 'models/person' +require 'active_support/core_ext/hash/indifferent_access' +require 'jobs/kwargs_job' + +class ArgumentSerializationTest < ActiveSupport::TestCase + setup do + @person = Person.find('5') + end + + [ nil, 1, 1.0, 1_000_000_000_000_000_000_000, + 'a', true, false, + [ 1, 'a' ], + { 'a' => 1 } + ].each do |arg| + test "serializes #{arg.class} verbatim" do + assert_arguments_unchanged arg + end + end + + [ :a, Object.new, self, Person.find('5').to_gid ].each do |arg| + test "does not serialize #{arg.class}" do + assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [ arg ] + end + + assert_raises ActiveJob::DeserializationError do + ActiveJob::Arguments.deserialize [ arg ] + end + end + end + + test 'should convert records to Global IDs' do + assert_arguments_roundtrip [@person] + end + + test 'should dive deep into arrays and hashes' do + assert_arguments_roundtrip [3, [@person]] + assert_arguments_roundtrip [{ 'a' => @person }] + end + + test 'should maintain string and symbol keys' do + assert_arguments_roundtrip([a: 1, "b" => 2]) + end + + test 'should maintain hash with indifferent access' do + symbol_key = { a: 1 } + string_key = { 'a' => 1 } + indifferent_access = { a: 1 }.with_indifferent_access + + assert_not_instance_of ActiveSupport::HashWithIndifferentAccess, perform_round_trip([symbol_key]).first + assert_not_instance_of ActiveSupport::HashWithIndifferentAccess, perform_round_trip([string_key]).first + assert_instance_of ActiveSupport::HashWithIndifferentAccess, perform_round_trip([indifferent_access]).first + end + + test 'should disallow non-string/symbol hash keys' do + assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [ { 1 => 2 } ] + end + + assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [ { :a => [{ 2 => 3 }] } ] + end + + assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [ '_aj_globalid' => 1 ] + end + + assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [ :_aj_globalid => 1 ] + end + end + + test 'should not allow non-primitive objects' do + assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [Object.new] + end + + assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [1, [Object.new]] + end + end + + test 'allows for keyword arguments' do + KwargsJob.perform_later(argument: 2) + + assert_equal "Job with argument: 2", JobBuffer.last_value + end + + test 'raises a friendly SerializationError for records without ids' do + err = assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize [Person.new(nil)] + end + assert_match 'Unable to serialize Person without an id.', err.message + end + + private + def assert_arguments_unchanged(*args) + assert_arguments_roundtrip args + end + + def assert_arguments_roundtrip(args) + assert_equal args, perform_round_trip(args) + end + + def perform_round_trip(args) + ActiveJob::Arguments.deserialize(ActiveJob::Arguments.serialize(args)) + end +end diff --git a/activejob/test/cases/async_job_test.rb b/activejob/test/cases/async_job_test.rb new file mode 100644 index 0000000000..2642cfc608 --- /dev/null +++ b/activejob/test/cases/async_job_test.rb @@ -0,0 +1,42 @@ +require 'helper' +require 'jobs/hello_job' +require 'jobs/queue_as_job' + +class AsyncJobTest < ActiveSupport::TestCase + def using_async_adapter? + ActiveJob::Base.queue_adapter.is_a? ActiveJob::QueueAdapters::AsyncAdapter + end + + setup do + ActiveJob::AsyncJob.perform_asynchronously! + end + + teardown do + ActiveJob::AsyncJob::QUEUES.clear + ActiveJob::AsyncJob.perform_immediately! + end + + test "#create_thread_pool returns a thread_pool" do + thread_pool = ActiveJob::AsyncJob.create_thread_pool + assert thread_pool.is_a? Concurrent::ExecutorService + assert_not thread_pool.is_a? Concurrent::ImmediateExecutor + end + + test "#create_thread_pool returns an ImmediateExecutor after #perform_immediately! is called" do + ActiveJob::AsyncJob.perform_immediately! + thread_pool = ActiveJob::AsyncJob.create_thread_pool + assert thread_pool.is_a? Concurrent::ImmediateExecutor + end + + test "enqueuing without specifying a queue uses the default queue" do + skip unless using_async_adapter? + HelloJob.perform_later + assert ActiveJob::AsyncJob::QUEUES.key? 'default' + end + + test "enqueuing to a queue that does not exist creates the queue" do + skip unless using_async_adapter? + QueueAsJob.perform_later + assert ActiveJob::AsyncJob::QUEUES.key? QueueAsJob::MY_QUEUE.to_s + end +end diff --git a/activejob/test/cases/callbacks_test.rb b/activejob/test/cases/callbacks_test.rb new file mode 100644 index 0000000000..9af2380767 --- /dev/null +++ b/activejob/test/cases/callbacks_test.rb @@ -0,0 +1,23 @@ +require 'helper' +require 'jobs/callback_job' + +require 'active_support/core_ext/object/inclusion' + +class CallbacksTest < ActiveSupport::TestCase + test 'perform callbacks' do + performed_callback_job = CallbackJob.new("A-JOB-ID") + performed_callback_job.perform_now + assert "CallbackJob ran before_perform".in? performed_callback_job.history + assert "CallbackJob ran after_perform".in? performed_callback_job.history + assert "CallbackJob ran around_perform_start".in? performed_callback_job.history + assert "CallbackJob ran around_perform_stop".in? performed_callback_job.history + end + + test 'enqueue callbacks' do + enqueued_callback_job = CallbackJob.perform_later + assert "CallbackJob ran before_enqueue".in? enqueued_callback_job.history + assert "CallbackJob ran after_enqueue".in? enqueued_callback_job.history + assert "CallbackJob ran around_enqueue_start".in? enqueued_callback_job.history + assert "CallbackJob ran around_enqueue_stop".in? enqueued_callback_job.history + end +end diff --git a/activejob/test/cases/job_serialization_test.rb b/activejob/test/cases/job_serialization_test.rb new file mode 100644 index 0000000000..229517774e --- /dev/null +++ b/activejob/test/cases/job_serialization_test.rb @@ -0,0 +1,32 @@ +require 'helper' +require 'jobs/gid_job' +require 'jobs/hello_job' +require 'models/person' + +class JobSerializationTest < ActiveSupport::TestCase + setup do + JobBuffer.clear + @person = Person.find(5) + end + + test 'serialize job with gid' do + GidJob.perform_later @person + assert_equal "Person with ID: 5", JobBuffer.last_value + end + + test 'serialize includes current locale' do + assert_equal :en, HelloJob.new.serialize['locale'] + end + + test 'deserialize sets locale' do + job = HelloJob.new + job.deserialize 'locale' => :es + assert_equal :es, job.locale + end + + test 'deserialize sets default locale' do + job = HelloJob.new + job.deserialize({}) + assert_equal :en, job.locale + end +end diff --git a/activejob/test/cases/logging_test.rb b/activejob/test/cases/logging_test.rb new file mode 100644 index 0000000000..820e9112de --- /dev/null +++ b/activejob/test/cases/logging_test.rb @@ -0,0 +1,122 @@ +require 'helper' +require "active_support/log_subscriber/test_helper" +require 'active_support/core_ext/numeric/time' +require 'jobs/hello_job' +require 'jobs/logging_job' +require 'jobs/nested_job' +require 'models/person' + +class LoggingTest < ActiveSupport::TestCase + include ActiveSupport::LogSubscriber::TestHelper + include ActiveSupport::Logger::Severity + + class TestLogger < ActiveSupport::Logger + def initialize + @file = StringIO.new + super(@file) + end + + def messages + @file.rewind + @file.read + end + end + + def setup + super + JobBuffer.clear + @old_logger = ActiveJob::Base.logger + @logger = ActiveSupport::TaggedLogging.new(TestLogger.new) + set_logger @logger + ActiveJob::Logging::LogSubscriber.attach_to :active_job + end + + def teardown + super + ActiveJob::Logging::LogSubscriber.log_subscribers.pop + set_logger @old_logger + end + + def set_logger(logger) + ActiveJob::Base.logger = logger + end + + + def test_uses_active_job_as_tag + HelloJob.perform_later "Cristian" + assert_match(/\[ActiveJob\]/, @logger.messages) + end + + def test_uses_job_name_as_tag + LoggingJob.perform_later "Dummy" + assert_match(/\[LoggingJob\]/, @logger.messages) + end + + def test_uses_job_id_as_tag + LoggingJob.perform_later "Dummy" + assert_match(/\[LOGGING-JOB-ID\]/, @logger.messages) + end + + def test_logs_correct_queue_name + original_queue_name = LoggingJob.queue_name + LoggingJob.queue_as :php_jobs + LoggingJob.perform_later("Dummy") + assert_match(/to .*?\(php_jobs\).*/, @logger.messages) + ensure + LoggingJob.queue_name = original_queue_name + end + + def test_globalid_parameter_logging + person = Person.new(123) + LoggingJob.perform_later person + assert_match(%r{Enqueued.*gid://aj/Person/123}, @logger.messages) + assert_match(%r{Dummy, here is it: #<Person:.*>}, @logger.messages) + assert_match(%r{Performing.*gid://aj/Person/123}, @logger.messages) + end + + def test_globalid_nested_parameter_logging + person = Person.new(123) + LoggingJob.perform_later(person: person) + assert_match(%r{Enqueued.*gid://aj/Person/123}, @logger.messages) + assert_match(%r{Dummy, here is it: .*#<Person:.*>}, @logger.messages) + assert_match(%r{Performing.*gid://aj/Person/123}, @logger.messages) + end + + def test_enqueue_job_logging + HelloJob.perform_later "Cristian" + assert_match(/Enqueued HelloJob \(Job ID: .*?\) to .*?:.*Cristian/, @logger.messages) + end + + def test_perform_job_logging + LoggingJob.perform_later "Dummy" + assert_match(/Performing LoggingJob from .*? with arguments:.*Dummy/, @logger.messages) + assert_match(/Dummy, here is it: Dummy/, @logger.messages) + assert_match(/Performed LoggingJob from .*? in .*ms/, @logger.messages) + end + + def test_perform_nested_jobs_logging + NestedJob.perform_later + assert_match(/\[LoggingJob\] \[.*?\]/, @logger.messages) + assert_match(/\[ActiveJob\] Enqueued NestedJob \(Job ID: .*\) to/, @logger.messages) + assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Performing NestedJob from/, @logger.messages) + assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Enqueued LoggingJob \(Job ID: .*?\) to .* with arguments: "NestedJob"/, @logger.messages) + assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Performing LoggingJob from .* with arguments: "NestedJob"/, @logger.messages) + assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Dummy, here is it: NestedJob/, @logger.messages) + assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Performed LoggingJob from .* in/, @logger.messages) + assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Performed NestedJob from .* in/, @logger.messages) + end + + def test_enqueue_at_job_logging + HelloJob.set(wait_until: 24.hours.from_now).perform_later "Cristian" + assert_match(/Enqueued HelloJob \(Job ID: .*\) to .*? at.*Cristian/, @logger.messages) + rescue NotImplementedError + skip + end + + def test_enqueue_in_job_logging + HelloJob.set(wait: 2.seconds).perform_later "Cristian" + assert_match(/Enqueued HelloJob \(Job ID: .*\) to .*? at.*Cristian/, @logger.messages) + rescue NotImplementedError + skip + end +end diff --git a/activejob/test/cases/queue_adapter_test.rb b/activejob/test/cases/queue_adapter_test.rb new file mode 100644 index 0000000000..fb3fdc392f --- /dev/null +++ b/activejob/test/cases/queue_adapter_test.rb @@ -0,0 +1,56 @@ +require 'helper' + +module ActiveJob + module QueueAdapters + class StubOneAdapter + def enqueue(*); end + def enqueue_at(*); end + end + + class StubTwoAdapter + def enqueue(*); end + def enqueue_at(*); end + end + end +end + +class QueueAdapterTest < ActiveJob::TestCase + test 'should forbid nonsense arguments' do + assert_raises(ArgumentError) { ActiveJob::Base.queue_adapter = Mutex } + assert_raises(ArgumentError) { ActiveJob::Base.queue_adapter = Mutex.new } + end + + test 'should warn on passing an adapter class' do + klass = Class.new do + def self.name + 'fake' + end + + def enqueue(*); end + def enqueue_at(*); end + end + + assert_deprecated { ActiveJob::Base.queue_adapter = klass } + end + + test 'should allow overriding the queue_adapter at the child class level without affecting the parent or its sibling' do + base_queue_adapter = ActiveJob::Base.queue_adapter + + child_job_one = Class.new(ActiveJob::Base) + child_job_one.queue_adapter = :stub_one + + assert_not_equal ActiveJob::Base.queue_adapter, child_job_one.queue_adapter + assert_kind_of ActiveJob::QueueAdapters::StubOneAdapter, child_job_one.queue_adapter + + child_job_two = Class.new(ActiveJob::Base) + child_job_two.queue_adapter = :stub_two + + assert_kind_of ActiveJob::QueueAdapters::StubTwoAdapter, child_job_two.queue_adapter + assert_kind_of ActiveJob::QueueAdapters::StubOneAdapter, child_job_one.queue_adapter, "child_job_one's queue adapter should remain unchanged" + assert_equal base_queue_adapter, ActiveJob::Base.queue_adapter, "ActiveJob::Base's queue adapter should remain unchanged" + + child_job_three = Class.new(ActiveJob::Base) + + assert_not_nil child_job_three.queue_adapter + end +end diff --git a/activejob/test/cases/queue_naming_test.rb b/activejob/test/cases/queue_naming_test.rb new file mode 100644 index 0000000000..898016a704 --- /dev/null +++ b/activejob/test/cases/queue_naming_test.rb @@ -0,0 +1,102 @@ +require 'helper' +require 'jobs/hello_job' +require 'jobs/logging_job' +require 'jobs/nested_job' + +class QueueNamingTest < ActiveSupport::TestCase + test 'name derived from base' do + assert_equal "default", HelloJob.queue_name + end + + test 'uses given queue name job' do + original_queue_name = HelloJob.queue_name + + begin + HelloJob.queue_as :greetings + assert_equal "greetings", HelloJob.new.queue_name + ensure + HelloJob.queue_name = original_queue_name + end + end + + test 'allows a blank queue name' do + original_queue_name = HelloJob.queue_name + + begin + HelloJob.queue_as "" + assert_equal "", HelloJob.new.queue_name + ensure + HelloJob.queue_name = original_queue_name + end + end + + test 'does not use a nil queue name' do + original_queue_name = HelloJob.queue_name + + begin + HelloJob.queue_as nil + assert_equal "default", HelloJob.new.queue_name + ensure + HelloJob.queue_name = original_queue_name + end + end + + test 'evals block given to queue_as to determine queue' do + original_queue_name = HelloJob.queue_name + + begin + HelloJob.queue_as { :another } + assert_equal "another", HelloJob.new.queue_name + ensure + HelloJob.queue_name = original_queue_name + end + end + + test 'can use arguments to determine queue_name in queue_as block' do + original_queue_name = HelloJob.queue_name + + begin + HelloJob.queue_as { self.arguments.first=='1' ? :one : :two } + assert_equal "one", HelloJob.new('1').queue_name + assert_equal "two", HelloJob.new('3').queue_name + ensure + HelloJob.queue_name = original_queue_name + end + end + + test 'queue_name_prefix prepended to the queue name with default delimiter' do + original_queue_name_prefix = ActiveJob::Base.queue_name_prefix + original_queue_name = HelloJob.queue_name + + begin + ActiveJob::Base.queue_name_prefix = 'aj' + HelloJob.queue_as :low + assert_equal 'aj_low', HelloJob.queue_name + ensure + ActiveJob::Base.queue_name_prefix = original_queue_name_prefix + HelloJob.queue_name = original_queue_name + end + end + + test 'queue_name_prefix prepended to the queue name with custom delimiter' do + original_queue_name_prefix = ActiveJob::Base.queue_name_prefix + original_queue_name_delimiter = ActiveJob::Base.queue_name_delimiter + original_queue_name = HelloJob.queue_name + + begin + ActiveJob::Base.queue_name_delimiter = '.' + ActiveJob::Base.queue_name_prefix = 'aj' + HelloJob.queue_as :low + assert_equal 'aj.low', HelloJob.queue_name + ensure + ActiveJob::Base.queue_name_prefix = original_queue_name_prefix + ActiveJob::Base.queue_name_delimiter = original_queue_name_delimiter + HelloJob.queue_name = original_queue_name + end + end + + test 'uses queue passed to #set' do + job = HelloJob.set(queue: :some_queue).perform_later + assert_equal "some_queue", job.queue_name + end +end diff --git a/activejob/test/cases/queue_priority_test.rb b/activejob/test/cases/queue_priority_test.rb new file mode 100644 index 0000000000..ca17b51dad --- /dev/null +++ b/activejob/test/cases/queue_priority_test.rb @@ -0,0 +1,47 @@ +require 'helper' +require 'jobs/hello_job' + +class QueuePriorityTest < ActiveSupport::TestCase + test 'priority unset by default' do + assert_equal nil, HelloJob.priority + end + + test 'uses given priority' do + original_priority = HelloJob.priority + + begin + HelloJob.queue_with_priority 90 + assert_equal 90, HelloJob.new.priority + ensure + HelloJob.priority = original_priority + end + end + + test 'evals block given to priority to determine priority' do + original_priority = HelloJob.priority + + begin + HelloJob.queue_with_priority { 25 } + assert_equal 25, HelloJob.new.priority + ensure + HelloJob.priority = original_priority + end + end + + test 'can use arguments to determine priority in priority block' do + original_priority = HelloJob.priority + + begin + HelloJob.queue_with_priority { self.arguments.first=='1' ? 99 : 11 } + assert_equal 99, HelloJob.new('1').priority + assert_equal 11, HelloJob.new('3').priority + ensure + HelloJob.priority = original_priority + end + end + + test 'uses priority passed to #set' do + job = HelloJob.set(priority: 123).perform_later + assert_equal 123, job.priority + end +end diff --git a/activejob/test/cases/queuing_test.rb b/activejob/test/cases/queuing_test.rb new file mode 100644 index 0000000000..0eeabbf693 --- /dev/null +++ b/activejob/test/cases/queuing_test.rb @@ -0,0 +1,44 @@ +require 'helper' +require 'jobs/hello_job' +require 'active_support/core_ext/numeric/time' + + +class QueuingTest < ActiveSupport::TestCase + setup do + JobBuffer.clear + end + + test 'run queued job' do + HelloJob.perform_later + assert_equal "David says hello", JobBuffer.last_value + end + + test 'run queued job with arguments' do + HelloJob.perform_later "Jamie" + assert_equal "Jamie says hello", JobBuffer.last_value + end + + test 'run queued job later' do + begin + result = HelloJob.set(wait_until: 1.second.ago).perform_later "Jamie" + assert result + rescue NotImplementedError + skip + end + end + + test 'job returned by enqueue has the arguments available' do + job = HelloJob.perform_later "Jamie" + assert_equal [ "Jamie" ], job.arguments + end + + + test 'job returned by perform_at has the timestamp available' do + begin + job = HelloJob.set(wait_until: Time.utc(2014, 1, 1)).perform_later + assert_equal Time.utc(2014, 1, 1).to_f, job.scheduled_at + rescue NotImplementedError + skip + end + end +end diff --git a/activejob/test/cases/rescue_test.rb b/activejob/test/cases/rescue_test.rb new file mode 100644 index 0000000000..58c9ca8992 --- /dev/null +++ b/activejob/test/cases/rescue_test.rb @@ -0,0 +1,34 @@ +require 'helper' +require 'jobs/rescue_job' +require 'models/person' + +class RescueTest < ActiveSupport::TestCase + setup do + JobBuffer.clear + end + + test 'rescue perform exception with retry' do + job = RescueJob.new("david") + job.perform_now + assert_equal [ "rescued from ArgumentError", "performed beautifully" ], JobBuffer.values + end + + test 'let through unhandled perform exception' do + job = RescueJob.new("other") + assert_raises(RescueJob::OtherError) do + job.perform_now + end + end + + test 'rescue from deserialization errors' do + RescueJob.perform_later Person.new(404) + assert_includes JobBuffer.values, 'rescued from DeserializationError' + assert_includes JobBuffer.values, 'DeserializationError original exception was Person::RecordNotFound' + assert_not_includes JobBuffer.values, 'performed beautifully' + end + + test "should not wrap DeserializationError in DeserializationError" do + RescueJob.perform_later [Person.new(404)] + assert_includes JobBuffer.values, 'DeserializationError original exception was Person::RecordNotFound' + end +end diff --git a/activejob/test/cases/test_case_test.rb b/activejob/test/cases/test_case_test.rb new file mode 100644 index 0000000000..616454a4b6 --- /dev/null +++ b/activejob/test/cases/test_case_test.rb @@ -0,0 +1,23 @@ +require 'helper' +require 'jobs/hello_job' +require 'jobs/logging_job' +require 'jobs/nested_job' + +class ActiveJobTestCaseTest < ActiveJob::TestCase + # this tests that this job class doesn't get its adapter set. + # that's the correct behavior since we don't want to break + # the `class_attribute` inheritance + class TestClassAttributeInheritanceJob < ActiveJob::Base + def self.queue_adapter=(*) + raise 'Attempting to break `class_attribute` inheritance, bad!' + end + end + + def test_include_helper + assert_includes self.class.ancestors, ActiveJob::TestHelper + end + + def test_set_test_adapter + assert_kind_of ActiveJob::QueueAdapters::TestAdapter, self.queue_adapter + end +end diff --git a/activejob/test/cases/test_helper_test.rb b/activejob/test/cases/test_helper_test.rb new file mode 100644 index 0000000000..f7ee763e8a --- /dev/null +++ b/activejob/test/cases/test_helper_test.rb @@ -0,0 +1,511 @@ +require 'helper' +require 'active_support/core_ext/time' +require 'active_support/core_ext/date' +require 'jobs/hello_job' +require 'jobs/logging_job' +require 'jobs/nested_job' +require 'jobs/rescue_job' +require 'models/person' + +class EnqueuedJobsTest < ActiveJob::TestCase + def test_assert_enqueued_jobs + assert_nothing_raised do + assert_enqueued_jobs 1 do + HelloJob.perform_later('david') + end + end + end + + def test_repeated_enqueued_jobs_calls + assert_nothing_raised do + assert_enqueued_jobs 1 do + HelloJob.perform_later('abdelkader') + end + end + + assert_nothing_raised do + assert_enqueued_jobs 2 do + HelloJob.perform_later('sean') + HelloJob.perform_later('yves') + end + end + end + + def test_assert_enqueued_jobs_message + HelloJob.perform_later('sean') + e = assert_raises Minitest::Assertion do + assert_enqueued_jobs 2 do + HelloJob.perform_later('sean') + end + end + assert_match "Expected: 2", e.message + assert_match "Actual: 1", e.message + end + + def test_assert_enqueued_jobs_with_no_block + assert_nothing_raised do + HelloJob.perform_later('rafael') + assert_enqueued_jobs 1 + end + + assert_nothing_raised do + HelloJob.perform_later('aaron') + HelloJob.perform_later('matthew') + assert_enqueued_jobs 3 + end + end + + def test_assert_no_enqueued_jobs_with_no_block + assert_nothing_raised do + assert_no_enqueued_jobs + end + end + + def test_assert_no_enqueued_jobs + assert_nothing_raised do + assert_no_enqueued_jobs do + HelloJob.perform_now + end + end + end + + def test_assert_enqueued_jobs_too_few_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_jobs 2 do + HelloJob.perform_later('xavier') + end + end + + assert_match(/2 .* but 1/, error.message) + end + + def test_assert_enqueued_jobs_too_many_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_jobs 1 do + HelloJob.perform_later('cristian') + HelloJob.perform_later('guillermo') + end + end + + assert_match(/1 .* but 2/, error.message) + end + + def test_assert_no_enqueued_jobs_failure + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_no_enqueued_jobs do + HelloJob.perform_later('jeremy') + end + end + + assert_match(/0 .* but 1/, error.message) + end + + def test_assert_enqueued_jobs_with_only_option + assert_nothing_raised do + assert_enqueued_jobs 1, only: HelloJob do + HelloJob.perform_later('jeremy') + LoggingJob.perform_later + end + end + end + + def test_assert_enqueued_jobs_with_only_option_and_none_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_jobs 1, only: HelloJob do + LoggingJob.perform_later + end + end + + assert_match(/1 .* but 0/, error.message) + end + + def test_assert_enqueued_jobs_with_only_option_and_too_few_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_jobs 5, only: HelloJob do + HelloJob.perform_later('jeremy') + 4.times { LoggingJob.perform_later } + end + end + + assert_match(/5 .* but 1/, error.message) + end + + def test_assert_enqueued_jobs_with_only_option_and_too_many_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_jobs 1, only: HelloJob do + 2.times { HelloJob.perform_later('jeremy') } + end + end + + assert_match(/1 .* but 2/, error.message) + end + + def test_assert_enqueued_jobs_with_only_option_as_array + assert_nothing_raised do + assert_enqueued_jobs 2, only: [HelloJob, LoggingJob] do + HelloJob.perform_later('jeremy') + LoggingJob.perform_later('stewie') + RescueJob.perform_later('david') + end + end + end + + def test_assert_no_enqueued_jobs_with_only_option + assert_nothing_raised do + assert_no_enqueued_jobs only: HelloJob do + LoggingJob.perform_later + end + end + end + + def test_assert_no_enqueued_jobs_with_only_option_failure + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_no_enqueued_jobs only: HelloJob do + HelloJob.perform_later('jeremy') + LoggingJob.perform_later + end + end + + assert_match(/0 .* but 1/, error.message) + end + + def test_assert_no_enqueued_jobs_with_only_option_as_array + assert_nothing_raised do + assert_no_enqueued_jobs only: [HelloJob, RescueJob] do + LoggingJob.perform_later + end + end + end + + def test_assert_enqueued_job + assert_enqueued_with(job: LoggingJob, queue: 'default') do + LoggingJob.set(wait_until: Date.tomorrow.noon).perform_later + end + end + + def test_assert_enqueued_job_returns + job = assert_enqueued_with(job: LoggingJob) do + LoggingJob.set(wait_until: 5.minutes.from_now).perform_later(1, 2, 3) + end + + assert_instance_of LoggingJob, job + assert_in_delta 5.minutes.from_now, job.scheduled_at, 1 + assert_equal 'default', job.queue_name + assert_equal [1, 2, 3], job.arguments + end + + def test_assert_enqueued_job_failure + assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_with(job: LoggingJob, queue: 'default') do + NestedJob.perform_later + end + end + + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_with(job: NestedJob, queue: 'low') do + NestedJob.perform_later + end + end + + assert_equal 'No enqueued job found with {:job=>NestedJob, :queue=>"low"}', error.message + end + + def test_assert_enqueued_job_args + assert_raise ArgumentError do + assert_enqueued_with(class: LoggingJob) do + NestedJob.set(wait_until: Date.tomorrow.noon).perform_later + end + end + end + + def test_assert_enqueued_job_with_at_option + assert_enqueued_with(job: HelloJob, at: Date.tomorrow.noon) do + HelloJob.set(wait_until: Date.tomorrow.noon).perform_later + end + end + + def test_assert_enqueued_job_with_global_id_args + ricardo = Person.new(9) + assert_enqueued_with(job: HelloJob, args: [ricardo]) do + HelloJob.perform_later(ricardo) + end + end + + def test_assert_enqueued_job_failure_with_global_id_args + ricardo = Person.new(9) + wilma = Person.new(11) + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_with(job: HelloJob, args: [wilma]) do + HelloJob.perform_later(ricardo) + end + end + + assert_equal "No enqueued job found with {:job=>HelloJob, :args=>[#{wilma.inspect}]}", error.message + end + + def test_assert_enqueued_job_does_not_change_jobs_count + HelloJob.perform_later + assert_enqueued_with(job: HelloJob) do + HelloJob.perform_later + end + + assert_equal 2, ActiveJob::Base.queue_adapter.enqueued_jobs.count + end +end + +class PerformedJobsTest < ActiveJob::TestCase + def test_performed_enqueue_jobs_with_only_option_doesnt_leak_outside_the_block + assert_equal nil, queue_adapter.filter + perform_enqueued_jobs only: HelloJob do + assert_equal HelloJob, queue_adapter.filter + end + assert_equal nil, queue_adapter.filter + end + + def test_assert_performed_jobs + assert_nothing_raised do + assert_performed_jobs 1 do + HelloJob.perform_later('david') + end + end + end + + def test_repeated_performed_jobs_calls + assert_nothing_raised do + assert_performed_jobs 1 do + HelloJob.perform_later('abdelkader') + end + end + + assert_nothing_raised do + assert_performed_jobs 2 do + HelloJob.perform_later('sean') + HelloJob.perform_later('yves') + end + end + end + + def test_assert_performed_jobs_message + HelloJob.perform_later('sean') + e = assert_raises Minitest::Assertion do + assert_performed_jobs 2 do + HelloJob.perform_later('sean') + end + end + assert_match "Expected: 2", e.message + assert_match "Actual: 1", e.message + end + + def test_assert_performed_jobs_with_no_block + assert_nothing_raised do + perform_enqueued_jobs do + HelloJob.perform_later('rafael') + end + assert_performed_jobs 1 + end + + assert_nothing_raised do + perform_enqueued_jobs do + HelloJob.perform_later('aaron') + HelloJob.perform_later('matthew') + assert_performed_jobs 3 + end + end + end + + def test_assert_no_performed_jobs_with_no_block + assert_nothing_raised do + assert_no_performed_jobs + end + end + + def test_assert_no_performed_jobs + assert_nothing_raised do + assert_no_performed_jobs do + # empty block won't perform jobs + end + end + end + + def test_assert_performed_jobs_too_few_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_jobs 2 do + HelloJob.perform_later('xavier') + end + end + + assert_match(/2 .* but 1/, error.message) + end + + def test_assert_performed_jobs_too_many_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_jobs 1 do + HelloJob.perform_later('cristian') + HelloJob.perform_later('guillermo') + end + end + + assert_match(/1 .* but 2/, error.message) + end + + def test_assert_no_performed_jobs_failure + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_no_performed_jobs do + HelloJob.perform_later('jeremy') + end + end + + assert_match(/0 .* but 1/, error.message) + end + + def test_assert_performed_jobs_with_only_option + assert_nothing_raised do + assert_performed_jobs 1, only: HelloJob do + HelloJob.perform_later('jeremy') + LoggingJob.perform_later + end + end + end + + def test_assert_performed_jobs_with_only_option_as_array + 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 test_assert_performed_jobs_with_only_option_and_none_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_jobs 1, only: HelloJob do + LoggingJob.perform_later + end + end + + assert_match(/1 .* but 0/, error.message) + end + + def test_assert_performed_jobs_with_only_option_and_too_few_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_jobs 5, only: HelloJob do + HelloJob.perform_later('jeremy') + 4.times { LoggingJob.perform_later } + end + end + + assert_match(/5 .* but 1/, error.message) + end + + def test_assert_performed_jobs_with_only_option_and_too_many_sent + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_jobs 1, only: HelloJob do + 2.times { HelloJob.perform_later('jeremy') } + end + end + + assert_match(/1 .* but 2/, error.message) + end + + def test_assert_no_performed_jobs_with_only_option + assert_nothing_raised do + assert_no_performed_jobs only: HelloJob do + LoggingJob.perform_later + end + end + end + + def test_assert_no_performed_jobs_with_only_option_as_array + assert_nothing_raised do + assert_no_performed_jobs only: [HelloJob, RescueJob] do + LoggingJob.perform_later + end + end + end + + def test_assert_no_performed_jobs_with_only_option_failure + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_no_performed_jobs only: HelloJob do + HelloJob.perform_later('jeremy') + LoggingJob.perform_later + end + end + + assert_match(/0 .* but 1/, error.message) + end + + def test_assert_performed_job + assert_performed_with(job: NestedJob, queue: 'default') do + NestedJob.perform_later + end + end + + def test_assert_performed_job_returns + job = assert_performed_with(job: NestedJob, queue: 'default') do + NestedJob.perform_later + end + + assert_instance_of NestedJob, job + assert_nil job.scheduled_at + assert_equal [], job.arguments + assert_equal 'default', job.queue_name + end + + def test_assert_performed_job_failure + assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_with(job: LoggingJob) do + HelloJob.perform_later + end + end + + assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_with(job: HelloJob, queue: 'low') do + HelloJob.set(queue: 'important').perform_later + end + end + end + + def test_assert_performed_job_with_at_option + assert_performed_with(job: HelloJob, at: Date.tomorrow.noon) do + HelloJob.set(wait_until: Date.tomorrow.noon).perform_later + end + + assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_with(job: HelloJob, at: Date.today.noon) do + HelloJob.set(wait_until: Date.tomorrow.noon).perform_later + end + end + end + + def test_assert_performed_job_with_global_id_args + ricardo = Person.new(9) + assert_performed_with(job: HelloJob, args: [ricardo]) do + HelloJob.perform_later(ricardo) + end + end + + def test_assert_performed_job_failure_with_global_id_args + ricardo = Person.new(9) + wilma = Person.new(11) + error = assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_with(job: HelloJob, args: [wilma]) do + HelloJob.perform_later(ricardo) + end + end + + assert_equal "No performed job found with {:job=>HelloJob, :args=>[#{wilma.inspect}]}", error.message + end + + def test_assert_performed_job_does_not_change_jobs_count + assert_performed_with(job: HelloJob) do + HelloJob.perform_later + end + + assert_performed_with(job: HelloJob) do + HelloJob.perform_later + end + + assert_equal 2, ActiveJob::Base.queue_adapter.performed_jobs.count + end +end diff --git a/activejob/test/cases/translation_test.rb b/activejob/test/cases/translation_test.rb new file mode 100644 index 0000000000..d5e3aaf9e3 --- /dev/null +++ b/activejob/test/cases/translation_test.rb @@ -0,0 +1,20 @@ +require 'helper' +require 'jobs/translated_hello_job' + +class TranslationTest < ActiveSupport::TestCase + setup do + JobBuffer.clear + I18n.available_locales = [:en, :de] + @job = TranslatedHelloJob.new('Johannes') + end + + teardown do + I18n.available_locales = [:en] + end + + test 'it performs the job in the given locale' do + @job.locale = :de + @job.perform_now + assert_equal "Johannes says Guten Tag", JobBuffer.last_value + end +end diff --git a/activejob/test/helper.rb b/activejob/test/helper.rb new file mode 100644 index 0000000000..7e86415f48 --- /dev/null +++ b/activejob/test/helper.rb @@ -0,0 +1,18 @@ +require File.expand_path('../../../load_paths', __FILE__) + +require 'active_job' +require 'support/job_buffer' + +ActiveSupport.halt_callback_chains_on_return_false = false +GlobalID.app = 'aj' + +@adapter = ENV['AJ_ADAPTER'] || 'inline' + +if ENV['AJ_INTEGRATION_TESTS'] + require 'support/integration/helper' +else + ActiveJob::Base.logger = Logger.new(nil) + require "adapters/#{@adapter}" +end + +require 'active_support/testing/autorun' diff --git a/activejob/test/integration/queuing_test.rb b/activejob/test/integration/queuing_test.rb new file mode 100644 index 0000000000..e435ed4aa6 --- /dev/null +++ b/activejob/test/integration/queuing_test.rb @@ -0,0 +1,99 @@ +require 'helper' +require 'jobs/logging_job' +require 'jobs/hello_job' +require 'active_support/core_ext/numeric/time' + +class QueuingTest < ActiveSupport::TestCase + test 'should run jobs enqueued on a listening queue' do + TestJob.perform_later @id + wait_for_jobs_to_finish_for(5.seconds) + assert job_executed + end + + test 'should not run jobs queued on a non-listening queue' do + skip if adapter_is?(:inline, :async, :sucker_punch, :que) + old_queue = TestJob.queue_name + + begin + TestJob.queue_as :some_other_queue + TestJob.perform_later @id + wait_for_jobs_to_finish_for(2.seconds) + assert_not job_executed + ensure + TestJob.queue_name = old_queue + end + end + + test 'should supply a wrapped class name to Sidekiq' do + skip unless adapter_is?(:sidekiq) + Sidekiq::Testing.fake! do + ::HelloJob.perform_later + hash = ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper.jobs.first + assert_equal "ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper", hash['class'] + assert_equal "HelloJob", hash['wrapped'] + end + end + + test 'should not run job enqueued in the future' do + begin + TestJob.set(wait: 10.minutes).perform_later @id + wait_for_jobs_to_finish_for(5.seconds) + assert_not job_executed + rescue NotImplementedError + skip + end + end + + test 'should run job enqueued in the future at the specified time' do + begin + TestJob.set(wait: 5.seconds).perform_later @id + wait_for_jobs_to_finish_for(2.seconds) + assert_not job_executed + wait_for_jobs_to_finish_for(10.seconds) + assert job_executed + rescue NotImplementedError + skip + end + end + + test 'should supply a provider_job_id when available for immediate jobs' do + skip unless adapter_is?(:delayed_job, :sidekiq, :qu, :que, :queue_classic) + test_job = TestJob.perform_later @id + assert test_job.provider_job_id, 'Provider job id should be set by provider' + end + + test 'should supply a provider_job_id when available for delayed jobs' do + skip unless adapter_is?(:delayed_job, :sidekiq, :que, :queue_classic) + delayed_test_job = TestJob.set(wait: 1.minute).perform_later @id + assert delayed_test_job.provider_job_id, 'Provider job id should by set for delayed jobs by provider' + end + + test 'current locale is kept while running perform_later' do + skip if adapter_is?(:inline) + + begin + I18n.available_locales = [:en, :de] + I18n.locale = :de + + TestJob.perform_later @id + wait_for_jobs_to_finish_for(5.seconds) + assert job_executed + assert_equal 'de', job_output + ensure + I18n.available_locales = [:en] + I18n.locale = :en + end + end + + test 'should run job with higher priority first' do + skip unless adapter_is?(:delayed_job, :que) + + wait_until = Time.now + 3.seconds + TestJob.set(wait_until: wait_until, priority: 20).perform_later "#{@id}.1" + TestJob.set(wait_until: wait_until, priority: 10).perform_later "#{@id}.2" + wait_for_jobs_to_finish_for(10.seconds) + assert job_executed "#{@id}.1" + assert job_executed "#{@id}.2" + assert job_executed_at("#{@id}.2") < job_executed_at("#{@id}.1") + end +end diff --git a/activejob/test/jobs/callback_job.rb b/activejob/test/jobs/callback_job.rb new file mode 100644 index 0000000000..891ed9464e --- /dev/null +++ b/activejob/test/jobs/callback_job.rb @@ -0,0 +1,29 @@ +class CallbackJob < ActiveJob::Base + before_perform ->(job) { job.history << "CallbackJob ran before_perform" } + after_perform ->(job) { job.history << "CallbackJob ran after_perform" } + + before_enqueue ->(job) { job.history << "CallbackJob ran before_enqueue" } + after_enqueue ->(job) { job.history << "CallbackJob ran after_enqueue" } + + around_perform do |job, block| + job.history << "CallbackJob ran around_perform_start" + block.call + job.history << "CallbackJob ran around_perform_stop" + end + + around_enqueue do |job, block| + job.history << "CallbackJob ran around_enqueue_start" + block.call + job.history << "CallbackJob ran around_enqueue_stop" + end + + + def perform(person = "david") + # NOTHING! + end + + def history + @history ||= [] + end + +end diff --git a/activejob/test/jobs/gid_job.rb b/activejob/test/jobs/gid_job.rb new file mode 100644 index 0000000000..e485bfa2dd --- /dev/null +++ b/activejob/test/jobs/gid_job.rb @@ -0,0 +1,8 @@ +require_relative '../support/job_buffer' + +class GidJob < ActiveJob::Base + def perform(person) + JobBuffer.add("Person with ID: #{person.id}") + end +end + diff --git a/activejob/test/jobs/hello_job.rb b/activejob/test/jobs/hello_job.rb new file mode 100644 index 0000000000..022fa58e4a --- /dev/null +++ b/activejob/test/jobs/hello_job.rb @@ -0,0 +1,7 @@ +require_relative '../support/job_buffer' + +class HelloJob < ActiveJob::Base + def perform(greeter = "David") + JobBuffer.add("#{greeter} says hello") + end +end diff --git a/activejob/test/jobs/kwargs_job.rb b/activejob/test/jobs/kwargs_job.rb new file mode 100644 index 0000000000..2df17d15ae --- /dev/null +++ b/activejob/test/jobs/kwargs_job.rb @@ -0,0 +1,7 @@ +require_relative '../support/job_buffer' + +class KwargsJob < ActiveJob::Base + def perform(argument: 1) + JobBuffer.add("Job with argument: #{argument}") + end +end diff --git a/activejob/test/jobs/logging_job.rb b/activejob/test/jobs/logging_job.rb new file mode 100644 index 0000000000..d84ed8589b --- /dev/null +++ b/activejob/test/jobs/logging_job.rb @@ -0,0 +1,10 @@ +class LoggingJob < ActiveJob::Base + def perform(dummy) + logger.info "Dummy, here is it: #{dummy}" + end + + def job_id + "LOGGING-JOB-ID" + end +end + diff --git a/activejob/test/jobs/nested_job.rb b/activejob/test/jobs/nested_job.rb new file mode 100644 index 0000000000..8c4ec549a6 --- /dev/null +++ b/activejob/test/jobs/nested_job.rb @@ -0,0 +1,10 @@ +class NestedJob < ActiveJob::Base + def perform + LoggingJob.perform_later "NestedJob" + end + + def job_id + "NESTED-JOB-ID" + end +end + diff --git a/activejob/test/jobs/queue_as_job.rb b/activejob/test/jobs/queue_as_job.rb new file mode 100644 index 0000000000..897aef52e5 --- /dev/null +++ b/activejob/test/jobs/queue_as_job.rb @@ -0,0 +1,10 @@ +require_relative '../support/job_buffer' + +class QueueAsJob < ActiveJob::Base + MY_QUEUE = :low_priority + queue_as MY_QUEUE + + def perform(greeter = "David") + JobBuffer.add("#{greeter} says hello") + end +end diff --git a/activejob/test/jobs/rescue_job.rb b/activejob/test/jobs/rescue_job.rb new file mode 100644 index 0000000000..f1b9c9349e --- /dev/null +++ b/activejob/test/jobs/rescue_job.rb @@ -0,0 +1,27 @@ +require_relative '../support/job_buffer' + +class RescueJob < ActiveJob::Base + class OtherError < StandardError; end + + rescue_from(ArgumentError) do + JobBuffer.add('rescued from ArgumentError') + arguments[0] = "DIFFERENT!" + retry_job + end + + rescue_from(ActiveJob::DeserializationError) do |e| + JobBuffer.add('rescued from DeserializationError') + JobBuffer.add("DeserializationError original exception was #{e.original_exception.class.name}") + end + + def perform(person = "david") + case person + when "david" + raise ArgumentError, "Hair too good" + when "other" + raise OtherError + else + JobBuffer.add('performed beautifully') + end + end +end diff --git a/activejob/test/jobs/translated_hello_job.rb b/activejob/test/jobs/translated_hello_job.rb new file mode 100644 index 0000000000..9657cd3f54 --- /dev/null +++ b/activejob/test/jobs/translated_hello_job.rb @@ -0,0 +1,10 @@ +require_relative '../support/job_buffer' + +class TranslatedHelloJob < ActiveJob::Base + def perform(greeter = "David") + translations = { en: 'Hello', de: 'Guten Tag' } + hello = translations[I18n.locale] + + JobBuffer.add("#{greeter} says #{hello}") + end +end diff --git a/activejob/test/models/person.rb b/activejob/test/models/person.rb new file mode 100644 index 0000000000..76a8f40616 --- /dev/null +++ b/activejob/test/models/person.rb @@ -0,0 +1,20 @@ +class Person + class RecordNotFound < StandardError; end + + include GlobalID::Identification + + attr_reader :id + + def self.find(id) + raise RecordNotFound.new("Cannot find person with ID=404") if id.to_i==404 + new(id) + end + + def initialize(id) + @id = id + end + + def ==(other_person) + other_person.is_a?(Person) && id.to_s == other_person.id.to_s + end +end diff --git a/activejob/test/support/backburner/inline.rb b/activejob/test/support/backburner/inline.rb new file mode 100644 index 0000000000..f761b53e27 --- /dev/null +++ b/activejob/test/support/backburner/inline.rb @@ -0,0 +1,8 @@ +require 'backburner' + +Backburner::Worker.class_eval do + class << self; alias_method :original_enqueue, :enqueue; end + def self.enqueue(job_class, args=[], opts={}) + job_class.perform(*args) + end +end
\ No newline at end of file diff --git a/activejob/test/support/delayed_job/delayed/backend/test.rb b/activejob/test/support/delayed_job/delayed/backend/test.rb new file mode 100644 index 0000000000..f80ec3a5a6 --- /dev/null +++ b/activejob/test/support/delayed_job/delayed/backend/test.rb @@ -0,0 +1,111 @@ +#copied from https://github.com/collectiveidea/delayed_job/blob/master/spec/delayed/backend/test.rb +require 'ostruct' + +# An in-memory backend suitable only for testing. Tries to behave as if it were an ORM. +module Delayed + module Backend + module Test + class Job + attr_accessor :id + attr_accessor :priority + attr_accessor :attempts + attr_accessor :handler + attr_accessor :last_error + attr_accessor :run_at + attr_accessor :locked_at + attr_accessor :locked_by + attr_accessor :failed_at + attr_accessor :queue + + include Delayed::Backend::Base + + cattr_accessor :id + self.id = 0 + + def initialize(hash = {}) + self.attempts = 0 + self.priority = 0 + self.id = (self.class.id += 1) + hash.each{|k,v| send(:"#{k}=", v)} + end + + @jobs = [] + def self.all + @jobs + end + + def self.count + all.size + end + + def self.delete_all + all.clear + end + + def self.create(attrs = {}) + new(attrs).tap(&:save) + end + + def self.create!(*args); create(*args); end + + def self.clear_locks!(worker_name) + all.select{|j| j.locked_by == worker_name}.each {|j| j.locked_by = nil; j.locked_at = nil} + end + + # Find a few candidate jobs to run (in case some immediately get locked by others). + def self.find_available(worker_name, limit = 5, max_run_time = Worker.max_run_time) + jobs = all.select do |j| + j.run_at <= db_time_now && + (j.locked_at.nil? || j.locked_at < db_time_now - max_run_time || j.locked_by == worker_name) && + !j.failed? + end + + jobs = jobs.select{|j| Worker.queues.include?(j.queue)} if Worker.queues.any? + jobs = jobs.select{|j| j.priority >= Worker.min_priority} if Worker.min_priority + jobs = jobs.select{|j| j.priority <= Worker.max_priority} if Worker.max_priority + jobs.sort_by{|j| [j.priority, j.run_at]}[0..limit-1] + end + + # Lock this job for this worker. + # Returns true if we have the lock, false otherwise. + def lock_exclusively!(max_run_time, worker) + now = self.class.db_time_now + if locked_by != worker + # We don't own this job so we will update the locked_by name and the locked_at + self.locked_at = now + self.locked_by = worker + end + + return true + end + + def self.db_time_now + Time.current + end + + def update_attributes(attrs = {}) + attrs.each{|k,v| send(:"#{k}=", v)} + save + end + + def destroy + self.class.all.delete(self) + end + + def save + self.run_at ||= Time.current + + self.class.all << self unless self.class.all.include?(self) + true + end + + def save!; save; end + + def reload + reset + self + end + end + end + end +end diff --git a/activejob/test/support/delayed_job/delayed/serialization/test.rb b/activejob/test/support/delayed_job/delayed/serialization/test.rb new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/activejob/test/support/delayed_job/delayed/serialization/test.rb diff --git a/activejob/test/support/integration/adapters/async.rb b/activejob/test/support/integration/adapters/async.rb new file mode 100644 index 0000000000..42beb12b1f --- /dev/null +++ b/activejob/test/support/integration/adapters/async.rb @@ -0,0 +1,9 @@ +module AsyncJobsManager + def setup + ActiveJob::Base.queue_adapter = :async + end + + def clear_jobs + ActiveJob::AsyncJob::QUEUES.clear + end +end diff --git a/activejob/test/support/integration/adapters/backburner.rb b/activejob/test/support/integration/adapters/backburner.rb new file mode 100644 index 0000000000..2e82562948 --- /dev/null +++ b/activejob/test/support/integration/adapters/backburner.rb @@ -0,0 +1,38 @@ +module BackburnerJobsManager + def setup + ActiveJob::Base.queue_adapter = :backburner + Backburner.configure do |config| + config.logger = Rails.logger + end + unless can_run? + puts "Cannot run integration tests for backburner. To be able to run integration tests for backburner you need to install and start beanstalkd.\n" + exit + end + end + + def clear_jobs + tube.clear + end + + def start_workers + @thread = Thread.new { Backburner.work "integration-tests" } # backburner dasherizes the queue name + end + + def stop_workers + @thread.kill + end + + def tube + @tube ||= Beaneater::Tube.new(Backburner::Worker.connection, "backburner.worker.queue.integration-tests") # backburner dasherizes the queue name + end + + def can_run? + begin + Backburner::Worker.connection.send :connect! + rescue + return false + end + true + end +end + diff --git a/activejob/test/support/integration/adapters/delayed_job.rb b/activejob/test/support/integration/adapters/delayed_job.rb new file mode 100644 index 0000000000..0b591964bc --- /dev/null +++ b/activejob/test/support/integration/adapters/delayed_job.rb @@ -0,0 +1,20 @@ +require 'delayed_job' +require 'delayed_job_active_record' + +module DelayedJobJobsManager + def setup + ActiveJob::Base.queue_adapter = :delayed_job + end + def clear_jobs + Delayed::Job.delete_all + end + + def start_workers + @worker = Delayed::Worker.new(quiet: true, sleep_delay: 0.5, queues: %w(integration_tests)) + @thread = Thread.new { @worker.start } + end + + def stop_workers + @worker.stop + end +end diff --git a/activejob/test/support/integration/adapters/inline.rb b/activejob/test/support/integration/adapters/inline.rb new file mode 100644 index 0000000000..83c38f706f --- /dev/null +++ b/activejob/test/support/integration/adapters/inline.rb @@ -0,0 +1,15 @@ +module InlineJobsManager + def setup + ActiveJob::Base.queue_adapter = :inline + end + + def clear_jobs + end + + def start_workers + end + + def stop_workers + end +end + diff --git a/activejob/test/support/integration/adapters/qu.rb b/activejob/test/support/integration/adapters/qu.rb new file mode 100644 index 0000000000..256ddb3cf3 --- /dev/null +++ b/activejob/test/support/integration/adapters/qu.rb @@ -0,0 +1,38 @@ +module QuJobsManager + def setup + require 'qu-rails' + require 'qu-redis' + ActiveJob::Base.queue_adapter = :qu + ENV['REDISTOGO_URL'] = "redis://127.0.0.1:6379/12" + backend = Qu::Backend::Redis.new + backend.namespace = "active_jobs_int_test" + Qu.backend = backend + Qu.logger = Rails.logger + Qu.interval = 0.5 + unless can_run? + puts "Cannot run integration tests for qu. To be able to run integration tests for qu you need to install and start redis.\n" + exit + end + end + + def clear_jobs + Qu.clear "integration_tests" + end + + def start_workers + @thread = Thread.new { Qu::Worker.new("integration_tests").start } + end + + def stop_workers + @thread.kill + end + + def can_run? + begin + Qu.backend.connection.client.connect + rescue + return false + end + true + end +end diff --git a/activejob/test/support/integration/adapters/que.rb b/activejob/test/support/integration/adapters/que.rb new file mode 100644 index 0000000000..0cd8952a28 --- /dev/null +++ b/activejob/test/support/integration/adapters/que.rb @@ -0,0 +1,39 @@ +module QueJobsManager + def setup + require 'sequel' + ActiveJob::Base.queue_adapter = :que + Que.mode = :off + Que.worker_count = 1 + end + + def clear_jobs + Que.clear! + end + + def start_workers + que_url = ENV['QUE_DATABASE_URL'] || 'postgres:///active_jobs_que_int_test' + uri = URI.parse(que_url) + user = uri.user||ENV['USER'] + pass = uri.password + db = uri.path[1..-1] + %x{#{"PGPASSWORD=\"#{pass}\"" if pass} psql -c 'drop database if exists "#{db}"' -U #{user} -t template1} + %x{#{"PGPASSWORD=\"#{pass}\"" if pass} psql -c 'create database "#{db}"' -U #{user} -t template1} + Que.connection = Sequel.connect(que_url) + Que.migrate! + + @thread = Thread.new do + loop do + Que::Job.work + sleep 0.5 + end + end + + rescue Sequel::DatabaseConnectionError + puts "Cannot run integration tests for que. To be able to run integration tests for que you need to install and start postgresql.\n" + exit + end + + def stop_workers + @thread.kill + end +end diff --git a/activejob/test/support/integration/adapters/queue_classic.rb b/activejob/test/support/integration/adapters/queue_classic.rb new file mode 100644 index 0000000000..29c04bf625 --- /dev/null +++ b/activejob/test/support/integration/adapters/queue_classic.rb @@ -0,0 +1,37 @@ +module QueueClassicJobsManager + def setup + ENV['QC_DATABASE_URL'] ||= 'postgres:///active_jobs_qc_int_test' + ENV['QC_RAILS_DATABASE'] = 'false' + ENV['QC_LISTEN_TIME'] = "0.5" + ActiveJob::Base.queue_adapter = :queue_classic + end + + def clear_jobs + QC::Queue.new("integration_tests").delete_all + end + + def start_workers + uri = URI.parse(ENV['QC_DATABASE_URL']) + user = uri.user||ENV['USER'] + pass = uri.password + db = uri.path[1..-1] + %x{#{"PGPASSWORD=\"#{pass}\"" if pass} psql -c 'drop database if exists "#{db}"' -U #{user} -t template1} + %x{#{"PGPASSWORD=\"#{pass}\"" if pass} psql -c 'create database "#{db}"' -U #{user} -t template1} + QC::Setup.create + + QC.default_conn_adapter.disconnect + QC.default_conn_adapter = nil + @pid = fork do + worker = QC::Worker.new(q_name: 'integration_tests') + worker.start + end + + rescue PG::ConnectionBad + puts "Cannot run integration tests for queue_classic. To be able to run integration tests for queue_classic you need to install and start postgresql.\n" + exit + end + + def stop_workers + Process.kill 'HUP', @pid + end +end diff --git a/activejob/test/support/integration/adapters/resque.rb b/activejob/test/support/integration/adapters/resque.rb new file mode 100644 index 0000000000..912f4bc387 --- /dev/null +++ b/activejob/test/support/integration/adapters/resque.rb @@ -0,0 +1,49 @@ +module ResqueJobsManager + def setup + ActiveJob::Base.queue_adapter = :resque + Resque.redis = Redis::Namespace.new 'active_jobs_int_test', redis: Redis.connect(url: "redis://127.0.0.1:6379/12", :thread_safe => true) + Resque.logger = Rails.logger + unless can_run? + puts "Cannot run integration tests for resque. To be able to run integration tests for resque you need to install and start redis.\n" + exit + end + end + + def clear_jobs + Resque.queues.each { |queue_name| Resque.redis.del "queue:#{queue_name}" } + Resque.redis.keys("delayed:*").each { |key| Resque.redis.del "#{key}" } + Resque.redis.del "delayed_queue_schedule" + end + + def start_workers + @resque_thread = Thread.new do + w = Resque::Worker.new("integration_tests") + w.term_child = true + w.work(0.5) + end + @scheduler_thread = Thread.new do + Resque::Scheduler.configure do |c| + c.poll_sleep_amount = 0.5 + c.dynamic = true + c.quiet = true + c.logfile = nil + end + Resque::Scheduler.master_lock.release! + Resque::Scheduler.run + end + end + + def stop_workers + @resque_thread.kill + @scheduler_thread.kill + end + + def can_run? + begin + Resque.redis.client.connect + rescue + return false + end + true + end +end diff --git a/activejob/test/support/integration/adapters/sidekiq.rb b/activejob/test/support/integration/adapters/sidekiq.rb new file mode 100644 index 0000000000..9aa07bcb52 --- /dev/null +++ b/activejob/test/support/integration/adapters/sidekiq.rb @@ -0,0 +1,98 @@ +require 'sidekiq/api' + +require 'sidekiq/testing' +Sidekiq::Testing.disable! + +module SidekiqJobsManager + + def setup + ActiveJob::Base.queue_adapter = :sidekiq + unless can_run? + puts "Cannot run integration tests for sidekiq. To be able to run integration tests for sidekiq you need to install and start redis.\n" + exit + end + end + + def clear_jobs + Sidekiq::ScheduledSet.new.clear + Sidekiq::Queue.new("integration_tests").clear + end + + def start_workers + continue_read, continue_write = IO.pipe + death_read, death_write = IO.pipe + + @pid = fork do + continue_read.close + death_write.close + + # Celluloid & Sidekiq are not warning-clean :( + $VERBOSE = false + + $stdin.reopen('/dev/null') + $stdout.sync = true + $stderr.sync = true + + logfile = Rails.root.join("log/sidekiq.log").to_s + Sidekiq::Logging.initialize_logger(logfile) + + self_read, self_write = IO.pipe + trap "TERM" do + self_write.puts("TERM") + end + + Thread.new do + begin + death_read.read + rescue Exception + end + self_write.puts("TERM") + end + + require 'celluloid' + Celluloid.logger = nil + require 'sidekiq/launcher' + sidekiq = Sidekiq::Launcher.new({queues: ["integration_tests"], + environment: "test", + concurrency: 1, + timeout: 1, + }) + Sidekiq.average_scheduled_poll_interval = 0.5 + Sidekiq.options[:poll_interval_average] = 1 + begin + sidekiq.run + continue_write.puts "started" + while readable_io = IO.select([self_read]) + signal = readable_io.first[0].gets.strip + raise Interrupt if signal == "TERM" + end + rescue Interrupt + end + + sidekiq.stop + exit! + end + continue_write.close + death_read.close + @worker_lifeline = death_write + + raise "Failed to start worker" unless continue_read.gets == "started\n" + end + + def stop_workers + if @pid + Process.kill 'TERM', @pid + Process.wait @pid + end + end + + def can_run? + begin + Sidekiq.redis(&:info) + Sidekiq.logger = nil + rescue + return false + end + true + end +end diff --git a/activejob/test/support/integration/adapters/sneakers.rb b/activejob/test/support/integration/adapters/sneakers.rb new file mode 100644 index 0000000000..875803a2d8 --- /dev/null +++ b/activejob/test/support/integration/adapters/sneakers.rb @@ -0,0 +1,90 @@ +require 'sneakers/runner' +require 'sneakers/publisher' +require 'timeout' + +module Sneakers + class Publisher + def safe_ensure_connected + @mutex.synchronize do + ensure_connection! unless connected? + end + end + end +end + + +module SneakersJobsManager + def setup + ActiveJob::Base.queue_adapter = :sneakers + Sneakers.configure :heartbeat => 2, + :amqp => 'amqp://guest:guest@localhost:5672', + :vhost => '/', + :exchange => 'active_jobs_sneakers_int_test', + :exchange_type => :direct, + :daemonize => true, + :threads => 1, + :workers => 1, + :pid_path => Rails.root.join("tmp/sneakers.pid").to_s, + :log => Rails.root.join("log/sneakers.log").to_s + unless can_run? + puts "Cannot run integration tests for sneakers. To be able to run integration tests for sneakers you need to install and start rabbitmq.\n" + exit + end + end + + def clear_jobs + bunny_queue.purge + end + + def start_workers + @pid = fork do + queues = %w(integration_tests) + workers = queues.map do |q| + worker_klass = "ActiveJobWorker"+Digest::MD5.hexdigest(q) + Sneakers.const_set(worker_klass, Class.new(ActiveJob::QueueAdapters::SneakersAdapter::JobWrapper) do + from_queue q + end) + end + Sneakers::Runner.new(workers).run + end + begin + Timeout.timeout(10) do + while bunny_queue.status[:consumer_count] == 0 + sleep 0.5 + end + end + rescue Timeout::Error + stop_workers + raise "Failed to start sneakers worker" + end + end + + def stop_workers + Process.kill 'TERM', @pid + Process.kill 'TERM', File.open(Rails.root.join("tmp/sneakers.pid").to_s).read.to_i + rescue + end + + def can_run? + begin + bunny_publisher + rescue + return false + end + true + end + + protected + def bunny_publisher + @bunny_publisher ||= begin + p = ActiveJob::QueueAdapters::SneakersAdapter::JobWrapper.send(:publisher) + p.safe_ensure_connected + p + end + end + + def bunny_queue + @queue ||= bunny_publisher.exchange.channel.queue "integration_tests", durable: true + end + +end diff --git a/activejob/test/support/integration/adapters/sucker_punch.rb b/activejob/test/support/integration/adapters/sucker_punch.rb new file mode 100644 index 0000000000..9c0d66b469 --- /dev/null +++ b/activejob/test/support/integration/adapters/sucker_punch.rb @@ -0,0 +1,6 @@ +module SuckerPunchJobsManager + def setup + ActiveJob::Base.queue_adapter = :sucker_punch + SuckerPunch.logger = nil + end +end diff --git a/activejob/test/support/integration/dummy_app_template.rb b/activejob/test/support/integration/dummy_app_template.rb new file mode 100644 index 0000000000..0c062a025e --- /dev/null +++ b/activejob/test/support/integration/dummy_app_template.rb @@ -0,0 +1,26 @@ +if ENV['AJ_ADAPTER'] == 'delayed_job' + generate "delayed_job:active_record", "--quiet" +end + +rake("db:migrate") + +initializer 'activejob.rb', <<-CODE +require "#{File.expand_path("../jobs_manager.rb", __FILE__)}" +JobsManager.current_manager.setup +CODE + +initializer 'i18n.rb', <<-CODE +I18n.available_locales = [:en, :de] +CODE + +file 'app/jobs/test_job.rb', <<-CODE +class TestJob < ActiveJob::Base + queue_as :integration_tests + + def perform(x) + File.open(Rails.root.join("tmp/\#{x}"), "w+") do |f| + f.write I18n.locale + end + end +end +CODE diff --git a/activejob/test/support/integration/helper.rb b/activejob/test/support/integration/helper.rb new file mode 100644 index 0000000000..4a1b0bfbcb --- /dev/null +++ b/activejob/test/support/integration/helper.rb @@ -0,0 +1,30 @@ +puts "\n\n*** rake aj:integration:#{ENV['AJ_ADAPTER']} ***\n" + +ENV["RAILS_ENV"] = "test" +ActiveJob::Base.queue_name_prefix = nil + +require 'rails/generators/rails/app/app_generator' + +dummy_app_path = Dir.mktmpdir + "/dummy" +dummy_app_template = File.expand_path("../dummy_app_template.rb", __FILE__) +args = Rails::Generators::ARGVScrubber.new(["new", dummy_app_path, "--skip-gemfile", "--skip-bundle", + "--skip-git", "--skip-spring", "-d", "sqlite3", "--skip-javascript", "--force", "--quiet", + "--template", dummy_app_template]).prepare! +Rails::Generators::AppGenerator.start args + +require "#{dummy_app_path}/config/environment.rb" + +ActiveRecord::Migrator.migrations_paths = [ Rails.root.join('db/migrate').to_s ] +require 'rails/test_help' + +Rails.backtrace_cleaner.remove_silencers! + +require_relative 'test_case_helpers' +ActiveSupport::TestCase.include(TestCaseHelpers) + +JobsManager.current_manager.start_workers + +Minitest.after_run do + JobsManager.current_manager.stop_workers + JobsManager.current_manager.clear_jobs +end diff --git a/activejob/test/support/integration/jobs_manager.rb b/activejob/test/support/integration/jobs_manager.rb new file mode 100644 index 0000000000..78d48e8d9a --- /dev/null +++ b/activejob/test/support/integration/jobs_manager.rb @@ -0,0 +1,27 @@ +class JobsManager + @@managers = {} + attr :adapter_name + + def self.current_manager + @@managers[ENV['AJ_ADAPTER']] ||= new(ENV['AJ_ADAPTER']) + end + + def initialize(adapter_name) + @adapter_name = adapter_name + require_relative "adapters/#{adapter_name}" + extend "#{adapter_name.camelize}JobsManager".constantize + end + + def setup + ActiveJob::Base.queue_adapter = nil + end + + def clear_jobs + end + + def start_workers + end + + def stop_workers + end +end diff --git a/activejob/test/support/integration/test_case_helpers.rb b/activejob/test/support/integration/test_case_helpers.rb new file mode 100644 index 0000000000..8319d09520 --- /dev/null +++ b/activejob/test/support/integration/test_case_helpers.rb @@ -0,0 +1,56 @@ +require 'active_support/concern' +require 'support/integration/jobs_manager' + +module TestCaseHelpers + extend ActiveSupport::Concern + + included do + self.use_transactional_tests = false + + setup do + clear_jobs + @id = "AJ-#{SecureRandom.uuid}" + end + + teardown do + clear_jobs + end + end + + protected + + def jobs_manager + JobsManager.current_manager + end + + def clear_jobs + jobs_manager.clear_jobs + end + + def adapter_is?(*adapter_class_symbols) + adapter_class_symbols.map(&:to_s).include?(ActiveJob::Base.queue_adapter.class.name.split("::").last.gsub(/Adapter$/, '').underscore) + end + + def wait_for_jobs_to_finish_for(seconds=60) + begin + Timeout.timeout(seconds) do + while !job_executed do + sleep 0.25 + end + end + rescue Timeout::Error + end + end + + def job_executed(id=@id) + Dummy::Application.root.join("tmp/#{id}").exist? + end + + def job_executed_at(id=@id) + File.new(Dummy::Application.root.join("tmp/#{id}")).ctime + end + + def job_output + File.read Dummy::Application.root.join("tmp/#{@id}") + end +end diff --git a/activejob/test/support/job_buffer.rb b/activejob/test/support/job_buffer.rb new file mode 100644 index 0000000000..620cb5288d --- /dev/null +++ b/activejob/test/support/job_buffer.rb @@ -0,0 +1,19 @@ +module JobBuffer + class << self + def clear + values.clear + end + + def add(value) + values << value + end + + def values + @values ||= [] + end + + def last_value + values.last + end + end +end diff --git a/activejob/test/support/que/inline.rb b/activejob/test/support/que/inline.rb new file mode 100644 index 0000000000..0950e52d28 --- /dev/null +++ b/activejob/test/support/que/inline.rb @@ -0,0 +1,14 @@ +require 'que' + +Que::Job.class_eval do + class << self; alias_method :original_enqueue, :enqueue; end + def self.enqueue(*args) + if args.last.is_a?(Hash) + options = args.pop + options.delete(:run_at) + options.delete(:priority) + args << options unless options.empty? + end + self.run(*args) + end +end diff --git a/activejob/test/support/queue_classic/inline.rb b/activejob/test/support/queue_classic/inline.rb new file mode 100644 index 0000000000..5743d5bbb5 --- /dev/null +++ b/activejob/test/support/queue_classic/inline.rb @@ -0,0 +1,23 @@ +require 'queue_classic' + +module QC + class Queue + def enqueue(method, *args) + receiver_str, _, message = method.rpartition('.') + receiver = eval(receiver_str) + receiver.send(message, *args) + end + + def enqueue_in(seconds, method, *args) + receiver_str, _, message = method.rpartition('.') + receiver = eval(receiver_str) + receiver.send(message, *args) + end + + def enqueue_at(not_before, method, *args) + receiver_str, _, message = method.rpartition('.') + receiver = eval(receiver_str) + receiver.send(message, *args) + end + end +end diff --git a/activejob/test/support/sneakers/inline.rb b/activejob/test/support/sneakers/inline.rb new file mode 100644 index 0000000000..16d9b830fa --- /dev/null +++ b/activejob/test/support/sneakers/inline.rb @@ -0,0 +1,12 @@ +require 'sneakers' + +module Sneakers + module Worker + module ClassMethods + def enqueue(msg) + worker = self.new(nil, nil, {}) + worker.work(*msg) + end + end + end +end |