ruby beetle: support creating queue policies synchronously
MODIFY
lib/beetle/base.rb
lib/beetle/base.rb
@@ -97,6 +97,10 @@ module Beetle
# avoid endless recursion
return if options[:queue_name] == @client.config.beetle_policy_updates_queue_name
payload = options.merge(:server => @server)
+ if @client.config.update_queue_properties_synchronously
+ logger.debug("Beetle: updating policy options on #{@server}: #{payload.inspect}")
+ @client.update_queue_properties!(payload)
+ else
logger.debug("Beetle: publishing policy options on #{@server}: #{payload.inspect}")
# make sure to declare the queue, so the message does not get lost
ActiveSupport::Notifications.instrument('publish.beetle') do
@@ -106,6 +110,7 @@ module Beetle
exchange(@client.config.beetle_policy_exchange_name).publish(data, opts)
end
end
+ end
end
end

MODIFY
lib/beetle/configuration.rb
lib/beetle/configuration.rb
@@ -97,10 +97,13 @@ module Beetle
attr_accessor :dead_lettering_enabled
alias_method :dead_lettering_enabled?, :dead_lettering_enabled
- # the time a message spends in the dead letter queue if dead lettering is enabled, before it is returned
+ # The time a message spends in the dead letter queue if dead lettering is enabled, before it is returned
# to the original queue
attr_accessor :dead_lettering_msg_ttl
+ # Whether to update quueue policies synchronously or asynchronously.
+ attr_accessor :update_queue_properties_synchronously
+
# Read timeout for http requests to create dead letter bindings
attr_accessor :rabbitmq_api_read_timeout
@@ -175,6 +178,8 @@ module Beetle
self.lazy_queues_enabled = false
self.throttling_refresh_interval = 60 # seconds
+ self.update_queue_properties_synchronously = false
+
self.publishing_timeout = 0
self.publisher_connect_timeout = 5 # seconds
self.tmpdir = "/tmp"

MODIFY
test/beetle/base_test.rb
test/beetle/base_test.rb
@@ -104,5 +104,13 @@ module Beetle
@bs.__send__(:publish_policy_options, options)
end
+ test "publish_policy_options calls the RabbitMQ API if asked to do so" do
+ options = { :lazy => true, :dead_lettering => true }
+ @bs.logger.stubs(:debug)
+ @client.config.expects(:update_queue_properties_synchronously).returns(true)
+ @client.expects(:update_queue_properties!).with(options.merge(:server => "localhost:5672"))
+ @bs.__send__(:publish_policy_options, options)
+ end
+
end
end

ruby beetle: support creating queue policies synchronously