Merge branch 'master' into fix_tube_names_override

Conflicts:
	lib/backburner/workers/threads_on_fork.rb

CHANGELOG.md

@@ -1,5 +1,13 @@
 # CHANGELOG
 
+## Version 0.4.6 (October 26 2014)
+
+ * NEW Add job to on_error handler if the handler has a 4th argument (@Nitrodist)
+ * NEW Use a timeout when looking for a job to reserve (@EasyPost)
+ * NEW Support configuring settings on threads on fork class (@silentshade)
+ * FIX queue override by existing queues (@silentshade)
+ * FIX Use thread to log exit message (@silentshade)
+
 ## Version 0.4.5 (December 16 2013)
 
  * FIX #47 Create a backburner connection per thread (Thanks @thcrock)

README.md

@@ -99,6 +99,7 @@ Backburner.configure do |config|
   config.logger           = Logger.new(STDOUT)
   config.primary_queue    = "backburner-jobs"
   config.priority_labels  = { :custom => 50, :useless => 1000 }
+  config.reserve_timeout  = nil
 end
 ```
 
@@ -115,6 +116,7 @@ The key options available are:
 | `logger`          | Logger recorded to when backburner wants to report info or errors.   |
 | `primary_queue`   | Primary queue used for a job when an alternate queue is not given.   |
 | `priority_labels` | Hash of named priority definitions for your app.                     |
+| `reserve_timeout` | Duration to wait for work from a single server, or nil for forever.  |
 
 ## Breaking Changes
 
@@ -227,7 +229,7 @@ Backburner.work
 This will process jobs in all queues but you can also restrict processing to specific queues:
 
 ```ruby
-Backburner.work('newsletter-sender,push-notifier')
+Backburner.work('newsletter-sender', 'push-notifier')
 ```
 
 The Backburner worker also exists as a rake task:
@@ -239,7 +241,7 @@ require 'backburner/tasks'
 so you can run:
 
 ```
-$ QUEUES=newsletter-sender,push-notifier rake backburner:work
+$ QUEUE=newsletter-sender,push-notifier rake backburner:work
 ```
 
 You can also run the backburner binary for a convenient worker:
@@ -360,7 +362,7 @@ Backburner.work('newsletter-sender', :worker => Backburner::Workers::ThreadsOnFo
 or through associated rake tasks with:
 
 ```
-$ QUEUES=newsletter-sender,push-message THREADS=2 GARBAGE=1000 rake backburner:threads_on_fork:work
+$ QUEUE=newsletter-sender,push-message THREADS=2 GARBAGE=1000 rake backburner:threads_on_fork:work
 ```
 
 For more information on the threads_on_fork worker, check out the
@@ -459,7 +461,7 @@ and then you can start the rake task with:
 
 ```bash
 $ rake backburner:work
