Merge pull request #8 from BranchIntl/shashankmehra/bep-8470-add-support-for-specifying-delayed-queue-options-in-advanced

Add support for specifying delayed queue options in advanced sneakers

Author: shashankmehra

README.md

@@ -175,6 +175,10 @@ AdvancedSneakersActiveJob.configure do |config|
   # Delayed queues can be filtered by this prefix (e.g. delayed:60 - queue for messages with 1 minute delay)
   config.delayed_queue_prefix = 'delayed'
 
+  # Delayed options can be customized (e.g. to remove 'x-queue-mode' => 'lazy' for Quorum queues)
+  # Default: { 'x-queue-mode' => 'lazy' }
+  # config.delayed_queue_options = {}
+
   # Custom sneakers configuration for ActiveJob publisher & runner
   config.sneakers = {
     connection: Bunny.new('CUSTOM_URL', with: { other: 'options' }),

lib/advanced_sneakers_activejob/configuration.rb

@@ -15,6 +15,7 @@ class Configuration
     config_accessor(:activejob_workers_strategy) { :include } # [:include, :exclude, :only]
     config_accessor(:delay_proc) { ->(timestamp) { (timestamp - Time.now.to_f).round } } # seconds
     config_accessor(:delayed_queue_prefix) { 'delayed' }
+    config_accessor(:delayed_queue_options) { { 'x-queue-mode' => 'lazy' } }
     config_accessor(:retry_delay_proc) { ->(count) { AdvancedSneakersActiveJob::EXPONENTIAL_BACKOFF[count] } } # seconds
     config_accessor(:log_level) { :info } # debug logs are too noizy because of Bunny
 

lib/advanced_sneakers_activejob/delayed_publisher.rb

@@ -8,7 +8,7 @@ class DelayedPublisher < ::BunnyPublisher::Base
 
     delegate :logger, to: :'::ActiveJob::Base'
 
-    delegate :name_prefix, :delayed_queue_prefix,
+    delegate :name_prefix, :delayed_queue_prefix, :delayed_queue_options,
              to: :'AdvancedSneakersActiveJob.config',
              prefix: :config
 
@@ -33,11 +33,10 @@ def log_message
     def declare_republish_queue
       queue_name = delayed_queue_name(delay: delay)
 
-      queue_arguments = {
-        'x-queue-mode' => 'lazy', # tell RabbitMQ not to use RAM for this queue as it won't be consumed
+      queue_arguments = config_delayed_queue_options.merge(
         'x-message-ttl' => delay * 1000, # make messages die after requested time
         'x-dead-letter-exchange' => dlx_exchange_name # dead messages go to original exchange and then routed to proper queues
-      }
+      )
 
       logger.debug { "Creating delayed queue [#{queue_name}]" }