Permalink
Browse files

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.
  • Loading branch information...
1 parent b24f1ce commit 1385388452c6dc86afe0668c41e0f5a491dc193a @carlosantoniodasilva carlosantoniodasilva committed May 3, 2012
@@ -66,7 +66,7 @@ def inherited(base)
end
end
- attr_accessor :assets, :sandbox
+ attr_accessor :assets, :sandbox, :queue_consumer
alias_method :sandbox?, :sandbox
attr_reader :reloaders
attr_writer :queue
@@ -8,10 +8,11 @@ class Configuration < ::Rails::Engine::Configuration
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 @@ def initialize(*)
@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
@@ -96,8 +96,8 @@ module Finisher
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
@@ -63,7 +63,7 @@ def job.run
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 @@ def run
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 @@ def push(job)
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 @@ def run
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

0 comments on commit 1385388

Please sign in to comment.