removed ms resolution, added max delay for exp bo
MODIFY
examples/attempts_with_dead_letter_and_exponential_backoff.rb
examples/attempts_with_dead_letter_and_exponential_backoff.rb
@@ -12,7 +12,7 @@ Beetle.config.logger.level = Logger::INFO
# setup client with dead lettering enabled
config = Beetle::Configuration.new
config.dead_lettering_enabled = true
-config.dead_lettering_msg_ttl = 100 # millis
+config.dead_lettering_msg_ttl = 1000 # millis
client = Beetle::Client.new(config)
client.register_queue(:test)
client.register_message(:test)
@@ -25,7 +25,7 @@ client.deduplication_store.flushdb
# setup our counter
$completed = 0
-$exceptions_limit = 5
+$exceptions_limit = 4
# store the start time
$start_time = Time.now.to_f
@@ -48,7 +48,7 @@ class Handler < Beetle::Handler
end
# register our handler to the message, configure it to our max_attempts limit, we configure a (base) delay of 0.5
-client.register_handler(:test, Handler, exceptions: $exceptions_limit, delay: 0.2, exponential_back_off: true)
+client.register_handler(:test, Handler, exceptions: $exceptions_limit, delay: 1, exponential_back_off: true)
# publish test messages
client.publish(:test, 1) # publish returns the number of servers the message has been sent to
puts "published 1 test message"

MODIFY
examples/attempts_with_exponential_backoff.rb
examples/attempts_with_exponential_backoff.rb
@@ -45,8 +45,8 @@ class Handler < Beetle::Handler
end
end
-# register our handler to the message, configure it to our max_attempts limit, we configure a (base) delay of 0.5
-client.register_handler(:test, Handler, exceptions: $exceptions_limit, delay: 0.2, exponential_back_off: true)
+# register our handler to the message, configure it to our max_attempts limit, we configure a (base) delay of 1
+client.register_handler(:test, Handler, exceptions: $exceptions_limit, delay: 1, exponential_back_off: true)
# publish test messages
client.publish(:test, 1) # publish returns the number of servers the message has been sent to

MODIFY
lib/beetle/message.rb
lib/beetle/message.rb
@@ -13,14 +13,14 @@ module Beetle
# flag for encoding redundant messages
FLAG_REDUNDANT = 1
# default lifetime of messages
- DEFAULT_TTL = 86_400
+ DEFAULT_TTL = 1.day
# forcefully abort a running handler after this many seconds.
# can be overriden when registering a handler.
- DEFAULT_HANDLER_TIMEOUT = 600.0
+ DEFAULT_HANDLER_TIMEOUT = 600
# how many times we should try to run a handler before giving up
DEFAULT_HANDLER_EXECUTION_ATTEMPTS = 1
# how many seconds we should wait before retrying handler execution
- DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY = 10.0
+ DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY = 10
# how many exceptions should be tolerated before giving up
DEFAULT_EXCEPTION_LIMIT = 0
@@ -67,7 +67,7 @@ module Beetle
@server = opts[:server]
@timeout = opts[:timeout] || DEFAULT_HANDLER_TIMEOUT
@delay = opts[:delay] || DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY
- @exp_bo = !!opts[:exponential_back_off]
+ @max_exp_bo = opts[:exponential_back_off]&.to_i
@attempts_limit = opts[:attempts] || DEFAULT_HANDLER_EXECUTION_ATTEMPTS
@exceptions_limit = opts[:exceptions] || DEFAULT_EXCEPTION_LIMIT
@attempts_limit = @exceptions_limit + 1 if @attempts_limit <= @exceptions_limit
@@ -127,12 +127,12 @@ module Beetle
# current time (UNIX timestamp)
def now #:nodoc:
- Time.now.to_f
+ Time.now.to_i
end
# current time (UNIX timestamp)
def self.now #:nodoc:
- Time.now.to_f
+ Time.now.to_i
end
# a message has expired if the header expiration timestamp is msaller than the current time
@@ -162,7 +162,7 @@ module Beetle
# handler timed out?
def timed_out?
- (t = @store.get(msg_id, :timeout)) && t.to_f < now
+ (t = @store.get(msg_id, :timeout)) && t.to_i < now
end
# reset handler timeout in the deduplication store
@@ -182,7 +182,7 @@ module Beetle
# whether we should wait before running the handler
def delayed?
- (t = @store.get(msg_id, :delay)) && t.to_f > now
+ (t = @store.get(msg_id, :delay)) && t.to_i > now
end
# store delay value in the deduplication store
@@ -283,10 +283,10 @@ module Beetle
if status == "completed"
ack!
RC::OK
- elsif delay && delay.to_f > now
+ elsif delay && delay.to_i > now
logger.warn "Beetle: ignored delayed message (#{msg_id})!"
RC::Delayed
- elsif !(timeout && timeout.to_f < now)
+ elsif !(timeout && timeout.to_i < now)
RC::HandlerNotYetTimedOut
elsif attempts.to_i >= attempts_limit
completed!
@@ -366,15 +366,18 @@ module Beetle
end
def exponential_back_off?
- @exp_bo
+ @max_exp_bo.present?
+ end
+
+ def max_exp_bo_delay
+ @max_exp_bo
end
def calculate_next_delay(n, base_delay)
unless exponential_back_off?
base_delay
else
- base_delay_n = base_delay * (2**n)
- rand((base_delay_n - base_delay_n / 4.0)..(base_delay_n + base_delay_n / 2.0))
+ [base_delay * (2**n), max_exp_bo_delay].min
end
end
end

