asynchronous policy setup
MODIFY
DEAD_LETTERING.md
DEAD_LETTERING.md
# Preserving Rabbit 2.x requeueing behaviour with dead letter queues
## Introduction
-Rabbit 3.x changed the requeueing behaviour when messages are rejected with `reject(:requeue => true)`. In 2.x the requeued messages re-joined the queue at the end of the queue. Since 3.x they preserve their position in the queue. This can lead to throughput degradation due to Head-of-line blocking.
+Rabbit 3.x changed the requeueing behaviour when messages are rejected with
+`reject(:requeue => true)`. In 2.x the requeued messages re-joined the queue at the end of
+the queue. Since 3.x they preserve their position in the queue. This can lead to
+throughput degradation due to Head-of-line blocking.
## Dead Letter Queues
-RabbitMQ provides a [dead letter queue extension](https://www.rabbitmq.com/dlx.html). A queue can have a `dead-letter-exchange` and a `dead-letter-routing` key configured. These settings can be configured either via the queue settings or via a policy.
+RabbitMQ provides a [dead letter queue extension](https://www.rabbitmq.com/dlx.html). A
+queue can have a `dead-letter-exchange` and a `dead-letter-routing` key configured. These
+settings can be configured either via the queue settings or via a policy.
If a queue has those settings, they will be used when
@@ -12,13 +17,43 @@ If a queue has those settings, they will be used when
* the message expired due to a message ttl
* a max queue length was specified and is reached.
-RabbitMQ also knows the concept of the default exchange. It's identified by an empty string. When a message has this exchange configured as `dead-letter-exchange`, the `dead-letter-routing-key` can be used to specify the queue name to which the message should be delivered.
+RabbitMQ also knows the concept of the default exchange. It's identified by an empty
+string. When a message has this exchange configured as `dead-letter-exchange`, the
+`dead-letter-routing-key` can be used to specify the queue name to which the message
+should be delivered.
## How Beetle uses this
-When dead lettering is enabled in Beetle by setting `Beetle.config.dead_lettering_enabled` to `true`, Beetle automatically creates a dead letter queue for each queue and glues them together with RabbitMQ policies. The name of the dead letter queue is the original name suffixed with `_dead_letter`.
-
-The original queue and its dead letter queue have a cyclic configuration. Both have the default exchange configured as the `dead-letter-exchange`. The original queue has `dead-letter-routing-key` set to the name of the dead letter queue and the dead letter queue has `dead-letter-routing-key` set to the name of the original queue.
-
-The only difference is that the dead letter queue has a `message-ttl` configured which specifies how long the message stays in the dead letter queue, before it's republished to the original queue.
-
-Beetle will automatically create policies on the configured Rabbit servers to bind the queues together. This has the advantage that the new behaviour can be enabled on existing queues and the message-ttls can be changed without having to delete queues.
+When dead lettering is enabled in Beetle by setting `Beetle.config.dead_lettering_enabled`
+to `true`, Beetle automatically creates a dead letter queue for each queue and glues them
+together with RabbitMQ policies. The name of the dead letter queue is the original name
+suffixed with `_dead_letter`.
+
+The original queue and its dead letter queue have a cyclic configuration. Both have the
+default exchange configured as the `dead-letter-exchange`. The original queue has
+`dead-letter-routing-key` set to the name of the dead letter queue and the dead letter
+queue has `dead-letter-routing-key` set to the name of the original queue.
+
+The only difference is that the dead letter queue has a `message-ttl` configured which
+specifies how long the message stays in the dead letter queue, before it's republished to
+the original queue.
+
+In order for this to work, you need to run a beetle handler which listens on the beetle
+policy updates queue and calls setup\_queue\_policies! with the parsed JSON payload. For
+example:
+
+```ruby
+client = Beetle::Client.new.configure do |config|
+ config.handler(Beetle.config.beetle_policy_updates_queue_name) do |message|
+ Beetle.config.logger.info "received beetle policy update message': #{message.data}"
+ client.set_queue_policies!(JSON.parse(message.data))
+ end
+end
+```
+
+The policy setup is asynchronous because we learned the hard way that installing queue policies
+from thousands of worker processes can easily bring down a RabbitMQ server.
+
+Beetle will automatically create dead letter queues and policies on the configured
+RabbitMQ servers to bind the queues together. This has the advantage that the new
+behaviour can be enabled on existing queues and the message-ttls can be changed without
+having to delete queues.

MODIFY
features/step_definitions/redis_auto_failover_steps.rb
features/step_definitions/redis_auto_failover_steps.rb
@@ -144,7 +144,7 @@ Then /^the redis master of the beetle handler should be "([^\"]*)"$/ do |redis_n
Beetle.config.servers = "127.0.0.1:5672" # rabbitmq
Beetle.config.logger.level = Logger::INFO
client = Beetle::Client.new.configure :auto_delete => true do |config|
- config.queue(:echo)
+ config.queue(:echo, :lazy => true, :dead_lettering => true)
config.message(:echo)
end
assert_match /#{TestDaemons::Redis[redis_name].ip_with_port}/, client.rpc(:echo, 'nil').second

MODIFY
features/support/beetle_handler
features/support/beetle_handler
require "rubygems"
require "daemons"
require "optparse"
+require 'json'
require File.expand_path("../../lib/beetle", File.dirname(__FILE__))
tmp_path = File.expand_path("../../tmp", File.dirname(__FILE__))
@@ -22,7 +23,7 @@ Daemons.run_proc("beetle_handler", :log_output => true, :dir_mode => :normal, :d
Beetle.config.logger.level = Logger::INFO
client = Beetle::Client.new.configure :auto_delete => true do |config|
- config.queue(:echo)
+ config.queue(:echo, :lazy => true, :dead_lettering => true)
config.message(:echo)
config.handler(:echo) do |message|
begin
@@ -32,6 +33,14 @@ Daemons.run_proc("beetle_handler", :log_output => true, :dir_mode => :normal, :d
"no redis master: exception: #{$!.class}(#{$!}), master_file: '#{master_file_content}'"
end
end
+ config.handler(Beetle.config.beetle_policy_updates_queue_name) do |message|
+ begin
+ Beetle.config.logger.info "received beetle policy update message': #{message.data}"
+ client.set_queue_policies!(JSON.parse(message.data))
+ rescue => e
+ Beetle.config.logger.error("#{e}:#{e.backtrace.join("\n")}")
+ end
+ end
end
client.listen do
puts "Started beetle handler for system: #{Beetle.config.system_name}"

MODIFY
lib/beetle/base.rb
lib/beetle/base.rb
+require 'json'
+
module Beetle
# Abstract base class shared by Publisher and Subscriber
class Base
@@ -12,7 +14,6 @@ module Beetle
@server = @servers[rand @servers.size]
@exchanges = {}
@queues = {}
- @dead_lettering = DeadLettering.new(@client)
end
private
@@ -54,13 +55,7 @@ module Beetle
@queues[@server] ||= {}
end
- QueueInfo = Struct.new(:queue, :create_policies)
-
- def queue(name, create_policies: false)
- info = queues[name]
- if info && create_policies && !info.create_policies
- queues.delete(name)
- end
+ def queue(name)
queues[name] ||=
begin
opts = @client.queues[name]
@@ -72,11 +67,37 @@ module Beetle
@client.bindings[name].each do |binding_options|
exchange_name = binding_options[:exchange]
binding_options = binding_options.slice(*QUEUE_BINDING_KEYS)
- the_queue = bind_queue!(queue_name, creation_options, exchange_name, binding_options, create_policies: create_policies)
+ the_queue = bind_queue!(queue_name, creation_options, exchange_name, binding_options)
+ end
+ the_queue
end
- info = QueueInfo.new(the_queue, create_policies)
end
- info.queue
+
+ def bind_dead_letter_queue!(channel, target_queue, creation_keys = {})
+ policy_options = @client.queues[target_queue].slice(:dead_lettering, :lazy)
+ dead_letter_queue_name = "#{target_queue}_dead_letter"
+ if policy_options[:dead_lettering]
+ 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
+ return {
+ :queue_name => target_queue,
+ :dead_letter_queue_name => dead_letter_queue_name,
+ :message_ttl => @client.config.dead_lettering_msg_ttl,
+ }.merge(policy_options)
+ end
+
+ # called by <tt>bind_queue!</tt>
+ def publish_policy_options(options)
+ # avoid endless recursion
+ return if options[:queue_name] == @client.config.beetle_policy_updates_queue_name
+ payload = options.merge(:server => @server)
+ logger.debug("Beetle: publishing policy options on #{@server}: #{payload.inspect}")
+ # make sure to declare the queue, so the message does not get lost
+ queue(@client.config.beetle_policy_updates_queue_name)
+ data = payload.to_json
+ opts = Message.publishing_options(:key => @client.config.beetle_policy_updates_routing_key, :persistent => true, :redundant => false)
+ exchange(@client.config.beetle_policy_exchange_name).publish(data, opts)
end
end

MODIFY
lib/beetle/client.rb
lib/beetle/client.rb
+require 'json'
+
module Beetle
# This class provides the interface through which messaging is configured for both
# message producers and consumers. It keeps references to an instance of a
@@ -53,7 +55,17 @@ module Beetle
@messages = {}
@bindings = {}
@deduplication_store = DeduplicationStore.new(config)
+ @dead_lettering = DeadLettering.new(config)
load_brokers_from_config
+ register_exchange(config.beetle_policy_exchange_name)
+ # make sure dead lettering is false for the policy update queue
+ register_queue(
+ config.beetle_policy_updates_queue_name,
+ :exchange => config.beetle_policy_exchange_name,
+ :key => config.beetle_policy_updates_routing_key,
+ :dead_lettering => false,
+ :lazy => false,
+ )
end
# register an exchange with the given _name_ and a set of _options_:
@@ -209,11 +221,6 @@ module Beetle
publisher.purge(queues)
end
- # declares all queues, binds them and creates/updates all policies
- def setup_queues_and_policies(queues)
- publisher.setup_queues_and_policies(queues)
- end
-
# start listening to all registered queues. Calls #listen_queues internally
# runs the given block before entering the eventmachine loop.
def listen(_deprecated_messages=nil, &block)
@@ -292,6 +299,10 @@ module Beetle
@subscriber = nil
end
+ def set_queue_policies!(message_payload)
+ @dead_lettering.set_queue_policies!(message_payload)
+ end
+
private
def determine_queue_names(queues)

MODIFY
lib/beetle/configuration.rb
lib/beetle/configuration.rb
@@ -5,6 +5,14 @@ module Beetle
class Configuration
# system name (used for redis cluster partitioning) (defaults to <tt>system</tt>)
attr_accessor :system_name
+ # system_exchange_name is the name of the exchange on which to publish system internal messages, such as
+ # messages to set up queue policies. Whenever a queue is declared in the client, either by publisher or
+ # consumer, a message with the queue policy parameters is sent to this exachange. (defaults to <tt>beetle</tt>)
+ attr_accessor :beetle_policy_exchange_name
+ # Name of the policy update queue
+ attr_accessor :beetle_policy_updates_queue_name
+ # 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 :logger
# defaults to <tt>STDOUT</tt>
@@ -120,6 +128,9 @@ module Beetle
def initialize #:nodoc:
self.system_name = "system"
+ 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.gc_threshold = 1.hour.to_i
self.redis_server = "localhost:6379"

MODIFY
lib/beetle/dead_lettering.rb
lib/beetle/dead_lettering.rb
@@ -5,56 +5,44 @@ module Beetle
class DeadLettering
class FailedRabbitRequest < StandardError; end
- attr_reader :client, :config
+ attr_reader :config
- def initialize(client)
- @client = client
- @config = client.config
+ def initialize(config)
+ @config = config
end
- def bind_dead_letter_queues!(channel, servers, target_queue, creation_keys = {})
- 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
+ def set_queue_policies!(options)
+ # logger.debug "Setting queue policies: #{options.inspect}"
+ options = options.symbolize_keys
+ server = options[:server]
+ target_queue = options[:queue_name]
+ dead_letter_queue_name = options[:dead_letter_queue_name]
+ policy_options = options.slice(:lazy, :dead_lettering)
- 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, policy_options)
-
- 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,
- :routing_key => target_queue
- }.merge(policy_options)
- )
- end
- end
+ target_queue_options = policy_options.merge(:routing_key => dead_letter_queue_name)
+ set_queue_policy!(server, target_queue, target_queue_options)
- def set_queue_policies!(servers, queue_name, options={})
- servers.each { |server| set_queue_policy!(server, queue_name, options) }
+ dead_letter_queue_options = policy_options.merge(:routing_key => target_queue, :message_ttl => options[:message_ttl])
+ set_queue_policy!(server, dead_letter_queue_name, dead_letter_queue_options)
end
def set_queue_policy!(server, queue_name, options={})
+ logger.info "Setting queue policy: #{server}, #{queue_name}, #{options.inspect}"
+
raise ArgumentError.new("server missing") if server.blank?
raise ArgumentError.new("queue name missing") if queue_name.blank?
return unless options[:dead_lettering] || options[:lazy]
vhost = CGI.escape(config.vhost)
+ # 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")
request = Net::HTTP::Put.new(request_url)
# set up queue policy
definition = {}
if options[:dead_lettering]
- definition["dead-letter-routing-key"] = dead_letter_routing_key(queue_name, options)
+ definition["dead-letter-routing-key"] = options[:routing_key]
definition["dead-letter-exchange"] = ""
definition["message-ttl"] = options[:message_ttl] if options[:message_ttl]
end
@@ -80,19 +68,15 @@ module Beetle
:ok
end
- def dead_letter_routing_key(queue_name, options)
- options.fetch(:routing_key) { dead_letter_queue_name(queue_name) }
- end
-
- def dead_letter_queue_name(queue_name)
- "#{queue_name}_dead_letter"
- end
-
def run_rabbit_http_request(uri, request, &block)
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|
- block.call(http) if block_given?
+ http = Net::HTTP.new(uri.hostname, config.api_port)
+ http.read_timeout = config.dead_lettering_read_timeout
+ # don't do this in production:
+ # http.set_debug_output(logger.instance_eval{ @logdev.dev })
+ http.start do |instance|
+ block.call(instance) if block_given?
end
end
@@ -103,7 +87,8 @@ module Beetle
end
def logger
- config.logger
+ @config.logger
end
+
end
end

MODIFY
lib/beetle/publisher.rb
lib/beetle/publisher.rb
@@ -222,14 +222,13 @@ module Beetle
end
# TODO: Refactor, fetch the keys and stuff itself
- def bind_queue!(queue_name, creation_keys, exchange_name, binding_keys, create_policies: false)
+ def bind_queue!(queue_name, creation_keys, exchange_name, binding_keys)
logger.debug("Beetle: creating queue with opts: #{creation_keys.inspect}")
queue = bunny.queue(queue_name, creation_keys)
- if create_policies
- @dead_lettering.bind_dead_letter_queues!(bunny, @client.servers, queue_name, creation_keys)
- end
logger.debug("Beetle: binding queue #{queue_name} to #{exchange_name} with opts: #{binding_keys.inspect}")
queue.bind(exchange(exchange_name), binding_keys)
+ policy_options = bind_dead_letter_queue!(bunny, queue_name, creation_keys)
+ publish_policy_options(policy_options)
queue
end

MODIFY
lib/beetle/subscriber.rb
lib/beetle/subscriber.rb
@@ -144,19 +144,19 @@ module Beetle
callback = create_subscription_callback(queue_name, amqp_queue_name, handler, opts)
keys = opts.slice(*SUBSCRIPTION_KEYS).merge(:key => "#", :ack => true)
logger.debug "Beetle: subscribing to queue #{amqp_queue_name} with key # on server #{@server}"
- queues[queue_name].queue.subscribe(keys, &callback)
+ queues[queue_name].subscribe(keys, &callback)
subscriptions[queue_name] = [keys, callback]
end
def pause(queue_name)
- return unless queues[queue_name].queue.subscribed?
- queues[queue_name].queue.unsubscribe
+ return unless queues[queue_name].subscribed?
+ queues[queue_name].unsubscribe
end
def resume(queue_name)
- return if queues[queue_name].queue.subscribed?
+ return if queues[queue_name].subscribed?
keys, callback = subscriptions[queue_name]
- queues[queue_name].queue.subscribe(keys, &callback)
+ queues[queue_name].subscribe(keys, &callback)
end
def create_subscription_callback(queue_name, amqp_queue_name, handler, opts)
@@ -204,12 +204,10 @@ module Beetle
channel.__send__(opts[:type], name, opts.slice(*EXCHANGE_CREATION_KEYS))
end
- def bind_queue!(queue_name, creation_keys, exchange_name, binding_keys, create_policies: false)
+ def bind_queue!(queue_name, creation_keys, exchange_name, binding_keys)
queue = channel.queue(queue_name, creation_keys)
- target_servers = @client.servers + @client.additional_subscription_servers
- if create_policies
- @dead_lettering.bind_dead_letter_queues!(channel, target_servers, queue_name, creation_keys)
- end
+ policy_options = bind_dead_letter_queue!(channel, queue_name, creation_keys)
+ publish_policy_options(policy_options)
exchange = exchange(exchange_name)
queue.bind(exchange, binding_keys)
queue

MODIFY
test/beetle/base_test.rb
test/beetle/base_test.rb
@@ -53,4 +53,54 @@ module Beetle
assert_equal "goofy:123", @bs.send(:server_from_settings, {:host => "goofy", :port => 123})
end
end
+
+ class BindDeadLetterQueuesTest < Minitest::Test
+ def setup
+ @queue_name = "QUEUE_NAME"
+ @config = Configuration.new
+ @client = Client.new @config
+ @bs = Base.new(@client)
+ @config.logger = Logger.new("/dev/null")
+ end
+
+ test "it does not call out to rabbit if neither dead lettering nor lazy queues are enabled" do
+ @client.register_queue(@queue_name, :dead_lettering => false, :lazy => false)
+ channel = stub('channel')
+ expected_options = {
+ :queue_name => "QUEUE_NAME",
+ :dead_letter_queue_name=>"QUEUE_NAME_dead_letter",
+ :message_ttl => 1000,
+ :dead_lettering => false,
+ :lazy => false
+ }
+ assert_equal expected_options, @bs.__send__(:bind_dead_letter_queue!, channel, @queue_name)
+ end
+
+ test "creates and connects the dead letter queue via policies when enabled" do
+ @client.register_queue(@queue_name, :dead_lettering => true, :lazy => true)
+
+ channel = stub('channel')
+ channel.expects(:queue).with("#{@queue_name}_dead_letter", {})
+
+ expected_options = {
+ :queue_name => "QUEUE_NAME",
+ :dead_letter_queue_name=>"QUEUE_NAME_dead_letter",
+ :message_ttl => 1000,
+ :dead_lettering => true,
+ :lazy => true
+ }
+ assert_equal expected_options, @bs.__send__(:bind_dead_letter_queue!, channel, @queue_name)
+ end
+
+ test "publish_policy_options declares the beetle policy updates queue and publishes the options" do
+ options = { :lazy => true, :dead_lettering => true }
+ @bs.logger.stubs(:debug)
+ @bs.expects(:queue).with(@client.config.beetle_policy_updates_queue_name)
+ exchange = mock("exchange")
+ exchange.expects(:publish)
+ @bs.expects(:exchange).with(@client.config.beetle_policy_exchange_name).returns(exchange)
+ @bs.__send__(:publish_policy_options, options)
+ end
+
+ end
end

MODIFY
test/beetle/client_test.rb
test/beetle/client_test.rb
@@ -15,20 +15,20 @@ module Beetle
assert_equal [], @client.additional_subscription_servers
end
- test "should have no exchanges" do
- assert @client.exchanges.empty?
+ test "should have only the system exchange" do
+ assert_equal ["beetle-policies"], @client.exchanges.keys
end
- test "should have no queues" do
- assert @client.queues.empty?
+ test "should have only the beetle policy queue" do
+ assert_equal ["beetle-policy-updates"], @client.queues.keys
end
test "should have no messages" do
- assert @client.messages.empty?
+ assert_equal [], @client.messages.keys
end
- test "should have no bindings" do
- assert @client.bindings.empty?
+ test "should have only one binding for the beetle policy updates" do
+ assert_equal ["beetle-policy-updates"], @client.bindings.keys
end
end
@@ -308,13 +308,6 @@ module Beetle
assert_equal "ha!", client.purge(:queue1, :queue2)
end
- test "should delegate queue setup to the publisher instance" do
- client = Client.new
- client.register_queue(:queue)
- client.send(:publisher).expects(:setup_queues_and_policies).with(["queue"]).returns("ha!")
- assert_equal "ha!", client.setup_queues_and_policies(["queue"])
- end
-
test "should delegate rpc calls to the publisher instance" do
client = Client.new
client.register_message("deadletter")
@@ -363,6 +356,12 @@ module Beetle
client.stop_listening
end
+ test "should delegate set_queue_policies! to the dead lettering instance" do
+ client = Client.new
+ client.instance_variable_get(:@dead_lettering).expects(:set_queue_policies!)
+ client.set_queue_policies!({})
+ end
+
test "should delegate pause_listening to the subscriber instance" do
client = Client.new
client.register_queue(:test)

MODIFY
test/beetle/dead_lettering_test.rb
test/beetle/dead_lettering_test.rb
require File.expand_path(File.dirname(__FILE__) + '/../test_helper')
module Beetle
- class SetDeadLetteringsTest < Minitest::Test
- def setup
- @config = Configuration.new
- @client = Client.new @config
- @dead_lettering = DeadLettering.new(@client)
- @config.dead_lettering_enabled = true
- end
-
- test "creates a dead letter queue for each server" do
- servers = %w(a b)
-
- @dead_lettering.expects(:set_queue_policy!).
- with("a", "QUEUE_NAME", :message_ttl => 10000)
- @dead_lettering.expects(:set_queue_policy!).
- with("b", "QUEUE_NAME", :message_ttl => 10000)
-
- @dead_lettering.set_queue_policies!(servers, "QUEUE_NAME", :message_ttl => 10000)
- end
- end
-
class SetDeadLetterPolicyTest < Minitest::Test
def setup
@server = "localhost:15672"
@queue_name = "QUEUE_NAME"
@config = Configuration.new
- @client = Client.new @config
@config.logger = Logger.new("/dev/null")
- @dead_lettering = DeadLettering.new(@client)
+ @dead_lettering = DeadLettering.new(@config)
end
test "raises exception when queue name wasn't specified" do
@@ -43,6 +22,22 @@ module Beetle
end
end
+ test "set_queue_policies! calls set_queue_policy for both target queue and dead letter queue" do
+ options = {
+ :server => "server", :lazy => true, :dead_lettering => true,
+ :queue_name => "QUEUE_NAME", :dead_letter_queue_name => "QUEUE_NAME_dead_letter",
+ :message_ttl => 10000
+ }
+ @dead_lettering.expects(:set_queue_policy!).with("server", "QUEUE_NAME",
+ :lazy => true, :dead_lettering => true,
+ :routing_key => "QUEUE_NAME_dead_letter")
+ @dead_lettering.expects(:set_queue_policy!).with("server", "QUEUE_NAME_dead_letter",
+ :lazy => true, :dead_lettering => true,
+ :routing_key => "QUEUE_NAME",
+ :message_ttl => 10000)
+ @dead_lettering.set_queue_policies!(options)
+ end
+
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'])
@@ -56,7 +51,7 @@ module Beetle
}}.to_json)
.to_return(:status => 204)
- @dead_lettering.set_queue_policy!(@server, @queue_name, :lazy => false, :dead_lettering => true)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :lazy => false, :dead_lettering => true, :routing_key => "QUEUE_NAME_dead_letter")
end
test "creates a policy by posting to the rabbitmq if lazy queues are enabled" do
@@ -98,7 +93,7 @@ module Beetle
}}.to_json)
.to_return(:status => 204)
- @dead_lettering.set_queue_policy!(@server, @queue_name, :dead_lettering => true, :message_ttl => 10000)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :dead_lettering => true, :message_ttl => 10000, :routing_key => "QUEUE_NAME_dead_letter")
end
test "properly encodes the vhost from the configuration" do
@@ -116,42 +111,7 @@ module Beetle
@config.vhost = "foo/"
- @dead_lettering.set_queue_policy!(@server, @queue_name, :dead_lettering => true)
- end
- end
-
- class BindDeadLetterQueuesTest < Minitest::Test
- def setup
- @queue_name = "QUEUE_NAME"
- @config = Configuration.new
- @client = Client.new @config
- @config.logger = Logger.new("/dev/null")
- @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
- @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
- @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 => true, :lazy => false)
- @dead_lettering.expects(:set_queue_policies!).with(@servers, "#{@queue_name}_dead_letter",
- :routing_key => @queue_name,
- :message_ttl => 1000,
- :dead_lettering => true,
- :lazy => false
- )
-
- @dead_lettering.bind_dead_letter_queues!(channel, @servers, @queue_name)
+ @dead_lettering.set_queue_policy!(@server, @queue_name, :dead_lettering => true, :routing_key => "QUEUE_NAME_dead_letter")
end
end
end