-$ QUEUES=newsletter-sender,push-notifier rake backburner:work
+$ QUEUE=newsletter-sender,push-notifier rake backburner:work
 ```
 
 The best way to deploy these rake tasks is using a monitoring library. We suggest [God](https://github.com/mojombo/god/)
@@ -482,6 +484,8 @@ jobs processed by your beanstalk workers. An excellent addition to your Backburn
  * Kristen Tucker - Coming up with the gem name
  * [Tim Lee](https://github.com/timothy1ee), [Josh Hull](https://github.com/joshbuddy), [Nico Taing](https://github.com/Nico-Taing) - Helping me work through the idea
  * [Miso](http://gomiso.com) - Open-source friendly place to work
+ * [Evgeniy Denisov](https://github.com/silentshade) - Multiple fixes and cleanups
+ * [Andy Bakun](https://github.com/thwarted) - Fixes to how multiple beanstalkd instances are processed
  * [Renan T. Fernandes](https://github.com/ShadowBelmolve) - Added threads_on_fork worker
  * [Daniel Farrell](https://github.com/danielfarrell) - Added forking worker
 

backburner.gemspec

@@ -17,7 +17,7 @@ Gem::Specification.new do |s|
   s.license       = 'MIT'
 
   s.add_runtime_dependency 'beaneater', '~> 0.3.1'
-  s.add_runtime_dependency 'dante', '~> 0.1.5'
+  s.add_runtime_dependency 'dante', '> 0.1.5'
 
   s.add_development_dependency 'rake'
   s.add_development_dependency 'minitest', '3.2.0'

examples/god.rb

@@ -4,7 +4,7 @@
   w.env = { 'PADRINO_ENV' => 'production', 'QUEUES' => 'newsletter-sender,push-message' }
   w.group    = 'backburner-workers'
   w.interval = 30.seconds
-  w.start = "bundle exec rake -f Rakefile backburner:start"
+  w.start = "bundle exec rake -f Rakefile backburner:work"
   w.log   = "/var/log/god/backburner-worker-1.log"
 
   # restart if memory gets too high
@@ -43,4 +43,4 @@
       c.running = false
     end
   end
-end
+end

lib/backburner/cli.rb

@@ -17,7 +17,6 @@ def self.start(args)
       runner.execute do |opts|
         queues = (opts[:queues] ? opts[:queues].split(',') : nil) rescue nil
         load_enviroment(opts[:require])
-        p queues
         Backburner.work(queues)
       end
     end

lib/backburner/configuration.rb

@@ -14,6 +14,7 @@ class Configuration
     attr_accessor :default_worker     # default worker class
     attr_accessor :primary_queue      # the general queue
     attr_accessor :priority_labels    # priority labels
+    attr_accessor :reserve_timeout    # duration to wait to reserve on a single server
 
     def initialize
       @beanstalk_url     = "beanstalk://localhost"
@@ -28,6 +29,7 @@ def initialize
       @default_worker    = Backburner::Workers::Simple
       @primary_queue     = "backburner-jobs"
       @priority_labels   = PRIORITY_LABELS
+      @reserve_timeout   = nil
     end
   end # Configuration
-end # Backburner
+end # Backburner

lib/backburner/connection.rb

@@ -10,6 +10,7 @@ class BadURL < RuntimeError; end
     # `url` can be a string i.e 'localhost:3001' or an array of addresses.
     def initialize(url)
       @url = url
+      @beanstalk = nil
       connect!
     end
 
@@ -48,4 +49,4 @@ def beanstalk_host_and_port(uri_string)
       "#{uri.host}:#{uri.port || 11300}"
     end
   end # Connection
-end # Backburner
+end # Backburner

lib/backburner/helpers.rb

@@ -80,6 +80,7 @@ def queue_config
     # Expands a tube to include the prefix
     #
     # @example
+    #   expand_tube_name("foo_with_settings:3:100:6") # => <prefix>.foo_with_settings
     #   expand_tube_name("foo") # => <prefix>.foo
     #   expand_tube_name(FooJob) # => <prefix>.foo-job
     #
@@ -94,7 +95,7 @@ def expand_tube_name(tube)
       else # turn into a string
         tube.to_s
       end
-      [prefix.gsub(/\.$/, ''), dasherize(queue_name).gsub(/^#{prefix}/, '')].join(".").gsub(/\.+/, '.')
+      [prefix.gsub(/\.$/, ''), dasherize(queue_name).gsub(/^#{prefix}/, '')].join(".").gsub(/\.+/, '.').split(':').first
     end
 
     # Resolves job priority based on the value given. Can be integer, a class or nothing

lib/backburner/queue.rb

@@ -1,6 +1,12 @@
 module Backburner
   module Queue
     def self.included(base)
+      base.instance_variable_set(:@queue_name, nil)
+      base.instance_variable_set(:@queue_priority, nil)
+      base.instance_variable_set(:@queue_respond_timeout, nil)
+      base.instance_variable_set(:@queue_jobs_limit, nil)
+      base.instance_variable_set(:@queue_garbage_limit, nil)
+      base.instance_variable_set(:@queue_retry_limit, nil)
       base.extend ClassMethods
       Backburner::Worker.known_queue_classes << base
     end
@@ -47,6 +53,48 @@ def queue_respond_timeout(ttr=nil)
           @queue_respond_timeout
         end
       end
+
+      # Returns or assigns queue parallel active jobs limit (only ThreadsOnFork Worker)
+      #
+      # @example
+      #   queue_jobs_limit 5
+      #   @klass.queue_jobs_limit # => 5
+      #
+      def queue_jobs_limit(limit=nil)
+        if limit
+          @queue_jobs_limit = limit
+        else #accessor
+          @queue_jobs_limit
+        end
+      end
+
+      # Returns or assigns queue jobs garbage limit (only ThreadsOnFork Worker)
+      #
+      # @example
+      #   queue_garbage_limit 1000
+      #   @klass.queue_garbage_limit # => 1000
+      #
+      def queue_garbage_limit(limit=nil)
+        if limit
+          @queue_garbage_limit = limit
+        else #accessor
+          @queue_garbage_limit
+        end
+      end
+
+      # Returns or assigns queue retry limit (only ThreadsOnFork Worker)
+      #
+      # @example
+      #   queue_retry_limit 6
+      #   @klass.queue_retry_limit # => 6
+      #
+      def queue_retry_limit(limit=nil)
+        if limit
+          @queue_retry_limit = limit
+        else #accessor
+          @queue_retry_limit
+        end
+      end
     end # ClassMethods
   end # Queue
-end # Backburner
+end # Backburner

lib/backburner/version.rb

@@ -1,3 +1,3 @@
 module Backburner
-  VERSION = "0.4.5"
+  VERSION = "0.4.6"
 end

lib/backburner/worker.rb

@@ -58,22 +58,6 @@ def self.connection
       @connection ||= Connection.new(Backburner.configuration.beanstalk_url)
     end
 
-    # Retries the given command specified in the block several times if there is a connection error
-    # Used to execute beanstalkd commands in a retryable way
-    #
-    # @example
-    #   retryable_command { ... }
-    # @raise [Beaneater::NotConnected] If beanstalk fails to connect multiple times.
-    #
-    def self.retryable_command(max_tries=8, &block)
-      begin
-        yield
-      rescue Beaneater::NotConnected => e
-        retry_connection!(max_tries)
-        yield
-      end
-    end
-
     # List of tube names to be watched and processed
     attr_accessor :tube_names
 
@@ -112,7 +96,9 @@ def prepare
 
     # Triggers this worker to shutdown
     def shutdown
-      log_info 'Worker exiting...'
+      Thread.new do
+        log_info 'Worker exiting...'
+      end
       Kernel.exit
     end
 
@@ -138,7 +124,11 @@ def process_tube_names(tube_names)
     #
     def work_one_job(conn = nil)
       conn ||= self.connection
-      job = Backburner::Job.new(conn.tubes.reserve)
+      begin
+        job = Backburner::Job.new(conn.tubes.reserve(Backburner.configuration.reserve_timeout))
+      rescue Beaneater::TimedOutError => e
+        return
+      end
       self.log_job_begin(job.name, job.args)
       job.process
       self.log_job_end(job.name)
@@ -169,7 +159,7 @@ def work_one_job(conn = nil)
     def self.retryable_command(max_tries=8, &block)
       begin
         yield
-      rescue Beaneater::NotConnected => e
+      rescue Beaneater::NotConnected
         retry_connection!(max_tries)
         yield
       end