introduce broker_default_policy to reduce number of policies on server
MODIFY
lib/beetle/configuration.rb
lib/beetle/configuration.rb
@@ -14,6 +14,9 @@ module Beetle
# Name of the policy update routing key
attr_accessor :beetle_policy_updates_routing_key
# default logger (defaults to <tt>Logger.new(log_file)</tt>)
+ attr_accessor :broker_default_policy
+ # set this to whatever your brokers have installed as default policy. For example, if you have installed
+ # a policy that makes every queue lazy, it should be set to <tt>{"queue-moode" => "lazy"}</tt>.
attr_accessor :logger
# defaults to <tt>STDOUT</tt>
attr_accessor :redis_logger
@@ -146,6 +149,7 @@ module Beetle
self.beetle_policy_exchange_name = "beetle-policies"
self.beetle_policy_updates_queue_name = "beetle-policy-updates"
self.beetle_policy_updates_routing_key = "beetle.policy.update"
+ self.broker_default_policy = {}
self.gc_threshold = 1.hour.to_i
self.redis_server = "localhost:6379"

MODIFY
lib/beetle/queue_properties.rb
lib/beetle/queue_properties.rb
@@ -45,9 +45,11 @@ module Beetle
return unless options[:dead_lettering] || options[:lazy]
# no need to worry that the server has the port 5672. Net:HTTP will take care of this. See below.
- request_url = URI("http://#{server}/api/policies/#{vhost}/#{queue_name}_policy")
+ policy_name = "#{queue_name}_policy"
+ request_url = URI("http://#{server}/api/policies/#{vhost}/#{policy_name}")
get_request = Net::HTTP::Get.new(request_url)
put_request = Net::HTTP::Put.new(request_url)
+ delete_request = Net::HTTP::Delete.new(request_url)
# set up queue policy
definition = {}
@@ -66,13 +68,26 @@ module Beetle
"definition" => definition,
}
+ is_default_policy = definition == config.broker_default_policy
+
get_response = run_rabbit_http_request(request_url, get_request) do |http|
http.request(get_request, nil)
end
- if get_response.code == "200"
+ case get_response.code
+ when "200"
response_body = JSON.parse(get_response.body) rescue {}
- return :ok if put_request_body.all? { |k,v| response_body[k] == v }
+ same_policy = put_request_body.all? { |k,v| response_body[k] == v }
+ if same_policy
+ if is_default_policy
+ run_rabbit_http_request(request_url, delete_request) do |http|
+ http.request(get_request, nil)
+ end
+ end
+ return :ok
+ end
+ when "404"
+ return :ok if is_default_policy
end
put_response = run_rabbit_http_request(request_url, put_request) do |http|
@@ -141,9 +156,10 @@ module Beetle
def run_rabbit_http_request(uri, request, &block)
request.basic_auth(config.user, config.password)
- if request.class::METHOD == 'GET'
+ case request.class::METHOD
+ when 'GET'
request["Accept"] = "application/json"
- else
+ when 'PUT'
request["Content-Type"] = "application/json"
end
http = Net::HTTP.new(uri.hostname, config.api_port)

MODIFY
test/beetle/queue_properties_test.rb
test/beetle/queue_properties_test.rb
@@ -78,6 +78,36 @@ module Beetle
@queue_properties.set_queue_policy!(@server, @queue_name, :lazy => false, :dead_lettering => true, :routing_key => "QUEUE_NAME_dead_letter")
end
+ test "deletes policy if its definition corresponds to the broker default policy" do
+ @config.broker_default_policy = { "queue-mode" => "lazy" }
+ stub_request(:get, "http://localhost:15672/api/policies/%2F/QUEUE_NAME_policy")
+ .with(basic_auth: ['guest', 'guest'])
+ .to_return(:status => 200,
+ :body => {
+ "vhost" => "/",
+ "name" => "QUEUE_NAME_policy",
+ "pattern" => "^QUEUE_NAME$",
+ "priority" => 1,
+ "apply-to" => "queues",
+ "definition" => {
+ "queue-mode" => "lazy",
+ }}.to_json)
+ stub_request(:delete, "http://localhost:15672/api/policies/%2F/QUEUE_NAME_policy")
+ .with(basic_auth: ['guest', 'guest'])
+ .to_return(:status => 204)
+
+ @queue_properties.set_queue_policy!(@server, @queue_name, :lazy => true, :dead_lettering => false, :routing_key => "QUEUE_NAME_dead_letter")
+ end
+
+ test "does nothing if its definition corresponds to the broker default policy and the policy does not exist on the server" do
+ @config.broker_default_policy = { "queue-mode" => "lazy" }
+ stub_request(:get, "http://localhost:15672/api/policies/%2F/QUEUE_NAME_policy")
+ .with(basic_auth: ['guest', 'guest'])
+ .to_return(:status => 404)
+
+ @queue_properties.set_queue_policy!(@server, @queue_name, :lazy => true, :dead_lettering => false, :routing_key => "QUEUE_NAME_dead_letter")
+ end
+
test "creates a policy by posting to the rabbitmq if lazy queues are enabled" do
stub_request(:get, "http://localhost:15672/api/policies/%2F/QUEUE_NAME_policy")
.with(basic_auth: ['guest', 'guest'])

introduce broker_default_policy to reduce number of policies on server