MODIFY
test/beetle/publisher_test.rb
test/beetle/publisher_test.rb
@@ -263,7 +263,8 @@ module Beetle
class PublisherQueueManagementTest < Minitest::Test
def setup
- @client = Client.new
+ @config = Configuration.new
+ @client = Client.new(@config)
@pub = Publisher.new(@client)
end
@@ -275,48 +276,15 @@ module Beetle
test "binding a queue should create it using the config and bind it to the exchange with the name specified" do
@client.register_queue("some_queue", :exchange => "some_exchange", :key => "haha.#", :arguments => {"foo" => "fighter"})
@pub.expects(:exchange).with("some_exchange").returns(:the_exchange)
- q = mock("queue")
- q.expects(:bind).with(:the_exchange, {:key => "haha.#"})
- m = mock("Bunny")
- m.expects(:queue).with("some_queue", :durable => true, :passive => false, :auto_delete => false, :exclusive => false, :arguments => {"foo" => "fighter"}).returns(q)
- @pub.expects(:bunny).returns(m)
-
- @pub.send(:queue, "some_queue")
- assert_equal Base::QueueInfo.new(q, false), @pub.send(:queues)["some_queue"]
- end
-
- test "binding a queue should create it using the config and bind it to the exchange with the name specified and setup policies if requested" do
- @client.register_queue("some_queue", :exchange => "some_exchange", :key => "haha.#", :arguments => {"foo" => "fighter"})
- @pub.expects(:exchange).with("some_exchange").returns(:the_exchange)
+ @pub.expects(:publish_policy_options)
q = mock("queue")
q.expects(:bind).with(:the_exchange, {:key => "haha.#"})
m = mock("Bunny")
m.expects(:queue).with("some_queue", :durable => true, :passive => false, :auto_delete => false, :exclusive => false, :arguments => {"foo" => "fighter"}).returns(q)
@pub.expects(:bunny).returns(m).twice
- @pub.send(:queue, "some_queue", create_policies: true)
- assert_equal Base::QueueInfo.new(q, true), @pub.send(:queues)["some_queue"]
- end
-
- test "binding a queue with create_plocies: true after having already declared it with create_plocies: false creates the policies" do
- @client.register_queue("some_queue", :exchange => "some_exchange", :key => "haha.#", :arguments => {"foo" => "fighter"})
- @pub.expects(:exchange).with("some_exchange").returns(:the_exchange)
- q = mock("queue")
- q.expects(:bind).with(:the_exchange, {:key => "haha.#"})
- m = mock("Bunny")
- m.expects(:queue).with("some_queue", :durable => true, :passive => false, :auto_delete => false, :exclusive => false, :arguments => {"foo" => "fighter"}).returns(q)
- @pub.expects(:bunny).returns(m)
-
@pub.send(:queue, "some_queue")
- assert_equal Base::QueueInfo.new(q, false), @pub.send(:queues)["some_queue"]
-
- @pub.expects(:exchange).with("some_exchange").returns(:the_exchange)
- q.expects(:bind).with(:the_exchange, {:key => "haha.#"})
- m.expects(:queue).with("some_queue", :durable => true, :passive => false, :auto_delete => false, :exclusive => false, :arguments => {"foo" => "fighter"}).returns(q)
- @pub.expects(:bunny).returns(m).twice
-
- @pub.send(:queue, "some_queue", create_policies: true)
- assert_equal Base::QueueInfo.new(q, true), @pub.send(:queues)["some_queue"]
+ assert_equal q, @pub.send(:queues)["some_queue"]
end
test "should bind the defined queues for the used exchanges when publishing" do

