Add in config option to purge all queues on startup.

emque · Nov 1, 2016 · 08db6b8 · 08db6b8
1 parent 348adc7
commit 08db6b8
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 17 deletions.
diff --git a/lib/emque/consuming/adapters/rabbit_mq/worker.rb b/lib/emque/consuming/adapters/rabbit_mq/worker.rb
@@ -33,17 +33,22 @@ def initialize(connection, topic)
 
             self.queue =
               channel
-                .queue(
-                  "emque.#{config.app_name}.#{topic}",
-                  :durable => config.adapter.options[:durable],
-                  :auto_delete => config.adapter.options[:auto_delete],
-                  :arguments => {
-                    "x-dead-letter-exchange" => "#{config.app_name}.error"
-                  }
-                )
-                .bind(
-                  channel.fanout(topic, :durable => true, :auto_delete => false)
-                )
+              .queue(
+                "emque.#{config.app_name}.#{topic}",
+                :durable => config.adapter.options[:durable],
+                :auto_delete => config.adapter.options[:auto_delete],
+                :arguments => {
+                  "x-dead-letter-exchange" => "#{config.app_name}.error"
+                }
+              )
+              .bind(
+                channel.fanout(topic, :durable => true, :auto_delete => false)
+              )
+
+            if config.purge_queues_on_start
+              logger.info "#{log_prefix} is purging it's queue"
+              queue.purge
+            end
           end
 
           def start

diff --git a/lib/emque/consuming/configuration.rb b/lib/emque/consuming/configuration.rb
@@ -5,9 +5,9 @@ module Consuming
     class Configuration
       attr_accessor :app_name, :adapter, :auto_shutdown,
         :delayed_message_workers, :enable_delayed_message, :error_handlers,
-        :error_limit, :error_expiration, :retryable_errors,
-        :retryable_error_limit, :status, :status_port, :status_host,
-        :socket_path, :shutdown_handlers
+        :error_limit, :error_expiration, :purge_queues_on_start,
+        :retryable_errors, :retryable_error_limit, :status, :status_port,
+        :status_host, :socket_path, :shutdown_handlers
       attr_writer :env, :log_level
 
       def initialize
@@ -19,6 +19,7 @@ def initialize
         @error_limit             = 5
         @error_expiration        = 3600 # 60 minutes
         @log_level               = nil
+        @purge_queues_on_start   = false
         @retryable_errors        = []
         @retryable_error_limit   = 3
         @status_port             = 10000
@@ -56,6 +57,7 @@ def to_hsh
             :error_handlers,
             :error_limit,
             :error_expiration,
+            :purge_queues_on_start,
             :log_level,
             :retryable_errors,
             :retryable_error_limit,

diff --git a/spec/configuration_spec.rb b/spec/configuration_spec.rb
@@ -63,9 +63,9 @@
       accessors = [
         :app_name, :adapter, :auto_shutdown, :delayed_message_workers,
         :env, :enable_delayed_message, :error_handlers, :error_limit,
-        :error_expiration, :log_level, :retryable_errors,
-        :retryable_error_limit, :status_port, :status_host, :status,
-        :socket_path, :shutdown_handlers
+        :error_expiration, :purge_queues_on_start, :log_level,
+        :retryable_errors, :retryable_error_limit, :status_port, :status_host,
+        :status, :socket_path, :shutdown_handlers
       ]
       config = Emque::Consuming::Configuration.new