Module: Emque::Consuming::RetryableErrors

Includes:
Helpers
Included in:
Adapters::RabbitMq::DelayedMessageWorker, Adapters::RabbitMq::Worker
Defined in:
lib/emque/consuming/retryable_errors.rb

Instance Method Summary collapse

Instance Method Details

#delay_ms_time(retry_count) ⇒ Object 



14
15
16
 
# File 'lib/emque/consuming/retryable_errors.rb', line 14

def delay_ms_time(retry_count)
  retry_count * 500 * ( 2 ** retry_count)
end

#retry_error(delivery_info, metadata, payload, ex) ⇒ Object 



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/emque/consuming/retryable_errors.rb', line 18

def retry_error(delivery_info, metadata, payload, ex)
  headers = metadata[:headers] || {}
  retry_count = headers.fetch("x-retry-count", 0)

  if retry_count <= retryable_error_limit
    logger.info("Retrying Retryable Error #{ex.class}, with count " +
                "#{retry_count}")
    headers["x-retry-count"] = retry_count + 1
    headers["x-delay"] = delay_ms_time(retry_count)
    channel.ack(delivery_info.delivery_tag)
    delayed_message_exchange.publish(payload, { :headers => headers })
  else
    logger.info("Retryable Error: #{ex.class} ran out of retries at " +
                 "#{retry_count}")
    channel.nack(delivery_info.delivery_tag)
  end
end

#retryable_error_limit ⇒ Object 



10
11
12
 
# File 'lib/emque/consuming/retryable_errors.rb', line 10

def retryable_error_limit
  config.retryable_error_limit
end

#retryable_errors ⇒ Object 



6
7
8
 
# File 'lib/emque/consuming/retryable_errors.rb', line 6

def retryable_errors
  config.retryable_errors
end