diff options
Diffstat (limited to 'railties/lib/rails')
6 files changed, 9 insertions, 122 deletions
diff --git a/railties/lib/rails/application.rb b/railties/lib/rails/application.rb index 80b8e4b9ba..aa5b986862 100644 --- a/railties/lib/rails/application.rb +++ b/railties/lib/rails/application.rb @@ -1,4 +1,5 @@ require 'fileutils' +require 'active_support/queueing' require 'rails/engine' module Rails @@ -188,7 +189,7 @@ module Rails end def queue #:nodoc: - @queue ||= Queueing::Container.new(build_queue) + @queue ||= ActiveSupport::QueueContainer.new(build_queue) end def build_queue #:nodoc: diff --git a/railties/lib/rails/application/configuration.rb b/railties/lib/rails/application/configuration.rb index 7a0bb21043..c4c1fcca40 100644 --- a/railties/lib/rails/application/configuration.rb +++ b/railties/lib/rails/application/configuration.rb @@ -1,5 +1,6 @@ require 'active_support/core_ext/kernel/reporting' require 'active_support/file_update_checker' +require 'active_support/queueing' require 'rails/engine/configuration' module Rails @@ -41,8 +42,8 @@ module Rails @exceptions_app = nil @autoflush_log = true @log_formatter = ActiveSupport::Logger::SimpleFormatter.new - @queue = Rails::Queueing::SynchronousQueue - @queue_consumer = Rails::Queueing::ThreadedConsumer + @queue = ActiveSupport::SynchronousQueue + @queue_consumer = ActiveSupport::ThreadedQueueConsumer @eager_load = nil @assets = ActiveSupport::OrderedOptions.new diff --git a/railties/lib/rails/application/finisher.rb b/railties/lib/rails/application/finisher.rb index 777f0d269e..f9a3c00946 100644 --- a/railties/lib/rails/application/finisher.rb +++ b/railties/lib/rails/application/finisher.rb @@ -97,8 +97,8 @@ module Rails end initializer :activate_queue_consumer do |app| - if config.queue == Rails::Queueing::Queue - app.queue_consumer = config.queue_consumer.start(app.queue) + if config.queue == ActiveSupport::Queue + app.queue_consumer = config.queue_consumer.start(app.queue, Rails.logger) at_exit { app.queue_consumer.shutdown } end end diff --git a/railties/lib/rails/generators/rails/app/templates/config/environments/production.rb.tt b/railties/lib/rails/generators/rails/app/templates/config/environments/production.rb.tt index fdf011a510..cb3e8b123e 100644 --- a/railties/lib/rails/generators/rails/app/templates/config/environments/production.rb.tt +++ b/railties/lib/rails/generators/rails/app/templates/config/environments/production.rb.tt @@ -82,5 +82,5 @@ # Default the production mode queue to an synchronous queue. You will probably # want to replace this with an out-of-process queueing solution. - # config.queue = Rails::Queueing::SynchronousQueue + # config.queue = ActiveSupport::SynchronousQueue end diff --git a/railties/lib/rails/generators/rails/app/templates/config/environments/test.rb.tt b/railties/lib/rails/generators/rails/app/templates/config/environments/test.rb.tt index 8ab27eb6e1..75897ba8cd 100644 --- a/railties/lib/rails/generators/rails/app/templates/config/environments/test.rb.tt +++ b/railties/lib/rails/generators/rails/app/templates/config/environments/test.rb.tt @@ -40,5 +40,5 @@ config.active_support.deprecation = :stderr # Use the testing queue. - config.queue = Rails::Queueing::TestQueue + config.queue = ActiveSupport::TestQueue end diff --git a/railties/lib/rails/queueing.rb b/railties/lib/rails/queueing.rb deleted file mode 100644 index 7cd755b0f7..0000000000 --- a/railties/lib/rails/queueing.rb +++ /dev/null @@ -1,115 +0,0 @@ -require "thread" -require 'delegate' - -module Rails - module Queueing - # A container for multiple queues. This class delegates to a default Queue - # so that <tt>Rails.queue.push</tt> and friends will Just Work. To use this class - # with multiple queues: - # - # # In your configuration: - # Rails.queue[:image_queue] = SomeQueue.new - # Rails.queue[:mail_queue] = SomeQueue.new - # - # # In your app code: - # Rails.queue[:mail_queue].push SomeJob.new - # - class Container < DelegateClass(::Queue) - def initialize(default_queue) - @queues = { :default => default_queue } - super(default_queue) - end - - def [](queue_name) - @queues[queue_name] - end - - def []=(queue_name, queue) - @queues[queue_name] = queue - end - end - - # A Queue that simply inherits from STDLIB's Queue. Everytime this - # queue is used, Rails automatically sets up a ThreadedConsumer - # to consume it. - class Queue < ::Queue - end - - class SynchronousQueue < ::Queue - def push(job) - job.run - end - alias << push - alias enq push - end - - # In test mode, the Rails queue is backed by an Array so that assertions - # can be made about its contents. The test queue provides a +jobs+ - # method to make assertions about the queue's contents and a +drain+ - # method to drain the queue and run the jobs. - # - # 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 - # Get a list of the jobs off this queue. This method may not be - # available on production queues. - def jobs - @que.dup - end - - # Marshal and unmarshal job before pushing it onto the queue. This will - # raise an exception on any attempts in tests to push jobs that can't (or - # shouldn't) be marshalled. - 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 - - # The threaded consumer will run jobs in a background thread in - # development mode or in a VM where running jobs on a thread in - # production mode makes sense. - # - # When the process exits, the consumer pushes a nil onto the - # queue and joins the thread, which will ensure that all jobs - # are executed before the process finally dies. - class ThreadedConsumer - def self.start(queue) - new(queue).start - end - - def initialize(queue) - @queue = queue - end - - def start - @thread = Thread.new do - while job = @queue.pop - begin - job.run - rescue Exception => e - handle_exception e - end - end - end - self - end - - def shutdown - @queue.push nil - @thread.join - end - - def handle_exception(e) - Rails.logger.error "Job Error: #{e.message}\n#{e.backtrace.join("\n")}" - end - end - end -end |