aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--actionmailer/lib/action_mailer/base.rb2
-rw-r--r--actionmailer/lib/action_mailer/queued_message.rb30
-rw-r--r--actionmailer/test/abstract_unit.rb1
-rw-r--r--actionmailer/test/base_test.rb23
-rw-r--r--actionmailer/test/mailers/async_mailer.rb1
-rw-r--r--activerecord/test/cases/associations/has_many_associations_test.rb2
-rw-r--r--activesupport/lib/active_support/queueing.rb81
-rw-r--r--activesupport/test/queueing/synchronous_queue_test.rb27
-rw-r--r--activesupport/test/queueing/test_queue_test.rb4
-rw-r--r--activesupport/test/queueing/threaded_consumer_test.rb64
-rw-r--r--railties/lib/rails/application/finisher.rb2
11 files changed, 139 insertions, 98 deletions
diff --git a/actionmailer/lib/action_mailer/base.rb b/actionmailer/lib/action_mailer/base.rb
index 26787c9b5e..1676c23856 100644
--- a/actionmailer/lib/action_mailer/base.rb
+++ b/actionmailer/lib/action_mailer/base.rb
@@ -4,6 +4,7 @@ require 'action_mailer/collector'
require 'active_support/core_ext/string/inflections'
require 'active_support/core_ext/hash/except'
require 'active_support/core_ext/module/anonymous'
+require 'active_support/queueing'
require 'action_mailer/log_subscriber'
module ActionMailer #:nodoc:
@@ -393,6 +394,7 @@ module ActionMailer #:nodoc:
}.freeze
class_attribute :queue
+ self.queue = ActiveSupport::SynchronousQueue.new
class << self
# Register one or more Observers which will be notified when mail is delivered.
diff --git a/actionmailer/lib/action_mailer/queued_message.rb b/actionmailer/lib/action_mailer/queued_message.rb
index e5868ab43b..8d200617c4 100644
--- a/actionmailer/lib/action_mailer/queued_message.rb
+++ b/actionmailer/lib/action_mailer/queued_message.rb
@@ -5,23 +5,33 @@ module ActionMailer
attr_reader :queue
def initialize(queue, mailer_class, method_name, *args)
- @queue = queue
- @mailer_class = mailer_class
- @method_name = method_name
- @args = args
+ @queue = queue
+ @job = DeliveryJob.new(mailer_class, method_name, args)
end
def __getobj__
- @actual_message ||= @mailer_class.send(:new, @method_name, *@args).message
+ @job.message
end
- def run
- __getobj__.deliver
+ # Queues the message for delivery.
+ def deliver
+ tap { @queue.push @job }
end
- # Will push the message onto the Queue to be processed
- def deliver
- @queue << self
+ class DeliveryJob
+ def initialize(mailer_class, method_name, args)
+ @mailer_class = mailer_class
+ @method_name = method_name
+ @args = args
+ end
+
+ def message
+ @message ||= @mailer_class.send(:new, @method_name, *@args).message
+ end
+
+ def run
+ message.deliver
+ end
end
end
end
diff --git a/actionmailer/test/abstract_unit.rb b/actionmailer/test/abstract_unit.rb
index 0b418c4ea1..4b38d4bd31 100644
--- a/actionmailer/test/abstract_unit.rb
+++ b/actionmailer/test/abstract_unit.rb
@@ -27,7 +27,6 @@ ActionView::Template.register_template_handler :bak, lambda { |template| "Lame b
FIXTURE_LOAD_PATH = File.expand_path('fixtures', File.dirname(__FILE__))
ActionMailer::Base.view_paths = FIXTURE_LOAD_PATH
-ActionMailer::Base.queue = ActiveSupport::SynchronousQueue.new
class MockSMTP
def self.deliveries
diff --git a/actionmailer/test/base_test.rb b/actionmailer/test/base_test.rb
index 6a06cec041..4f2af50fdd 100644
--- a/actionmailer/test/base_test.rb
+++ b/actionmailer/test/base_test.rb
@@ -412,7 +412,7 @@ class BaseTest < ActiveSupport::TestCase
BaseMailer.deliveries.clear
BaseMailer.expects(:deliver_mail).once
mail = BaseMailer.welcome.deliver
- assert_instance_of Mail::Message, mail
+ assert_equal 'The first email on new API!', mail.subject
end
test "calling deliver on the action should increment the deliveries collection if using the test mailer" do
@@ -422,24 +422,15 @@ class BaseTest < ActiveSupport::TestCase
assert_equal(1, BaseMailer.deliveries.length)
end
- def stub_queue(klass, queue)
- Class.new(klass) {
- extend Module.new {
- define_method :queue do
- queue
- end
- }
- }
- end
-
test "delivering message asynchronously" do
- testing_queue = ActiveSupport::TestQueue.new
AsyncMailer.delivery_method = :test
AsyncMailer.deliveries.clear
- stub_queue(AsyncMailer, testing_queue).welcome.deliver
- assert_equal(0, AsyncMailer.deliveries.length)
- testing_queue.drain
- assert_equal(1, AsyncMailer.deliveries.length)
+
+ AsyncMailer.welcome.deliver
+ assert_equal 0, AsyncMailer.deliveries.length
+
+ AsyncMailer.queue.drain
+ assert_equal 1, AsyncMailer.deliveries.length
end
test "calling deliver, ActionMailer should yield back to mail to let it call :do_delivery on itself" do
diff --git a/actionmailer/test/mailers/async_mailer.rb b/actionmailer/test/mailers/async_mailer.rb
index 8a87e2e1cf..c21a464f38 100644
--- a/actionmailer/test/mailers/async_mailer.rb
+++ b/actionmailer/test/mailers/async_mailer.rb
@@ -1,2 +1,3 @@
class AsyncMailer < BaseMailer
+ self.queue = ActiveSupport::TestQueue.new
end
diff --git a/activerecord/test/cases/associations/has_many_associations_test.rb b/activerecord/test/cases/associations/has_many_associations_test.rb
index 5edabb4790..4405f34355 100644
--- a/activerecord/test/cases/associations/has_many_associations_test.rb
+++ b/activerecord/test/cases/associations/has_many_associations_test.rb
@@ -46,7 +46,7 @@ class HasManyAssociationsTestForCountWithCountSql < ActiveRecord::TestCase
end
end
-class HasManyAssociationsTestForCountWithFinderSql < ActiveRecord::TestCase
+class HasManyAssociationsTestForCountWithVariousFinderSqls < ActiveRecord::TestCase
class Invoice < ActiveRecord::Base
ActiveSupport::Deprecation.silence do
has_many :custom_line_items, :class_name => 'LineItem', :finder_sql => "SELECT DISTINCT line_items.amount from line_items"
diff --git a/activesupport/lib/active_support/queueing.rb b/activesupport/lib/active_support/queueing.rb
index f397e1c0c5..d36b5c17a8 100644
--- a/activesupport/lib/active_support/queueing.rb
+++ b/activesupport/lib/active_support/queueing.rb
@@ -2,17 +2,33 @@ require 'delegate'
require 'thread'
module ActiveSupport
- # A Queue that simply inherits from STDLIB's Queue. Everytime this
- # queue is used, Rails automatically sets up a ThreadedConsumer
- # to consume it.
+ # A Queue that simply inherits from STDLIB's Queue. When this
+ # queue is used, Rails automatically starts a job runner in a
+ # background thread.
class Queue < ::Queue
+ attr_writer :consumer
+
+ def initialize(consumer_options = {})
+ super()
+ @consumer_options = consumer_options
+ end
+
+ def consumer
+ @consumer ||= ThreadedQueueConsumer.new(self, @consumer_options)
+ end
+
+ # Drain the queue, running all jobs in a different thread. This method
+ # may not be available on production queues.
+ def drain
+ # run the jobs in a separate thread so assumptions of synchronous
+ # jobs are caught in test mode.
+ consumer.drain
+ end
end
- class SynchronousQueue < ::Queue
+ class SynchronousQueue < Queue
def push(job)
- result = nil
- Thread.new { result = job.run }.join
- result
+ super.tap { drain }
end
alias << push
alias enq push
@@ -25,7 +41,7 @@ module ActiveSupport
#
# Jobs are run in a separate thread to catch mistakes where code
# assumes that the job is run in the same thread.
- class TestQueue < ::Queue
+ class TestQueue < Queue
# Get a list of the jobs off this queue. This method may not be
# available on production queues.
def jobs
@@ -38,14 +54,6 @@ module ActiveSupport
def push(job)
super Marshal.load(Marshal.dump(job))
end
-
- # Drain the queue, running all jobs in a different thread. This method
- # may not be available on production queues.
- def drain
- # run the jobs in a separate thread so assumptions of synchronous
- # jobs are caught in test mode.
- Thread.new { pop.run until empty? }.join
- end
end
# A container for multiple queues. This class delegates to a default Queue
@@ -82,25 +90,17 @@ module ActiveSupport
# queue and joins the thread, which will ensure that all jobs
# are executed before the process finally dies.
class ThreadedQueueConsumer
- def self.start(queue, logger=nil)
- new(queue, logger).start
+ def self.start(*args)
+ new(*args).start
end
- def initialize(queue, logger=nil)
- @queue = queue
- @logger = logger
+ def initialize(queue, options = {})
+ @queue = queue
+ @logger = options[:logger]
end
def start
- @thread = Thread.new do
- while job = @queue.pop
- begin
- job.run
- rescue Exception => e
- handle_exception e
- end
- end
- end
+ @thread = Thread.new { consume }
self
end
@@ -109,8 +109,25 @@ module ActiveSupport
@thread.join
end
- def handle_exception(e)
- @logger.error "Job Error: #{e.message}\n#{e.backtrace.join("\n")}" if @logger
+ def drain
+ Thread.new { run(@queue.pop) until @queue.empty? }.join
+ end
+
+ def consume
+ while job = @queue.pop
+ run job
+ end
+ end
+
+ def run(job)
+ job.run
+ rescue Exception => exception
+ handle_exception job, exception
+ end
+
+ def handle_exception(job, exception)
+ raise unless @logger
+ @logger.error "Job Error: #{exception.message}\n#{exception.backtrace.join("\n")}"
end
end
end
diff --git a/activesupport/test/queueing/synchronous_queue_test.rb b/activesupport/test/queueing/synchronous_queue_test.rb
new file mode 100644
index 0000000000..86c39d0f6c
--- /dev/null
+++ b/activesupport/test/queueing/synchronous_queue_test.rb
@@ -0,0 +1,27 @@
+require 'abstract_unit'
+require 'active_support/queueing'
+
+class SynchronousQueueTest < ActiveSupport::TestCase
+ class Job
+ attr_reader :ran
+ def run; @ran = true end
+ end
+
+ class ExceptionRaisingJob
+ def run; raise end
+ end
+
+ def setup
+ @queue = ActiveSupport::SynchronousQueue.new
+ end
+
+ def test_runs_jobs_immediately
+ job = Job.new
+ @queue.push job
+ assert job.ran
+
+ assert_raises RuntimeError do
+ @queue.push ExceptionRaisingJob.new
+ end
+ end
+end
diff --git a/activesupport/test/queueing/test_queue_test.rb b/activesupport/test/queueing/test_queue_test.rb
index 4c08314366..9e74bc64ee 100644
--- a/activesupport/test/queueing/test_queue_test.rb
+++ b/activesupport/test/queueing/test_queue_test.rb
@@ -12,7 +12,7 @@ class TestQueueTest < ActiveSupport::TestCase
end
end
- def test_drain_raises
+ def test_drain_raises_exceptions_from_running_jobs
@queue.push ExceptionRaisingJob.new
assert_raises(RuntimeError) { @queue.drain }
end
@@ -41,8 +41,8 @@ class TestQueueTest < ActiveSupport::TestCase
end
def test_contents
- assert @queue.empty?
job = EquivalentJob.new
+ assert @queue.empty?
@queue.push job
refute @queue.empty?
assert_equal job, @queue.pop
diff --git a/activesupport/test/queueing/threaded_consumer_test.rb b/activesupport/test/queueing/threaded_consumer_test.rb
index 20a1cc4e8e..fc43cb555a 100644
--- a/activesupport/test/queueing/threaded_consumer_test.rb
+++ b/activesupport/test/queueing/threaded_consumer_test.rb
@@ -5,7 +5,7 @@ require "active_support/log_subscriber/test_helper"
class TestThreadConsumer < ActiveSupport::TestCase
class Job
attr_reader :id
- def initialize(id, &block)
+ def initialize(id = 1, &block)
@id = id
@block = block
end
@@ -16,83 +16,77 @@ class TestThreadConsumer < ActiveSupport::TestCase
end
def setup
- @queue = ActiveSupport::Queue.new
@logger = ActiveSupport::LogSubscriber::TestHelper::MockLogger.new
- @consumer = ActiveSupport::ThreadedQueueConsumer.start(@queue, @logger)
+ @queue = ActiveSupport::Queue.new(logger: @logger)
end
def teardown
- @queue.push nil
+ @queue.drain
end
test "the jobs are executed" do
ran = false
-
- job = Job.new(1) do
- ran = true
- end
+ job = Job.new { ran = true }
@queue.push job
- sleep 0.1
+ @queue.drain
+
assert_equal true, ran
end
test "the jobs are not executed synchronously" do
- ran = false
-
- job = Job.new(1) do
- sleep 0.1
- ran = true
- end
+ run, ran = Queue.new, Queue.new
+ job = Job.new { ran.push run.pop }
+ @queue.consumer.start
@queue.push job
- assert_equal false, ran
+ assert ran.empty?
+
+ run.push true
+ assert_equal true, ran.pop
end
test "shutting down the queue synchronously drains the jobs" do
ran = false
-
- job = Job.new(1) do
+ job = Job.new do
sleep 0.1
ran = true
end
+ @queue.consumer.start
@queue.push job
assert_equal false, ran
- @consumer.shutdown
-
+ @queue.consumer.shutdown
assert_equal true, ran
end
test "log job that raises an exception" do
- job = Job.new(1) do
- raise "RuntimeError: Error!"
- end
+ job = Job.new { raise "RuntimeError: Error!" }
@queue.push job
- sleep 0.1
+ @queue.drain
assert_equal 1, @logger.logged(:error).size
- assert_match(/Job Error: RuntimeError: Error!/, @logger.logged(:error).last)
+ assert_match 'Job Error: RuntimeError: Error!', @logger.logged(:error).last
end
test "test overriding exception handling" do
- @consumer.shutdown
- @consumer = Class.new(ActiveSupport::ThreadedQueueConsumer) do
- attr_reader :last_error
- def handle_exception(e)
- @last_error = e.message
+ @queue.consumer.instance_eval do
+ def handle_exception(job, exception)
+ @last_error = exception.message
end
- end.start(@queue)
- job = Job.new(1) do
- raise "RuntimeError: Error!"
+ def last_error
+ @last_error
+ end
end
+ job = Job.new { raise "RuntimeError: Error!" }
+
@queue.push job
- sleep 0.1
+ @queue.drain
- assert_equal "RuntimeError: Error!", @consumer.last_error
+ assert_equal "RuntimeError: Error!", @queue.consumer.last_error
end
end
diff --git a/railties/lib/rails/application/finisher.rb b/railties/lib/rails/application/finisher.rb
index f9a3c00946..d2a402aa51 100644
--- a/railties/lib/rails/application/finisher.rb
+++ b/railties/lib/rails/application/finisher.rb
@@ -98,7 +98,7 @@ module Rails
initializer :activate_queue_consumer do |app|
if config.queue == ActiveSupport::Queue
- app.queue_consumer = config.queue_consumer.start(app.queue, Rails.logger)
+ app.queue_consumer = config.queue_consumer.start(app.queue, {logger: Rails.logger})
at_exit { app.queue_consumer.shutdown }
end
end
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-12 17:52:55 +0000