MODIFY
lib/beetle/subscriber.rb
lib/beetle/subscriber.rb
@@ -176,7 +176,7 @@ module Beetle
if @client.config.dead_lettering_enabled?
header.reject(:requeue => false)
else
- sleep 0.1
+ sleep 1
header.reject(:requeue => true)
end
elsif reply_to = header.attributes[:reply_to]

MODIFY
test/beetle/message/settings_test.rb
test/beetle/message/settings_test.rb
@@ -16,78 +16,79 @@ module Beetle
end
test "set_delay! should store the current time plus the delay offset in the database" do
- message = Message.new("somequeue", header_with_params, 'foo', :delay => 1.75, :store => @store)
- message.expects(:now).returns(1.2)
+ message = Message.new("somequeue", header_with_params, 'foo', :delay => 2, :store => @store)
+ message.expects(:now).returns(9)
message.set_delay!
- assert_equal "2.95", @store.get(message.msg_id, :delay)
- message.expects(:now).returns(3.0)
+ assert_equal "11", @store.get(message.msg_id, :delay)
+ message.expects(:now).returns(12)
assert !message.delayed?
- message.expects(:now).returns(0.0)
+ message.expects(:now).returns(10)
assert message.delayed?
end
test "set_delay! should store the current time plus the exponential delay offset in the database" do
- message = Message.new("somequeue", header_with_params, 'foo', :delay => 1.0, :exponential_back_off => true, :store => @store)
- [0.0, 1.5, 3.33].each do |now_offset|
- message.stubs(:now).returns(now_offset)
- [
- [0.75, 1.5],
- [1.5, 3.0],
- [3.0, 6.0],
- [6.0, 12.0],
- ].each do |(lower, upper)|
+ message = Message.new("somequeue", header_with_params, 'foo', :delay => 3, :exponential_back_off => 30, :store => @store)
+ message.stubs(:now).returns(1)
+
+ [4, 7, 13, 25].each do |exp_delay|
message.set_delay!
- assert @store.get(message.msg_id, :delay).to_f.between?(lower + now_offset, upper + now_offset)
+ assert_equal exp_delay, @store.get(message.msg_id, :delay).to_i
message.increment_execution_attempts!
end
- @store.set(message.msg_id, :attempts, 0)
+ end
+
+ test "set_delay! should store the current time plus the exponential delay offset in the database up to given max value" do
+ message = Message.new("somequeue", header_with_params, 'foo', :delay => 3, :exponential_back_off => 10, :store => @store)
+ message.stubs(:now).returns(1)
+
+ [4, 7, 11, 11].each do |exp_delay|
+ message.set_delay!
+ assert_equal exp_delay, @store.get(message.msg_id, :delay).to_i
+ message.increment_execution_attempts!
end
end
test "set_delay! should store the current time plus the linear delay offset in the database" do
- message = Message.new("somequeue", header_with_params, 'foo', :delay => 0.4, :exponential_back_off => false, :store => @store)
- [
- [1.6, 2.0],
- [2.1, 2.5],
- [2.7, 3.1],
- ].each do |now_offset, exp_delay|
+ delay = 32
+ message = Message.new("somequeue", header_with_params, 'foo', :delay => delay, :exponential_back_off => nil, :store => @store)
+ [3, 5, 6].each do |now_offset|
message.stubs(:now).returns(now_offset)
message.set_delay!
- assert_equal @store.get(message.msg_id, :delay).to_f, exp_delay
+ assert_equal @store.get(message.msg_id, :delay).to_i, now_offset + delay
message.increment_execution_attempts!
end
end
test "set_delay! should use the default delay if the delay hasn't been set on the message instance" do
message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
- message.expects(:now).returns(0.0)
+ message.expects(:now).returns(0)
message.set_delay!
assert_equal "#{Message::DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY}", @store.get(message.msg_id, :delay)
message.expects(:now).returns(message.delay)
assert !message.delayed?
- message.expects(:now).returns(0.0)
+ message.expects(:now).returns(0)
assert message.delayed?
end
test "set_timeout! should store the current time plus the number of timeout seconds in the database" do
message = Message.new("somequeue", header_with_params, 'foo', :timeout => 1, :store => @store)
- message.expects(:now).returns(1.0)
+ message.expects(:now).returns(1)
message.set_timeout!
- assert_equal "2.0", @store.get(message.msg_id, :timeout)
- message.expects(:now).returns(2.0)
+ assert_equal "2", @store.get(message.msg_id, :timeout)
+ message.expects(:now).returns(2)
assert !message.timed_out?
- message.expects(:now).returns(3.0)
+ message.expects(:now).returns(3)
assert message.timed_out?
end
test "set_timeout! should use the default timeout if the timeout hasn't been set on the message instance" do
message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
- message.expects(:now).returns(0.0)
+ message.expects(:now).returns(0)
message.set_timeout!
assert_equal "#{Message::DEFAULT_HANDLER_TIMEOUT}", @store.get(message.msg_id, :timeout)
message.expects(:now).returns(message.timeout)
assert !message.timed_out?
- message.expects(:now).returns(Message::DEFAULT_HANDLER_TIMEOUT + 1.0)
+ message.expects(:now).returns(Message::DEFAULT_HANDLER_TIMEOUT + 1)
assert message.timed_out?
end

MODIFY
test/beetle/subscriber_test.rb
test/beetle/subscriber_test.rb
@@ -292,7 +292,7 @@ module Beetle
result = mock("result")
result.expects(:reject?).returns(true)
Message.any_instance.expects(:process).returns(result)
- @sub.expects(:sleep).with(0.1)
+ @sub.expects(:sleep).with(1)
mq = mock("MQ")
mq.expects(:closing?).returns(false)
@sub.expects(:channel).with(@sub.server).returns(mq)

removed ms resolution, added max delay for exp bo