diff options
Diffstat (limited to 'activejob')
70 files changed, 2551 insertions, 0 deletions
diff --git a/activejob/CHANGELOG.md b/activejob/CHANGELOG.md new file mode 100644 index 0000000000..b04883413e --- /dev/null +++ b/activejob/CHANGELOG.md @@ -0,0 +1 @@ +* Started project.
\ No newline at end of file diff --git a/activejob/MIT-LICENSE b/activejob/MIT-LICENSE new file mode 100644 index 0000000000..8b1e97b776 --- /dev/null +++ b/activejob/MIT-LICENSE @@ -0,0 +1,21 @@ +Copyright (c) 2014 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..5734ca413e --- /dev/null +++ b/activejob/README.md @@ -0,0 +1,141 @@ +# 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 ActionMailer'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 +# Adapters currently supported: :backburner, :delayed_job, :qu, :que, :queue_classic, +# :resque, :sidekiq, :sneakers, :sucker_punch +``` + +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 + +We currently have adapters for: + +* [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) +* [QueueClassic](https://github.com/ryandotsmith/queue_classic) +* [Resque 1.x](https://github.com/resque/resque) +* [Sidekiq](https://github.com/mperham/sidekiq) +* [Sneakers](https://github.com/jondot/sneakers) +* [Sucker Punch](https://github.com/brandonhilkert/sucker_punch) + + +## Auxiliary gems + +* [activejob-stats](https://github.com/seuros/activejob-stats) + +## Download and installation + +The latest version of Active Job can be installed with RubyGems: + +``` + % [sudo] 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 + +ActiveJob 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..484cd1d0b8 --- /dev/null +++ b/activejob/Rakefile @@ -0,0 +1,85 @@ +require 'rake/testtask' +require 'rubygems/package_task' + +dir = File.dirname(__FILE__) + +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 + +task default: :test + +ACTIVEJOB_ADAPTERS = %w(inline delayed_job qu que queue_classic resque sidekiq sneakers sucker_punch backburner) +ACTIVEJOB_ADAPTERS -= %w(queue_classic) if defined?(JRUBY_VERSION) + +desc 'Run all adapter tests' +task :test do + tasks = ACTIVEJOB_ADAPTERS.map{|a| "test_#{a}" } + run_without_aborting(*tasks) +end + +namespace :test do + desc 'Run all adapter tests in isolation' + task :isolated do + tasks = ACTIVEJOB_ADAPTERS.map{|a| "isolated_test_#{a}" } + run_without_aborting(*tasks) + end +end + + +ACTIVEJOB_ADAPTERS.each do |adapter| + namespace :test do + Rake::TestTask.new(adapter => "#{adapter}:env") do |t| + t.description = "" + t.libs << 'test' + t.test_files = FileList['test/cases/**/*_test.rb'] + t.verbose = true + end + + namespace :isolated do + task adapter => "#{adapter}:env" do + 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 + end + + namespace adapter do + task test: "test_#{adapter}" + task isolated_test: "isolated_test_#{adapter}" + + task(:env) { ENV['AJADAPTER'] = adapter } + end + + + desc "Run #{adapter} tests" + task "test_#{adapter}" => ["#{adapter}:env", "test:#{adapter}"] + + desc "Run #{adapter} tests in isolation" + task "isolated_test_#{adapter}" => ["#{adapter}:env", "test:isolated:#{adapter}"] +end + + +spec = eval(File.read('activejob.gemspec')) + +Gem::PackageTask.new(spec) do |p| + p.gem_spec = spec +end + +desc 'Release to rubygems' +task release: :package do + require 'rake/gemcutter' + Rake::Gemcutter::Tasks.new(spec).define + Rake::Task['gem:push'].invoke +end diff --git a/activejob/activejob.gemspec b/activejob/activejob.gemspec new file mode 100644 index 0000000000..9944b2a1bd --- /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 = '>= 1.9.3' + + 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.2.3' +end diff --git a/activejob/lib/active_job.rb b/activejob/lib/active_job.rb new file mode 100644 index 0000000000..1b582f5877 --- /dev/null +++ b/activejob/lib/active_job.rb @@ -0,0 +1,37 @@ +#-- +# Copyright (c) 2014 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 :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..69a4ad928d --- /dev/null +++ b/activejob/lib/active_job/arguments.rb @@ -0,0 +1,75 @@ +module ActiveJob + # Raised when an exception is raised during job arguments deserialization. + # + # Wraps the original exception raised as +original_exception+. + class DeserializationError < StandardError + 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 being set as job argument. We + # currently support NilClass, Fixnum, Float, String, TrueClass, FalseClass, + # Bignum and object that can be represented as GlobalIDs (ex: Active Record). + # Also raised if you set the key for a Hash something else than a string or + # a symbol. + class SerializationError < ArgumentError + end + + module Arguments + extend self + TYPE_WHITELIST = [ NilClass, Fixnum, Float, String, TrueClass, FalseClass, Bignum ] + + def serialize(arguments) + arguments.map { |argument| serialize_argument(argument) } + end + + def deserialize(arguments) + arguments.map { |argument| deserialize_argument(argument) } + rescue => e + raise DeserializationError.new(e) + end + + private + def serialize_argument(argument) + case argument + when GlobalID::Identification + argument.global_id.to_s + when *TYPE_WHITELIST + argument + when Array + argument.map { |arg| serialize_argument(arg) } + when Hash + Hash[ argument.map { |key, value| [ serialize_hash_key(key), serialize_argument(value) ] } ] + else + raise SerializationError.new("Unsupported argument type: #{argument.class.name}") + end + end + + def deserialize_argument(argument) + case argument + when Array + argument.map { |arg| deserialize_argument(arg) } + when Hash + Hash[ argument.map { |key, value| [ key, deserialize_argument(value) ] } ].with_indifferent_access + when String, GlobalID + GlobalID::Locator.locate(argument) || argument + else + argument + end + end + + def serialize_hash_key(key) + case key + when String, Symbol + key.to_s + else + raise SerializationError.new("Unsupported hash key type: #{key.class.name}") + 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..a3bec1f827 --- /dev/null +++ b/activejob/lib/active_job/base.rb @@ -0,0 +1,21 @@ +require 'active_job/core' +require 'active_job/queue_adapter' +require 'active_job/queue_name' +require 'active_job/enqueuing' +require 'active_job/execution' +require 'active_job/callbacks' +require 'active_job/logging' + +module ActiveJob + class Base + include Core + include QueueAdapter + include QueueName + include Enqueuing + include Execution + include Callbacks + include Logging + + 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..cafa3438c0 --- /dev/null +++ b/activejob/lib/active_job/callbacks.rb @@ -0,0 +1,144 @@ +require 'active_support/callbacks' + +module ActiveJob + # = Active Job Callbacks + # + # Active Job provides hooks during the lifecycle of a job. Callbacks allow you + # to trigger logic during the lifecycle 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 + + 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..b6dd03a0bc --- /dev/null +++ b/activejob/lib/active_job/core.rb @@ -0,0 +1,89 @@ +module ActiveJob + 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 on which the job should be run on. + attr_writer :queue_name + end + + 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.job_id = job_data['job_id'] + job.queue_name = job_data['queue_name'] + job.serialized_arguments = job_data['arguments'] + 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 + # + # ==== Examples + # + # VideoJob.set(queue: :some_queue).perform_later(Video.last) + # VideoJob.set(wait: 5.minutes).perform_later(Video.last) + # VideoJob.set(wait_until: Time.tomorroe).perform_later(Video.last) + # VideoJob.set(queue: :some_queue, wait: 5.minutes).perform_later(Video.last) + # VideoJob.set(queue: :some_queue, wait_until: Time.tomorroe).perform_later(Video.last) + def set(options={}) + ConfiguredJob.new(self, options) + end + end + + # Creates a new job instance. Takes as arguments 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 + 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, + 'arguments' => serialize_arguments(arguments) + } + 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..e8bc44cbc4 --- /dev/null +++ b/activejob/lib/active_job/enqueuing.rb @@ -0,0 +1,75 @@ +require 'active_job/arguments' + +module ActiveJob + module Enqueuing + extend ActiveSupport::Concern + + 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 args 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 + + # Reschedule the job to be re-executed. This is usefull 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 + # + # ==== 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 + + # Equeue 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 + # + # ==== 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 + 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] + 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..d6d67c46e3 --- /dev/null +++ b/activejob/lib/active_job/execution.rb @@ -0,0 +1,41 @@ +require 'active_support/rescuable' +require 'active_job/arguments' + +module ActiveJob + module Execution + extend ActiveSupport::Concern + include ActiveSupport::Rescuable + + 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 + # and will block the execution 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..1f25d48326 --- /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 = 4 + MINOR = 2 + TINY = 0 + PRE = "beta1" + + 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..962005cd15 --- /dev/null +++ b/activejob/lib/active_job/logging.rb @@ -0,0 +1,102 @@ +require 'active_support/core_ext/string/filters' +require 'active_support/tagged_logging' +require 'active_support/logger' + +module ActiveJob + module Logging + 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 + + before_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 + 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).to_s}ms" + end + end + + private + def queue_name(event) + event.payload[:adapter].name.demodulize.remove('Adapter') + "(#{event.payload[:job].queue_name})" + end + + def args_info(job) + job.arguments.any? ? " with arguments: #{job.arguments.map(&:inspect).join(", ")}" : "" + 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..fb54aec75e --- /dev/null +++ b/activejob/lib/active_job/queue_adapter.rb @@ -0,0 +1,29 @@ +require 'active_job/queue_adapters/inline_adapter' +require 'active_support/core_ext/string/inflections' + +module ActiveJob + module QueueAdapter + extend ActiveSupport::Concern + + module ClassMethods + mattr_reader(:queue_adapter) { ActiveJob::QueueAdapters::InlineAdapter } + + def queue_adapter=(name_or_adapter) + @@queue_adapter = \ + case name_or_adapter + when :test + ActiveJob::QueueAdapters::TestAdapter.new + when Symbol, String + load_adapter(name_or_adapter) + when Class + name_or_adapter + end + end + + private + def load_adapter(name) + "ActiveJob::QueueAdapters::#{name.to_s.camelize}Adapter".constantize + end + end + end +end
\ No newline at end of file diff --git a/activejob/lib/active_job/queue_adapters.rb b/activejob/lib/active_job/queue_adapters.rb new file mode 100644 index 0000000000..345b01ef00 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters.rb @@ -0,0 +1,17 @@ +module ActiveJob + module QueueAdapters + extend ActiveSupport::Autoload + + autoload :InlineAdapter + autoload :BackburnerAdapter + autoload :DelayedJobAdapter + autoload :QuAdapter + autoload :QueAdapter + autoload :QueueClassicAdapter + autoload :ResqueAdapter + autoload :SidekiqAdapter + autoload :SneakersAdapter + autoload :SuckerPunchAdapter + autoload :TestAdapter + 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..e1b00f1de7 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/backburner_adapter.rb @@ -0,0 +1,26 @@ +require 'backburner' + +module ActiveJob + module QueueAdapters + class BackburnerAdapter + class << self + def enqueue(job) + Backburner::Worker.enqueue JobWrapper, [ job.serialize ], queue: job.queue_name + end + + def enqueue_at(job, timestamp) + delay = timestamp - Time.current.to_f + Backburner::Worker.enqueue JobWrapper, [ job.serialize ], queue: job.queue_name, delay: delay + end + end + + class JobWrapper + 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..658799edfc --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb @@ -0,0 +1,23 @@ +require 'delayed_job' + +module ActiveJob + module QueueAdapters + class DelayedJobAdapter + class << self + def enqueue(job) + JobWrapper.new.delay(queue: job.queue_name).perform(job.serialize) + end + + def enqueue_at(job, timestamp) + JobWrapper.new.delay(queue: job.queue_name, run_at: Time.at(timestamp)).perform(job.serialize) + end + end + + class JobWrapper + def perform(job_data) + 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..fdefa38d5e --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/inline_adapter.rb @@ -0,0 +1,15 @@ +module ActiveJob + module QueueAdapters + class InlineAdapter + class << self + def enqueue(job) + Base.execute(job.serialize) + end + + def enqueue_at(*) + raise NotImplementedError.new("Use a queueing backend to enqueue jobs in the future. Read more at https://github.com/rails/activejob") + end + 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..f681fd7e8a --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/qu_adapter.rb @@ -0,0 +1,29 @@ +require 'qu' + +module ActiveJob + module QueueAdapters + class QuAdapter + class << self + def enqueue(job, *args) + Qu::Payload.new(klass: JobWrapper, args: [job.serialize]).tap do |payload| + payload.instance_variable_set(:@queue, job.queue_name) + end.push + end + + def enqueue_at(job, timestamp, *args) + raise NotImplementedError + end + end + + class JobWrapper < Qu::Job + 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..51891ab07b --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/que_adapter.rb @@ -0,0 +1,23 @@ +require 'que' + +module ActiveJob + module QueueAdapters + class QueAdapter + class << self + def enqueue(job) + JobWrapper.enqueue job.serialize, queue: job.queue_name + end + + def enqueue_at(job, timestamp) + JobWrapper.enqueue job.serialize, queue: job.queue_name, run_at: Time.at(timestamp) + end + end + + class JobWrapper < Que::Job + 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..ddcc868317 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/queue_classic_adapter.rb @@ -0,0 +1,40 @@ +require 'queue_classic' + +module ActiveJob + module QueueAdapters + class QueueClassicAdapter + class << self + def enqueue(job) + build_queue(job.queue_name).enqueue("#{JobWrapper.name}.perform", job.serialize) + end + + def enqueue_at(job, timestamp) + queue = build_queue(job.queue_name) + unless queue.respond_to?(:enqueue_at) + raise NotImplementedError, 'To be able to schedule jobs with Queue Classic ' \ + 'the QC::Queue needs to respond to `enqueue_at(timestamp, method, *args)`. ' + 'You can implement this yourself or you can use the queue_classic-later gem.' + end + queue.enqueue_at(timestamp, "#{JobWrapper.name}.perform", job.serialize) + end + + # Builds a <tt>QC::Queue</tt> object to schedule jobs on. + # + # If you have a custom <tt>QC::Queue</tt> subclass you'll need to suclass + # <tt>ActiveJob::QueueAdapters::QueueClassicAdapter</tt> and override the + # <tt>build_queue</tt> method. + def build_queue(queue_name) + QC::Queue.new(queue_name) + end + end + + class JobWrapper + 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..affa3bdfbc --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/resque_adapter.rb @@ -0,0 +1,41 @@ +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 + class ResqueAdapter + class << self + def enqueue(job) + Resque.enqueue_to job.queue_name, JobWrapper, job.serialize + end + + def enqueue_at(job, timestamp) + 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 + end + + class JobWrapper + 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..79926a1e24 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/sidekiq_adapter.rb @@ -0,0 +1,35 @@ +require 'sidekiq' + +module ActiveJob + module QueueAdapters + class SidekiqAdapter + class << self + def enqueue(job) + #Sidekiq::Client does not support symbols as keys + Sidekiq::Client.push \ + 'class' => JobWrapper, + 'queue' => job.queue_name, + 'args' => [ job.serialize ], + 'retry' => true + end + + def enqueue_at(job, timestamp) + Sidekiq::Client.push \ + 'class' => JobWrapper, + 'queue' => job.queue_name, + 'args' => [ job.serialize ], + 'retry' => true, + 'at' => timestamp + end + end + + class JobWrapper + 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..60633639f4 --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/sneakers_adapter.rb @@ -0,0 +1,34 @@ +require 'sneakers' +require 'thread' + +module ActiveJob + module QueueAdapters + class SneakersAdapter + @monitor = Monitor.new + + class << self + def enqueue(job) + @monitor.synchronize do + JobWrapper.from_queue job.queue_name + JobWrapper.enqueue ActiveSupport::JSON.encode(job.serialize) + end + end + + def enqueue_at(job, timestamp) + raise NotImplementedError + end + end + + class JobWrapper + include Sneakers::Worker + from_queue 'active_jobs_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..b19a38093f --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/sucker_punch_adapter.rb @@ -0,0 +1,25 @@ +require 'sucker_punch' + +module ActiveJob + module QueueAdapters + class SuckerPunchAdapter + class << self + def enqueue(job) + JobWrapper.new.async.perform job.serialize + end + + def enqueue_at(job, timestamp) + raise NotImplementedError + end + end + + class JobWrapper + 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..b9997efddf --- /dev/null +++ b/activejob/lib/active_job/queue_adapters/test_adapter.rb @@ -0,0 +1,76 @@ +module ActiveJob + module QueueAdapters + class TestAdapter + attr_accessor(:perform_enqueued_jobs) { false } + attr_accessor(:perform_enqueued_at_jobs) { false } + delegate :name, to: :class + + # Provides a store of all the enqueued jobs with the TestAdapter so you can check them. + def enqueued_jobs + @enqueued_jobs ||= [] + end + + # Allows you to overwrite the default enqueued jobs store from an array to some + # other object. If you just want to clear the store, + # call ActiveJob::QueueAdapters::TestAdapter.enqueued_jobs.clear. + # + # If you place another object here, please make sure it responds to: + # + # * << (message) + # * clear + # * length + # * size + # * and other common Array methods + def enqueued_jobs=(val) + @enqueued_jobs = val + end + + # Provides a store of all the performed jobs with the TestAdapter so you can check them. + def performed_jobs + @performed_jobs ||= [] + end + + # Allows you to overwrite the default performed jobs store from an array to some + # other object. If you just want to clear the store, + # call ActiveJob::QueueAdapters::TestAdapter.performed_jobs.clear. + # + # If you place another object here, please make sure it responds to: + # + # * << (message) + # * clear + # * length + # * size + # * and other common Array methods + def performed_jobs=(val) + @performed_jobs = val + end + + def enqueue(job) + if perform_enqueued_jobs? + performed_jobs << {job: job.class, args: job.arguments, queue: job.queue_name} + job.perform_now + else + enqueued_jobs << {job: job.class, args: job.arguments, queue: job.queue_name} + end + end + + def enqueue_at(job, timestamp) + if perform_enqueued_at_jobs? + performed_jobs << {job: job.class, args: job.arguments, queue: job.queue_name, at: timestamp} + job.perform_now + else + enqueued_jobs << {job: job.class, args: job.arguments, queue: job.queue_name, at: timestamp} + end + end + + private + def perform_enqueued_jobs? + perform_enqueued_jobs + end + + def perform_enqueued_at_jobs? + perform_enqueued_at_jobs + 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..45acb71605 --- /dev/null +++ b/activejob/lib/active_job/queue_name.rb @@ -0,0 +1,38 @@ +module ActiveJob + module QueueName + extend ActiveSupport::Concern + + module ClassMethods + mattr_accessor(:queue_name_prefix) + mattr_accessor(:default_queue_name) { "default" } + + 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.to_s.presence || default_queue_name + name_parts = [queue_name_prefix.presence, queue_name] + name_parts.compact.join('_') + end + end + + included do + class_attribute :queue_name, instance_accessor: false + self.queue_name = default_queue_name + 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/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..fa0576669e --- /dev/null +++ b/activejob/lib/active_job/test_helper.rb @@ -0,0 +1,196 @@ +module ActiveJob + # Provides helper methods for testing Active Job + module TestHelper + extend ActiveSupport::Concern + + included do + def before_setup + @old_queue_adapter = queue_adapter + ActiveJob::Base.queue_adapter = :test + clear_enqueued_jobs + clear_performed_jobs + super + end + + def after_teardown + super + ActiveJob::Base.queue_adapter = @old_queue_adapter + 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 + def assert_enqueued_jobs(number) + if block_given? + original_count = enqueued_jobs.size + yield + new_count = enqueued_jobs.size + assert_equal original_count + number, new_count, + "#{number} jobs expected, but #{new_count - original_count} were enqueued" + else + enqueued_jobs_size = enqueued_jobs.size + assert_equal number, enqueued_jobs_size, "#{number} jobs expected, but #{enqueued_jobs_size} were enqueued" + end + end + + # Assert that no job 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 + # + # Note: This assertion is simply a shortcut for: + # + # assert_enqueued_jobs 0 + def assert_no_enqueued_jobs(&block) + assert_enqueued_jobs 0, &block + end + + # Asserts that the number of performed jobs matches the given number. + # + # def test_jobs + # assert_performed_jobs 0 + # HelloJob.perform_later('xavier') + # assert_performed_jobs 1 + # HelloJob.perform_later('yves') + # assert_performed_jobs 2 + # 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 + def assert_performed_jobs(number) + if block_given? + original_count = performed_jobs.size + yield + new_count = performed_jobs.size + assert_equal original_count + number, new_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 + # HelloJob.perform_later('matthew') + # assert_performed_jobs 1 + # 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 + # + # Note: This assertion is simply a shortcut for: + # + # assert_performed_jobs 0 + def assert_no_performed_jobs(&block) + assert_performed_jobs 0, &block + end + + # Asserts that the job passed in the block has been enqueued with the given arguments. + # + # def assert_enqueued_job + # assert_enqueued_with(job: MyJob, args: [1,2,3], queue: 'low') do + # MyJob.perform_later(1,2,3) + # end + # end + def assert_enqueued_with(args = {}, &_block) + original_enqueued_jobs = enqueued_jobs.dup + clear_enqueued_jobs + args.assert_valid_keys(:job, :args, :at, :queue) + yield + matching_job = enqueued_jobs.any? do |job| + args.all? { |key, value| value == job[key] } + end + assert matching_job + ensure + queue_adapter.enqueued_jobs = original_enqueued_jobs + enqueued_jobs + 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 + # end + def assert_performed_with(args = {}, &_block) + original_performed_jobs = performed_jobs.dup + clear_performed_jobs + args.assert_valid_keys(:job, :args, :at, :queue) + yield + matching_job = performed_jobs.any? do |job| + args.all? { |key, value| value == job[key] } + end + assert matching_job, "No performed job found with #{args}" + ensure + queue_adapter.performed_jobs = original_performed_jobs + performed_jobs + 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 + enqueued_jobs.clear + end + + def clear_performed_jobs + performed_jobs.clear + 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..979ffcb748 --- /dev/null +++ b/activejob/lib/rails/generators/job/job_generator.rb @@ -0,0 +1,24 @@ +require 'rails/generators/named_base' + +module Rails + 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..462c71d917 --- /dev/null +++ b/activejob/lib/rails/generators/job/templates/job.rb @@ -0,0 +1,9 @@ +<% module_namespacing do -%> +class <%= class_name %>Job < ActiveJob::Base + queue_as :<%= options[:queue] %> + + def perform(*args) + # Do something later + end +end +<% end -%> 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/cases/adapter_test.rb b/activejob/test/cases/adapter_test.rb new file mode 100644 index 0000000000..4fc235ae40 --- /dev/null +++ b/activejob/test/cases/adapter_test.rb @@ -0,0 +1,8 @@ +require 'helper' + +class AdapterTest < ActiveSupport::TestCase + test "should load #{ENV['AJADAPTER']} adapter" do + ActiveJob::Base.queue_adapter = ENV['AJADAPTER'].to_sym + assert_equal ActiveJob::Base.queue_adapter, "active_job/queue_adapters/#{ENV['AJADAPTER']}_adapter".classify.constantize + 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..db22783030 --- /dev/null +++ b/activejob/test/cases/job_serialization_test.rb @@ -0,0 +1,15 @@ +require 'helper' +require 'jobs/gid_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 +end diff --git a/activejob/test/cases/logging_test.rb b/activejob/test/cases/logging_test.rb new file mode 100644 index 0000000000..9c56ee08b6 --- /dev/null +++ b/activejob/test/cases/logging_test.rb @@ -0,0 +1,105 @@ +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' + +class AdapterTest < 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 + ActiveJob::Base.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_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/parameters_test.rb b/activejob/test/cases/parameters_test.rb new file mode 100644 index 0000000000..78853c51e1 --- /dev/null +++ b/activejob/test/cases/parameters_test.rb @@ -0,0 +1,77 @@ +require 'helper' +require 'active_job/arguments' +require 'models/person' +require 'active_support/core_ext/hash/indifferent_access' + +class ParameterSerializationTest < ActiveSupport::TestCase + test 'should make no change to regular values' do + assert_equal [ 1, "something" ], ActiveJob::Arguments.serialize([ 1, "something" ]) + end + + test 'should not allow complex objects' do + assert_equal [ nil ], ActiveJob::Arguments.serialize([ nil ]) + assert_equal [ 1 ], ActiveJob::Arguments.serialize([ 1 ]) + assert_equal [ 1.0 ], ActiveJob::Arguments.serialize([ 1.0 ]) + assert_equal [ 'a' ], ActiveJob::Arguments.serialize([ 'a' ]) + assert_equal [ true ], ActiveJob::Arguments.serialize([ true ]) + assert_equal [ false ], ActiveJob::Arguments.serialize([ false ]) + assert_equal [ { "a" => 1, "b" => 2 } ], ActiveJob::Arguments.serialize([ { a: 1, "b" => 2 } ]) + assert_equal [ [ 1 ] ], ActiveJob::Arguments.serialize([ [ 1 ] ]) + assert_equal [ 1_000_000_000_000_000_000_000 ], ActiveJob::Arguments.serialize([ 1_000_000_000_000_000_000_000 ]) + + err = assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize([ 1, self ]) + end + assert_equal "Unsupported argument type: #{self.class.name}", err.message + end + + test 'should dive deep into arrays or hashes' do + assert_equal [ { "a" => Person.find(5).gid.to_s }.with_indifferent_access ], ActiveJob::Arguments.serialize([ { a: Person.find(5) } ]) + assert_equal [ [ Person.find(5).gid.to_s ] ], ActiveJob::Arguments.serialize([ [ Person.find(5) ] ]) + end + + test 'should dive deep into arrays or hashes and raise exception on complex objects' do + err = assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize([ 1, [self] ]) + end + assert_equal "Unsupported argument type: #{self.class.name}", err.message + end + + test 'shoud dive deep into hashes and allow raise exception on not string/symbol keys' do + err = assert_raises ActiveJob::SerializationError do + ActiveJob::Arguments.serialize([ [ { 1 => 2 } ] ]) + end + assert_equal "Unsupported hash key type: Fixnum", err.message + end + + test 'should serialize records with global id' do + assert_equal [ Person.find(5).gid.to_s ], ActiveJob::Arguments.serialize([ Person.find(5) ]) + end + + test 'should serialize values and records together' do + assert_equal [ 3, Person.find(5).gid.to_s ], ActiveJob::Arguments.serialize([ 3, Person.find(5) ]) + end +end + +class ParameterDeserializationTest < ActiveSupport::TestCase + test 'should make no change to regular values' do + assert_equal [ 1, "something" ], ActiveJob::Arguments.deserialize([ 1, "something" ]) + end + + test 'should deserialize records with global id' do + assert_equal [ Person.find(5) ], ActiveJob::Arguments.deserialize([ Person.find(5).gid ]) + end + + test 'should serialize values and records together' do + assert_equal [ 3, Person.find(5) ], ActiveJob::Arguments.deserialize([ 3, Person.find(5).gid ]) + end + + test 'should dive deep when deserialising arrays' do + assert_equal [ [ 3, Person.find(5) ] ], ActiveJob::Arguments.deserialize([ [ 3, Person.find(5).gid ] ]) + end + + test 'should dive deep when deserialising hashes' do + assert_equal [ { "5" => Person.find(5) } ], ActiveJob::Arguments.deserialize([ { "5" => Person.find(5).gid } ]) + 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..4052477543 --- /dev/null +++ b/activejob/test/cases/queue_naming_test.rb @@ -0,0 +1,61 @@ +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 + begin + original_queue_name = HelloJob.queue_name + HelloJob.queue_as :greetings + assert_equal "greetings", HelloJob.new.queue_name + ensure + HelloJob.queue_name = original_queue_name + end + end + + test 'evals block given to queue_as to determine queue' do + begin + original_queue_name = HelloJob.queue_name + 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 + begin + original_queue_name = HelloJob.queue_name + 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 'queu_name_prefix prepended to the queue name' do + begin + original_queue_name_prefix = ActiveJob::Base.queue_name_prefix + original_queue_name = HelloJob.queue_name + + 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 '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/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..1b6c2e9fac --- /dev/null +++ b/activejob/test/cases/rescue_test.rb @@ -0,0 +1,36 @@ +require 'helper' +require 'jobs/rescue_job' +require 'models/person' + +require 'active_support/core_ext/object/inclusion' + +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..1d0fdbd22d --- /dev/null +++ b/activejob/test/cases/test_case_test.rb @@ -0,0 +1,14 @@ +require 'helper' +require 'jobs/hello_job' +require 'jobs/logging_job' +require 'jobs/nested_job' + +class ActiveJobTestCaseTest < ActiveJob::TestCase + def test_include_helper + assert_includes self.class.ancestors, ActiveJob::TestHelper + end + + def test_set_test_adapter + assert_instance_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..eab540bb6c --- /dev/null +++ b/activejob/test/cases/test_helper_test.rb @@ -0,0 +1,214 @@ +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' + +class EnqueuedJobsTest < ActiveJob::TestCase + setup { queue_adapter.perform_enqueued_at_jobs = true } + + 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_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 + assert_nothing_raised do + assert_no_enqueued_jobs do + # Scheduled jobs are being performed in this context + HelloJob.set(wait_until: Date.tomorrow.noon).perform_later('godfrey') + 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_job + assert_enqueued_with(job: LoggingJob, queue: 'default') do + NestedJob.set(wait_until: Date.tomorrow.noon).perform_later + end + 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 + + assert_raise ActiveSupport::TestCase::Assertion do + assert_enqueued_with(job: NestedJob, queue: 'low') do + NestedJob.perform_later + end + end + 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 +end + +class PerformedJobsTest < ActiveJob::TestCase + setup { queue_adapter.perform_enqueued_jobs = true } + + 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_with_no_block + assert_nothing_raised do + HelloJob.perform_later('rafael') + assert_performed_jobs 1 + end + + assert_nothing_raised do + HelloJob.perform_later('aaron') + HelloJob.perform_later('matthew') + assert_performed_jobs 3 + end + end + + def test_assert_no_performed_jobs + assert_nothing_raised do + assert_no_performed_jobs do + # Scheduled jobs are being enqueued in this context + HelloJob.set(wait_until: Date.tomorrow.noon).perform_later('godfrey') + 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_job + assert_performed_with(job: NestedJob, queue: 'default') do + NestedJob.perform_later + end + end + + def test_assert_performed_job_failure + assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_with(job: LoggingJob, at: Date.tomorrow.noon, queue: 'default') do + NestedJob.set(wait_until: Date.tomorrow.noon).perform_later + end + end + + assert_raise ActiveSupport::TestCase::Assertion do + assert_performed_with(job: NestedJob, at: Date.tomorrow.noon, queue: 'low') do + NestedJob.set(queue: 'low', wait_until: Date.tomorrow.noon).perform_later + end + end + end +end diff --git a/activejob/test/helper.rb b/activejob/test/helper.rb new file mode 100644 index 0000000000..85094387ef --- /dev/null +++ b/activejob/test/helper.rb @@ -0,0 +1,24 @@ +require File.expand_path('../../../load_paths', __FILE__) + +require 'active_job' + +GlobalID.app = 'aj' + +@adapter = ENV['AJADAPTER'] || 'inline' + +def sidekiq? + @adapter == 'sidekiq' +end + +def ruby_193? + RUBY_VERSION == '1.9.3' && RUBY_ENGINE != 'java' +end + +# Sidekiq doesn't work with MRI 1.9.3 +exit if sidekiq? && ruby_193? + +require "adapters/#{@adapter}" + +require 'active_support/testing/autorun' + +ActiveJob::Base.logger.level = Logger::DEBUG diff --git a/activejob/test/jobs/callback_job.rb b/activejob/test/jobs/callback_job.rb new file mode 100644 index 0000000000..056dd073e8 --- /dev/null +++ b/activejob/test/jobs/callback_job.rb @@ -0,0 +1,32 @@ +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 :around_perform + around_enqueue :around_enqueue + + + def perform(person = "david") + # NOTHING! + end + + def history + @history ||= [] + end + + # FIXME: Not sure why these can't be declared inline like before/after + def around_perform + history << "CallbackJob ran around_perform_start" + yield + history << "CallbackJob ran around_perform_stop" + end + + def around_enqueue + history << "CallbackJob ran around_enqueue_start" + yield + history << "CallbackJob ran around_enqueue_stop" + 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/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/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/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..b50ed36fc2 --- /dev/null +++ b/activejob/test/support/delayed_job/delayed/backend/test.rb @@ -0,0 +1,113 @@ +#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 do |o| + o.save + end + 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/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..2e210acb6b --- /dev/null +++ b/activejob/test/support/que/inline.rb @@ -0,0 +1,9 @@ +require 'que' + +Que::Job.class_eval do + class << self; alias_method :original_enqueue, :enqueue; end + def self.enqueue(*args) + args.pop if args.last.is_a?(Hash) + 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 |