ruby client: allow queue policy specs per queue
MODIFY
lib/beetle/base.rb
lib/beetle/base.rb
@@ -12,7 +12,7 @@ module Beetle
@server = @servers[rand @servers.size]
@exchanges = {}
@queues = {}
- @dead_lettering = DeadLettering.new(@client.config)
+ @dead_lettering = DeadLettering.new(@client)
end
private

MODIFY
lib/beetle/client.rb
lib/beetle/client.rb
@@ -73,12 +73,19 @@ module Beetle
# the name of the exchange this queue will be bound to (defaults to the name of the queue)
# [<tt>:key</tt>]
# the binding key (defaults to the name of the queue)
+ # [<tt>:lazy</tt>]
+ # whether the queue should use lazy mode (defaults to <tt>config.lazy_queues_enabled</tt>)
+ # [<tt>:dead_lettering</tt>]
+ # whether the queue should use dead lettering (defaults to <tt>config.dead_lettering_enabled</tt>)
# automatically registers the specified exchange if it hasn't been registered yet
def register_queue(name, options={})
name = name.to_s
raise ConfigurationError.new("queue #{name} already configured") if queues.include?(name)
- opts = {:exchange => name, :key => name, :auto_delete => false, :amqp_name => name}.merge!(options.symbolize_keys)
+ opts = {
+ :exchange => name, :key => name, :auto_delete => false, :amqp_name => name,
+ :lazy => config.lazy_queues_enabled, :dead_lettering => config.dead_lettering_enabled
+ }.merge!(options.symbolize_keys)
opts.merge! :durable => true, :passive => false, :exclusive => false
exchange = opts.delete(:exchange).to_s
key = opts.delete(:key)

