ruby client: support lazy queues
MODIFY
examples/attempts.rb
examples/attempts.rb
@@ -16,6 +16,7 @@ Beetle.config.logger.level = Logger::INFO
# setup client
$client = Beetle::Client.new
$client.config.dead_lettering_enabled = true
+$client.config.lazy_queues_enabled = true
$client.configure(:key => "my.test.message") do
message(:test)
queue(:test)

MODIFY
lib/beetle/configuration.rb
lib/beetle/configuration.rb
@@ -62,6 +62,11 @@ module Beetle
# in bunny.
attr_accessor :channel_max
+ # Lazy queues have the advantage of consuming a lot less memory on the broker. For backwards
+ # compatibility, they are disabled by default.
+ attr_accessor :lazy_queues_enabled
+ alias_method :lazy_queues_enabled?, :lazy_queues_enabled
+
# In contrast to RabbitMQ 2.x, RabbitMQ 3.x preserves message order when requeing a message. This can lead to
# throughput degradation (when rejected messages block the processing of other messages
# at the head of the queue) in some cases.
@@ -138,6 +143,8 @@ module Beetle
self.dead_lettering_msg_ttl = 1000 #1 second
self.dead_lettering_read_timeout = 3 #3 seconds
+ self.lazy_queues_enabled = false
+
self.publishing_timeout = 0
self.tmpdir = "/tmp"

