Merge pull request #120 from jaronkk/max-retry-count

Fix max retry count for new version of RabbitMQ x-death format

lib/sneakers/handlers/maxretry.rb

@@ -165,9 +165,16 @@ def failure_count(headers)
         if headers.nil? || headers['x-death'].nil?
           0
         else
-          headers['x-death'].select do |x_death|
+          x_death_array = headers['x-death'].select do |x_death|
             x_death['queue'] == @worker_queue_name
-          end.count
+          end
+          if x_death_array.count > 0 && x_death_array.first['count']
+            # Newer versions of RabbitMQ return headers with a count key
+            x_death_array.inject(0) {|sum, x_death| sum + x_death['count']}
+          else
+            # Older versions return a separate x-death header for each failure
+            x_death_array.count
+          end
         end
       end
       private :failure_count

spec/sneakers/worker_handlers_spec.rb

@@ -103,6 +103,31 @@ def process(*args,&block)
 
   describe 'Maxretry' do
     let(:max_retries) { nil }
+    let(:props_with_x_death_count) {
+      {
+        :headers => {
+          "x-death" => [
+                        {
+                          "count" => 3,
+                          "reason" => "expired",
+                          "queue" => "downloads-retry",
+                          "time" => Time.now,
+                          "exchange" => "RawMail-retry",
+                          "routing-keys" => ["RawMail"]
+                        },
+                        {
+                          "count" => 3,
+                          "reason" => "rejected",
+                          "queue" => "downloads",
+                          "time" => Time.now,
+                          "exchange" => "",
+                          "routing-keys" => ["RawMail"]
+                        }
+                       ]
+        },
+        :delivery_mode => 1
+      }
+    }
 
     before(:each) do
       @opts = {
@@ -228,6 +253,21 @@ def publish(data, opts)
             Time.parse(data['failed_at']).wont_be_nil
           end
 
+          it 'counts the number of attempts using the count key' do
+            mock(@header).routing_key { '#' }
+            mock(channel).acknowledge(37, false)
+
+            @error_exchange.extend MockPublish
+            worker.do_work(@header, props_with_x_death_count, :reject, @handler)
+            @error_exchange.called.must_equal(true)
+            @error_exchange.opts.must_equal({ :routing_key => '#' })
+            data = JSON.parse(@error_exchange.data)
+            data['error'].must_equal('reject')
+            data['num_attempts'].must_equal(4)
+            data['payload'].must_equal(Base64.encode64(:reject.to_s))
+            Time.parse(data['failed_at']).wont_be_nil
+          end
+
         end
       end