Allow configuring a different queue consumer

Also make sure to not use default queue consumer with custom queue
implementation. It is up to the new queue implementation to
start / shutdown the consumer.

4 files changed, 44 insertions, 9 deletions

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/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