Merge pull request #6153 from carlosantoniodasilva/queue-consumer

Configurable queue consumer

7 files changed, 74 insertions, 12 deletions

diff --git a/railties/CHANGELOG.md b/railties/CHANGELOG.md
index bc34ced283..b7c042cee3 100644
--- a/railties/CHANGELOG.md
+++ b/railties/CHANGELOG.md
@@ -1,5 +1,9 @@
 ## Rails 4.0.0 (unreleased) ##
 
+*   Add `config.queue_consumer` to allow the default consumer to be configurable. *Carlos Antonio da Silva*
+
+*   Add Rails.queue as an interface with a default implementation that consumes jobs in a separate thread. *Yehuda Katz*
+
 *   Remove Rack::SSL in favour of ActionDispatch::SSL. *Rafael Mendonça França*
 
 *   Remove Active Resource from Rails framework. *Prem Sichangrist*
diff --git a/railties/lib/rails/application.rb b/railties/lib/rails/application.rb
index c7b19c964a..c4edbae55b 100644
--- a/railties/lib/rails/application.rb
+++ b/railties/lib/rails/application.rb
@@ -66,7 +66,7 @@ module Rails
       end
     end
 
-    attr_accessor :assets, :sandbox
+    attr_accessor :assets, :sandbox, :queue_consumer
     alias_method :sandbox?, :sandbox
     attr_reader :reloaders
     attr_writer :queue
diff --git a/railties/lib/rails/application/configuration.rb b/railties/lib/rails/application/configuration.rb
index 25bb680f69..a2e5dece16 100644
--- a/railties/lib/rails/application/configuration.rb
+++ b/railties/lib/rails/application/configuration.rb
@@ -8,10 +8,11 @@ module Rails
       attr_accessor :allow_concurrency, :asset_host, :asset_path, :assets, :autoflush_log,
                     :cache_classes, :cache_store, :consider_all_requests_local, :console,
                     :dependency_loading, :exceptions_app, :file_watcher, :filter_parameters,
-                    :force_ssl, :helpers_paths, :logger, :log_formatter, :log_tags, :preload_frameworks,
-                    :railties_order, :relative_url_root, :secret_token,
+                    :force_ssl, :helpers_paths, :logger, :log_formatter, :log_tags,
+                    :preload_frameworks, :railties_order, :relative_url_root, :secret_token,
                     :serve_static_assets, :ssl_options, :static_cache_control, :session_options,
-                    :time_zone, :reload_classes_only_on_change, :use_schema_cache_dump, :queue
+                    :time_zone, :reload_classes_only_on_change, :use_schema_cache_dump,
+                    :queue, :queue_consumer
 
       attr_writer :log_level
       attr_reader :encoding
@@ -44,6 +45,7 @@ module Rails
         @log_formatter                 = ActiveSupport::Logger::SimpleFormatter.new
         @use_schema_cache_dump         = true
         @queue                         = Rails::Queueing::Queue
+        @queue_consumer                = Rails::Queueing::ThreadedConsumer
 
         @assets = ActiveSupport::OrderedOptions.new
         @assets.enabled                  = false
diff --git a/railties/lib/rails/application/finisher.rb b/railties/lib/rails/application/finisher.rb
index 6a24e01f29..84f2601f28 100644
--- a/railties/lib/rails/application/finisher.rb
+++ b/railties/lib/rails/application/finisher.rb
@@ -96,8 +96,8 @@ module Rails
 
       initializer :activate_queue_consumer do |app|
         if config.queue == Rails::Queueing::Queue
-          consumer = Rails::Queueing::ThreadedConsumer.start(app.queue)
-          at_exit { consumer.shutdown }
+          app.queue_consumer = config.queue_consumer.start(app.queue)
+          at_exit { app.queue_consumer.shutdown }
         end
       end
     end
