Permalink
Browse files

Merge pull request #6153 from carlosantoniodasilva/queue-consumer

Configurable queue consumer
  • Loading branch information...
2 parents 3d9673d + 7a6116b commit ffad3600ea01cda457b447fafd4d38907f4d0bfb @josevalim josevalim committed May 4, 2012
@@ -76,6 +76,17 @@ NOTE. The +config.asset_path+ configuration is ignored if the asset pipeline is
* +config.consider_all_requests_local+ is a flag. If true then any error will cause detailed debugging information to be dumped in the HTTP response, and the +Rails::Info+ controller will show the application runtime context in +/rails/info/properties+. True by default in development and test environments, and false in production mode. For finer-grained control, set this to false and implement +local_request?+ in controllers to specify which requests should provide debugging information on errors.
+* +config.console+ allows you to set class that will be used as console you run +rails console+. It's best to run it in +console+ block:
+
+<ruby>
+console do
+ # this block is called only when running console,
+ # so we can safely require pry here
+ require "pry"
+ config.console = Pry
+end
+</ruby>
+
* +config.dependency_loading+ is a flag that allows you to disable constant autoloading setting it to false. It only has effect if +config.cache_classes+ is true, which it is by default in production mode. This flag is set to false by +config.threadsafe!+.
* +config.eager_load_paths+ accepts an array of paths from which Rails will eager load on boot if cache classes is enabled. Defaults to every folder in the +app+ directory of the application.
@@ -100,6 +111,10 @@ NOTE. The +config.asset_path+ configuration is ignored if the asset pipeline is
* +config.preload_frameworks+ enables or disables preloading all frameworks at startup. Enabled by +config.threadsafe!+. Defaults to +nil+, so is disabled.
+* +config.queue+ configures a different queue implementation for the application. Defaults to +Rails::Queueing::Queue+. Note that, if the default queue is changed, the default +queue_consumer+ is not going to be initialized, it is up to the new queue implementation to handle starting and shutting down its own consumer(s).
+
+* +config.queue_consumer+ configures a different consumer implementation for the default queue. Defaults to +Rails::Queueing::ThreadedConsumer+.
+
* +config.reload_classes_only_on_change+ enables or disables reloading of classes only when tracked files change. By default tracks everything on autoload paths and is set to true. If +config.cache_classes+ is true, this option is ignored.
* +config.secret_token+ used for specifying a key which allows sessions for the application to be verified against a known secure key to prevent tampering. Applications get +config.secret_token+ initialized to a random key in +config/initializers/secret_token.rb+.
@@ -122,17 +137,6 @@ WARNING: Threadsafe operation is incompatible with the normal workings of develo
* +config.whiny_nils+ enables or disables warnings when a certain set of methods are invoked on +nil+ and it does not respond to them. Defaults to true in development and test environments.
-* +config.console+ allows you to set class that will be used as console you run +rails console+. It's best to run it in +console+ block:
-
-<ruby>
-console do
- # this block is called only when running console,
- # so we can safely require pry here
- require "pry"
- config.console = Pry
-end
-</ruby>
-
h4. Configuring Assets
Rails 3.1, by default, is set up to use the +sprockets+ gem to manage assets within an application. This gem concatenates and compresses assets in order to make serving them much less painful.
View
@@ -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*
@@ -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
@@ -16,13 +16,13 @@ class Queue < ::Queue
# 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 @@ def start
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 @@ 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
@@ -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
@@ -78,4 +78,23 @@ def teardown
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

0 comments on commit ffad360

Please sign in to comment.