MODIFY
lib/beetle/dead_lettering.rb
lib/beetle/dead_lettering.rb
@@ -10,54 +10,64 @@ module Beetle
end
def bind_dead_letter_queues!(channel, servers, target_queue, creation_keys = {})
- return unless @config.dead_lettering_enabled?
-
+ if @config.dead_lettering_enabled?
dead_letter_queue_name = dead_letter_queue_name(target_queue)
-
logger.debug("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?
logger.debug("Beetle: setting #{dead_letter_queue_name} as dead letter queue of #{target_queue} on all servers")
- set_dead_letter_policies!(servers, target_queue)
+ end
+ set_queue_policies!(servers, target_queue)
+ if @config.dead_lettering_enabled?
logger.debug("Beetle: setting #{target_queue} as dead letter queue of #{dead_letter_queue_name} on all servers")
- set_dead_letter_policies!(
+ set_queue_policies!(
servers,
dead_letter_queue_name,
:message_ttl => @config.dead_lettering_msg_ttl,
:routing_key => target_queue
)
end
+ end
- def set_dead_letter_policies!(servers, queue_name, options={})
- servers.each { |server| set_dead_letter_policy!(server, queue_name, options) }
+ def set_queue_policies!(servers, queue_name, options={})
+ servers.each { |server| set_queue_policy!(server, queue_name, options) }
end
- def set_dead_letter_policy!(server, queue_name, options={})
+ def set_queue_policy!(server, queue_name, options={})
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?
+
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?
+ 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?
+
request_body = {
"pattern" => "^#{queue_name}$",
"priority" => 1,
"apply-to" => "queues",
- "definition" => {
- "dead-letter-routing-key" => dead_letter_routing_key(queue_name, options),
- "dead-letter-exchange" => ""
+ "definition" => definition,
}
- }
-
- request_body["definition"].merge!("message-ttl" => options[:message_ttl]) if options[:message_ttl]
response = run_rabbit_http_request(request_url, request) do |http|
http.request(request, request_body.to_json)
end
- if response.code != "204"
+ unless %w(200 201 204).include?(response.code)
log_error("Failed to create policy for queue #{queue_name}", response)
raise FailedRabbitRequest.new("Could not create policy")
end

MODIFY
test/beetle/dead_lettering_test.rb
test/beetle/dead_lettering_test.rb
@@ -3,18 +3,20 @@ require File.expand_path(File.dirname(__FILE__) + '/../test_helper')
module Beetle
class SetDeadLetteringsTest < Minitest::Test
def setup
- @dead_lettering = DeadLettering.new(Configuration.new)
+ @config = Configuration.new
+ @dead_lettering = DeadLettering.new(@config)
+ @config.dead_lettering_enabled = true
end
test "creates a dead letter queue for each server" do
servers = %w(a b)
- @dead_lettering.expects(:set_dead_letter_policy!).
+ @dead_lettering.expects(:set_queue_policy!).
with("a", "QUEUE_NAME", :message_ttl => 10000)
- @dead_lettering.expects(:set_dead_letter_policy!).
+ @dead_lettering.expects(:set_queue_policy!).
with("b", "QUEUE_NAME", :message_ttl => 10000)
- @dead_lettering.set_dead_letter_policies!(servers, "QUEUE_NAME", :message_ttl => 10000)
+ @dead_lettering.set_queue_policies!(servers, "QUEUE_NAME", :message_ttl => 10000)
end
end
@@ -25,21 +27,22 @@ module Beetle
@config = Configuration.new
@config.logger = Logger.new("/dev/null")
@dead_lettering = DeadLettering.new(@config)
+ @config.dead_lettering_enabled = true
end
test "raises exception when queue name wasn't specified" do
assert_raises ArgumentError do
- @dead_lettering.set_dead_letter_policy!(@server, "")
+ @dead_lettering.set_queue_policy!(@server, "")
end
end
test "raises exception when no server was specified" do
assert_raises ArgumentError do
- @dead_lettering.set_dead_letter_policy!("", @queue_name)
+ @dead_lettering.set_queue_policy!("", @queue_name)
end
end
- test "creates a policy by posting to the rabbitmq" do
+ test "creates a policy by posting to the rabbitmq if dead lettering is enabled" do
stub_request(:put, "http://localhost:15672/api/policies/%2F/QUEUE_NAME_policy")
.with(basic_auth: ['guest', 'guest'])
.with(:body => {
@@ -52,7 +55,24 @@ module Beetle
}}.to_json)
.to_return(:status => 204)
- @dead_lettering.set_dead_letter_policy!(@server, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name)
+ 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 => {
+ "pattern" => "^QUEUE_NAME$",
+ "priority" => 1,
+ "apply-to" => "queues",
+ "definition" => {
+ "queue-mode" => "lazy"
+ }}.to_json)
+ .to_return(:status => 204)
+
+ @dead_lettering.set_queue_policy!(@server, @queue_name)
end
test "raises exception when policy couldn't successfully be created" do
@@ -61,7 +81,7 @@ module Beetle
.to_return(:status => [405])
assert_raises DeadLettering::FailedRabbitRequest do
- @dead_lettering.set_dead_letter_policy!(@server, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name)
end
end
@@ -79,7 +99,7 @@ module Beetle
}}.to_json)
.to_return(:status => 204)
- @dead_lettering.set_dead_letter_policy!(@server, @queue_name, :message_ttl => 10000)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :message_ttl => 10000)
end
test "properly encodes the vhost from the configuration" do
@@ -97,7 +117,7 @@ module Beetle
@config.vhost = "foo/"
- @dead_lettering.set_dead_letter_policy!(@server, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name)
end
end
@@ -110,9 +130,10 @@ module Beetle
@servers = ["localhost:55672"]
end
- test "is turned off by default" do
+ test "is does not call out to rabbit if neither dead lettering nor lazy queues are enabled" do
+ @config.dead_lettering_enabled = false
channel = stub('channel')
- @dead_lettering.expects(:set_dead_letter_policies!).never
+ @dead_lettering.expects(:run_rabbit_http_request).never
@dead_lettering.bind_dead_letter_queues!(channel, @servers, @queue_name)
end
@@ -122,8 +143,8 @@ module Beetle
channel = stub('channel')
channel.expects(:queue).with("#{@queue_name}_dead_letter", {})
- @dead_lettering.expects(:set_dead_letter_policies!).with(@servers, @queue_name)
- @dead_lettering.expects(:set_dead_letter_policies!).with(@servers, "#{@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_letter",
:routing_key => @queue_name,
:message_ttl => 1000
)

ruby client: support lazy queues