diff --git a/railties/lib/rails/queueing.rb b/railties/lib/rails/queueing.rb
index 2e187b8555..b4bc7fcd18 100644
--- a/railties/lib/rails/queueing.rb
+++ b/railties/lib/rails/queueing.rb
@@ -16,13 +16,13 @@ module Rails
     # 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
+      # Get a list of the jobs off this queue. This method may not be
       # available on production queues.
       def jobs
         @que.dup
       end
 
-      # Drain the queue, running all jobs in a different thread.  This method
+      # 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
@@ -53,7 +53,7 @@ module Rails
             begin
               job.run
             rescue Exception => e
-              Rails.logger.error "Job Error: #{e.message}\n#{e.backtrace.join("\n")}"
+              handle_exception e
             end
           end
         end
@@ -64,6 +64,10 @@ module Rails
         @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
diff --git a/railties/test/application/queue_test.rb b/railties/test/application/queue_test.rb
index 71b0c2bc38..da8bdeed52 100644
--- a/railties/test/application/queue_test.rb
+++ b/railties/test/application/queue_test.rb
@@ -63,7 +63,7 @@ module ApplicationTests
     test "in test mode, the queue can be observed" do
       app("test")
 
-      job = Class.new(Struct.new(:id)) do
+      job = Struct.new(:id) do
         def run
         end
       end
@@ -79,7 +79,7 @@ module ApplicationTests
       assert_equal jobs, Rails.queue.jobs
     end
 
-    test "a custom queue implementation can be provided" do
+    def setup_custom_queue
       add_to_env_config "production", <<-RUBY
         require "my_queue"
         config.queue = MyQueue
@@ -94,10 +94,14 @@ module ApplicationTests
       RUBY
 
       app("production")
+    end
+
+    test "a custom queue implementation can be provided" do
+      setup_custom_queue
 
       assert_kind_of MyQueue, Rails.queue
 
-      job = Class.new(Struct.new(:id, :ran)) do
+      job = Struct.new(:id, :ran) do
         def run
           self.ran = true
         end
@@ -108,5 +112,34 @@ module ApplicationTests
 
       assert_equal true, job1.ran
     end
+
+    test "a custom consumer implementation can be provided" do
+      add_to_env_config "production", <<-RUBY
+        require "my_queue_consumer"
+        config.queue_consumer = MyQueueConsumer
+      RUBY
+
+      app_file "lib/my_queue_consumer.rb", <<-RUBY
+        class MyQueueConsumer < Rails::Queueing::ThreadedConsumer
+          attr_reader :started
+
+          def start
+            @started = true
+            self
+          end
+        end
+      RUBY
+
+      app("production")
+
+      assert_kind_of MyQueueConsumer, Rails.application.queue_consumer
+      assert Rails.application.queue_consumer.started
+    end
+
+    test "default consumer is not used with custom queue implementation" do
+      setup_custom_queue
+
+      assert_nil Rails.application.queue_consumer
+    end
   end
 end
diff --git a/railties/test/queueing/threaded_consumer_test.rb b/railties/test/queueing/threaded_consumer_test.rb
index 559de2a82d..c34a966d6e 100644
--- a/railties/test/queueing/threaded_consumer_test.rb
+++ b/railties/test/queueing/threaded_consumer_test.rb
@@ -78,4 +78,23 @@ class TestThreadConsumer < ActiveSupport::TestCase
     assert_equal 1, logger.logged(:error).size
     assert_match(/Job Error: RuntimeError: Error!/, logger.logged(:error).last)
   end
+
+  test "test overriding exception handling" do
+    @consumer.shutdown
+    @consumer = Class.new(Rails::Queueing::ThreadedConsumer) do
+      attr_reader :last_error
+      def handle_exception(e)
+        @last_error = e.message
+      end
+    end.start(@queue)
+
+    job = Job.new(1) do
+      raise "RuntimeError: Error!"
+    end
+
+    @queue.push job
+    sleep 0.1
+
+    assert_equal "RuntimeError: Error!", @consumer.last_error
+  end
 end