MODIFY
lib/beetle/dead_lettering.rb
lib/beetle/dead_lettering.rb
@@ -5,29 +5,34 @@ module Beetle
class DeadLettering
class FailedRabbitRequest < StandardError; end
- def initialize(config)
- @config = config
+ attr_reader :client, :config
+
+ def initialize(client)
+ @client = client
+ @config = client.config
end
def bind_dead_letter_queues!(channel, servers, target_queue, creation_keys = {})
- if @config.dead_lettering_enabled?
+ policy_options = client.queues[target_queue].slice(:dead_lettering, :lazy)
+ if policy_options[:dead_lettering]
dead_letter_queue_name = dead_letter_queue_name(target_queue)
logger.info("Beetle: creating dead letter queue #{dead_letter_queue_name} with opts: #{creation_keys.inspect}")
channel.queue(dead_letter_queue_name, creation_keys)
end
- if @config.dead_lettering_enabled?
+ if policy_options[:dead_lettering]
logger.info("Beetle: setting #{dead_letter_queue_name} as dead letter queue of #{target_queue} on all servers")
end
- set_queue_policies!(servers, target_queue)
+ set_queue_policies!(servers, target_queue, policy_options)
- if @config.dead_lettering_enabled?
+ if policy_options[:dead_lettering]
logger.info("Beetle: setting #{target_queue} as dead letter queue of #{dead_letter_queue_name} on all servers")
set_queue_policies!(
servers,
dead_letter_queue_name,
- :message_ttl => @config.dead_lettering_msg_ttl,
+ { :message_ttl => config.dead_lettering_msg_ttl,
:routing_key => target_queue
+ }.merge(policy_options)
)
end
end
@@ -40,21 +45,21 @@ module Beetle
raise ArgumentError.new("server missing") if server.blank?
raise ArgumentError.new("queue name missing") if queue_name.blank?
- return unless @config.dead_lettering_enabled? || @config.lazy_queues_enabled?
+ return unless options[:dead_lettering] || options[:lazy]
- vhost = CGI.escape(@config.vhost)
+ vhost = CGI.escape(config.vhost)
request_url = URI("http://#{server}/api/policies/#{vhost}/#{queue_name}_policy")
request = Net::HTTP::Put.new(request_url)
# set up queue policy
definition = {}
- if @config.dead_lettering_enabled?
+ if options[:dead_lettering]
definition["dead-letter-routing-key"] = dead_letter_routing_key(queue_name, options)
definition["dead-letter-exchange"] = ""
definition["message-ttl"] = options[:message_ttl] if options[:message_ttl]
end
- definition["queue-mode"] = "lazy" if @config.lazy_queues_enabled?
+ definition["queue-mode"] = "lazy" if options[:lazy]
request_body = {
"pattern" => "^#{queue_name}$",
@@ -84,9 +89,9 @@ module Beetle
end
def run_rabbit_http_request(uri, request, &block)
- request.basic_auth(@config.user, @config.password)
+ request.basic_auth(config.user, config.password)
request["Content-Type"] = "application/json"
- Net::HTTP.start(uri.hostname, @config.api_port, :read_timeout => @config.dead_lettering_read_timeout) do |http|
+ Net::HTTP.start(uri.hostname, config.api_port, :read_timeout => config.dead_lettering_read_timeout) do |http|
block.call(http) if block_given?
end
end
@@ -98,7 +103,7 @@ module Beetle
end
def logger
- @config.logger
+ config.logger
end
end
end

MODIFY
test/beetle/client_test.rb
test/beetle/client_test.rb
@@ -79,7 +79,7 @@ module Beetle
test "registering a queue should store it in the configuration with symbolized option keys and force durable=true and passive=false and set the amqp queue name" do
@client.register_queue("some_queue", "durable" => false, "exchange" => "some_exchange")
- assert_equal({:durable => true, :passive => false, :auto_delete => false, :exclusive => false, :amqp_name => "some_queue"}, @client.queues["some_queue"])
+ assert_equal({:durable => true, :passive => false, :lazy=>false, :dead_lettering=>false, :auto_delete => false, :exclusive => false, :amqp_name => "some_queue"}, @client.queues["some_queue"])
end
test "registering a queue should add the queue to the list of queues of the queue's exchange" do

MODIFY
test/beetle/dead_lettering_test.rb
test/beetle/dead_lettering_test.rb
@@ -4,7 +4,8 @@ module Beetle
class SetDeadLetteringsTest < Minitest::Test
def setup
@config = Configuration.new
- @dead_lettering = DeadLettering.new(@config)
+ @client = Client.new @config
+ @dead_lettering = DeadLettering.new(@client)
@config.dead_lettering_enabled = true
end
@@ -25,9 +26,9 @@ module Beetle
@server = "localhost:15672"
@queue_name = "QUEUE_NAME"
@config = Configuration.new
+ @client = Client.new @config
@config.logger = Logger.new("/dev/null")
- @dead_lettering = DeadLettering.new(@config)
- @config.dead_lettering_enabled = true
+ @dead_lettering = DeadLettering.new(@client)
end
test "raises exception when queue name wasn't specified" do
@@ -55,12 +56,10 @@ module Beetle
}}.to_json)
.to_return(:status => 204)
- @dead_lettering.set_queue_policy!(@server, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :lazy => false, :dead_lettering => true)
end
test "creates a policy by posting to the rabbitmq if lazy queues are enabled" do
- @config.lazy_queues_enabled = true
- @config.dead_lettering_enabled = false
stub_request(:put, "http://localhost:15672/api/policies/%2F/QUEUE_NAME_policy")
.with(basic_auth: ['guest', 'guest'])
.with(:body => {
@@ -72,7 +71,7 @@ module Beetle
}}.to_json)
.to_return(:status => 204)
- @dead_lettering.set_queue_policy!(@server, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :lazy => true, :dead_lettering => false)
end
test "raises exception when policy couldn't successfully be created" do
@@ -81,7 +80,7 @@ module Beetle
.to_return(:status => [405])
assert_raises DeadLettering::FailedRabbitRequest do
- @dead_lettering.set_queue_policy!(@server, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :lazy => true, :dead_lettering => true)
end
end
@@ -99,7 +98,7 @@ module Beetle
}}.to_json)
.to_return(:status => 204)
- @dead_lettering.set_queue_policy!(@server, @queue_name, :message_ttl => 10000)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :dead_lettering => true, :message_ttl => 10000)
end
test "properly encodes the vhost from the configuration" do
@@ -117,7 +116,7 @@ module Beetle
@config.vhost = "foo/"
- @dead_lettering.set_queue_policy!(@server, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :dead_lettering => true)
end
end
@@ -125,28 +124,31 @@ module Beetle
def setup
@queue_name = "QUEUE_NAME"
@config = Configuration.new
+ @client = Client.new @config
@config.logger = Logger.new("/dev/null")
- @dead_lettering = DeadLettering.new(@config)
+ @dead_lettering = DeadLettering.new(@client)
@servers = ["localhost:55672"]
end
test "is does not call out to rabbit if neither dead lettering nor lazy queues are enabled" do
- @config.dead_lettering_enabled = false
+ @client.register_queue(@queue_name, :dead_lettering => false, :lazy => false)
channel = stub('channel')
@dead_lettering.expects(:run_rabbit_http_request).never
@dead_lettering.bind_dead_letter_queues!(channel, @servers, @queue_name)
end
test "creates and connects the dead letter queue via policies when enabled" do
- @config.dead_lettering_enabled = true
+ @client.register_queue(@queue_name, :dead_lettering => true, :lazy => false)
channel = stub('channel')
channel.expects(:queue).with("#{@queue_name}_dead_letter", {})
- @dead_lettering.expects(:set_queue_policies!).with(@servers, @queue_name)
+ @dead_lettering.expects(:set_queue_policies!).with(@servers, @queue_name, :dead_lettering => true, :lazy => false)
@dead_lettering.expects(:set_queue_policies!).with(@servers, "#{@queue_name}_dead_letter",
:routing_key => @queue_name,
- :message_ttl => 1000
+ :message_ttl => 1000,
+ :dead_lettering => true,
+ :lazy => false
)
@dead_lettering.bind_dead_letter_queues!(channel, @servers, @queue_name)

ruby client: allow queue policy specs per queue