MODIFY
test/beetle/subscriber_test.rb
test/beetle/subscriber_test.rb
@@ -89,7 +89,7 @@ module Beetle
q = mock("queue a")
q.expects(:subscribed?).returns(true)
q.expects(:unsubscribe)
- @sub.stubs(:queues).returns({"a" => Base::QueueInfo.new(q,false)})
+ @sub.stubs(:queues).returns("a" => q)
@sub.__send__(:pause, "a")
end
@@ -97,7 +97,7 @@ module Beetle
q = mock("queue a")
q.expects(:subscribed?).returns(false)
q.expects(:unsubscribe).never
- @sub.stubs(:queues).returns({"a" => Base::QueueInfo.new(q,false)})
+ @sub.stubs(:queues).returns("a" => q)
@sub.__send__(:pause, "a")
end
@@ -105,7 +105,7 @@ module Beetle
q = mock("queue a")
q.expects(:subscribed?).returns(false)
q.expects(:subscribe)
- @sub.stubs(:queues).returns({"a" => Base::QueueInfo.new(q,false)})
+ @sub.stubs(:queues).returns("a" => q)
@sub.__send__(:resume, "a")
end
@@ -113,7 +113,7 @@ module Beetle
q = mock("queue a")
q.expects(:subscribed?).returns(true)
q.expects(:subscribe).never
- @sub.stubs(:queues).returns({"a" => Base::QueueInfo.new(q,false)})
+ @sub.stubs(:queues).returns("a" => q)
@sub.__send__(:resume, "a")
end
@@ -146,27 +146,17 @@ module Beetle
test "binding a queue should create it using the config and bind it to the exchange with the name specified" do
@client.register_queue("some_queue", "durable" => true, "exchange" => "some_exchange", "key" => "haha.#", "arguments" => {"schmu" => 5})
@sub.expects(:exchange).with("some_exchange").returns(:the_exchange)
+ @sub.expects(:publish_policy_options)
q = mock("queue")
q.expects(:bind).with(:the_exchange, {:key => "haha.#"})
- m = mock("MQ")
- m.expects(:queue).with("some_queue", :durable => true, :passive => false, :auto_delete => false, :exclusive => false, :arguments => {"schmu" => 5}).returns(q)
- @sub.expects(:channel).returns(m)
+ channel = mock("MQ")
+ creation_opts = {:durable => true, :passive => false, :auto_delete => false, :exclusive => false, :arguments => {"schmu" => 5}}
+ channel.expects(:queue).with("some_queue", creation_opts).returns(q)
+ @sub.expects(:channel).returns(channel).twice
+ @sub.expects(:bind_dead_letter_queue!).with(channel, "some_queue", creation_opts)
@sub.send(:queue, "some_queue")
- assert_equal q, @sub.send(:queues)["some_queue"].queue
- end
-
- test "binding a queue should create it using the config and bind it to the exchange with the name specified and create policies if requested" do
- @client.register_queue("some_queue", "durable" => true, "exchange" => "some_exchange", "key" => "haha.#", "arguments" => {"schmu" => 5})
- @sub.expects(:exchange).with("some_exchange").returns(:the_exchange)
- q = mock("queue")
- q.expects(:bind).with(:the_exchange, {:key => "haha.#"})
- m = mock("MQ")
- m.expects(:queue).with("some_queue", :durable => true, :passive => false, :auto_delete => false, :exclusive => false, :arguments => {"schmu" => 5}).returns(q)
- @sub.expects(:channel).returns(m).twice
-
- @sub.send(:queue, "some_queue", create_policies: true)
- assert_equal q, @sub.send(:queues)["some_queue"].queue
+ assert_equal q, @sub.send(:queues)["some_queue"]
end
test "binding queues should bind all queues" do
@@ -370,7 +360,7 @@ module Beetle
q = mock("QUEUE")
subscription_options = {:ack => true, :key => "#"}
q.expects(:subscribe).with(subscription_options).yields(header, "foo")
- @sub.expects(:queues).returns({"some_queue" => Base::QueueInfo.new(q,false)}).once
+ @sub.expects(:queues).returns("some_queue" => q).once
@sub.send(:subscribe, "some_queue")
assert block_called
assert @sub.__send__(:has_subscription?, "some_queue")

asynchronous policy setup