declare queues only once, even with many bindings
declare queues only once, even with many bindings
A queue (and its policies) were declared for every binding that
this queue had, which is not needed. This PR makes subscriber or
publisher declare the queue and policies and lets the base bind
directly.
MODIFY
lib/beetle/base.rb
lib/beetle/base.rb
@@ -67,23 +67,25 @@ module Beetle
logger.debug("Beetle: binding queue #{name} with internal name #{opts[:amqp_name]} on server #{@server}")
queue_name = opts[:amqp_name]
creation_options = opts.slice(*QUEUE_CREATION_KEYS)
- the_queue = nil
+
+ the_queue = declare_queue!(queue_name, creation_options)
@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)
+ logger.debug("Beetle: binding queue #{queue_name} to #{exchange_name} with opts: #{binding_options.inspect}")
+ bind_queue!(the_queue, exchange_name, binding_options)
end
the_queue
end
end
- def bind_dead_letter_queue!(channel, target_queue, creation_keys = {})
+ def bind_dead_letter_queue!(channel, target_queue, creation_options = {})
policy_options = @client.queues[target_queue].slice(:dead_lettering, :lazy, :dead_lettering_msg_ttl)
policy_options[:message_ttl] = policy_options.delete(:dead_lettering_msg_ttl)
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)
+ logger.debug("Beetle: creating dead letter queue #{dead_letter_queue_name} with opts: #{creation_options.inspect}")
+ channel.queue(dead_letter_queue_name, creation_options)
end
return {
:queue_name => target_queue,
@@ -93,7 +95,7 @@ module Beetle
}.merge(policy_options)
end
- # called by <tt>bind_queue!</tt>
+ # called by <tt>declare_queue!</tt>
def publish_policy_options(options)
# avoid endless recursion
return if options[:queue_name] == @client.config.beetle_policy_updates_queue_name

MODIFY
lib/beetle/publisher.rb
lib/beetle/publisher.rb
@@ -247,17 +247,19 @@ module Beetle
@exchanges_with_bound_queues[exchange_name] = true
end
- # TODO: Refactor, fetch the keys and stuff itself
- 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)
- 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)
+ def declare_queue!(queue_name, creation_options)
+ logger.debug("Beetle: creating queue with opts: #{creation_options.inspect}")
+ queue = bunny.queue(queue_name, creation_options)
+
+ policy_options = bind_dead_letter_queue!(bunny, queue_name, creation_options)
publish_policy_options(policy_options)
queue
end
+ def bind_queue!(queue, exchange_name, binding_options)
+ queue.bind(exchange(exchange_name), binding_options)
+ end
+
def stop!(exception=nil)
return unless bunny?
timeout = @client.config.publishing_timeout + @client.config.publisher_connect_timeout + 1

MODIFY
lib/beetle/subscriber.rb
lib/beetle/subscriber.rb
@@ -214,18 +214,20 @@ module Beetle
channel.__send__(opts[:type], name, opts.slice(*EXCHANGE_CREATION_KEYS))
end
- def bind_queue!(queue_name, creation_keys, exchange_name, binding_keys)
- queue = channel.queue(queue_name, creation_keys)
+ def declare_queue!(queue_name, creation_options)
+ queue = channel.queue(queue_name, creation_options)
unless tracing?
# we don't want to create dead-letter queues for tracing
- policy_options = bind_dead_letter_queue!(channel, queue_name, creation_keys)
+ policy_options = bind_dead_letter_queue!(channel, queue_name, creation_options)
publish_policy_options(policy_options)
end
- exchange = exchange(exchange_name)
- queue.bind(exchange, binding_keys)
queue
end
+ def bind_queue!(queue, exchange_name, binding_options)
+ queue.bind(exchange(exchange_name), binding_options)
+ end
+
def connection_settings
{
:host => current_host, :port => current_port, :logging => false,

MODIFY
test/beetle/publisher_test.rb
test/beetle/publisher_test.rb
@@ -304,7 +304,9 @@ module Beetle
@client.register_queue('test_queue_1', :exchange => 'test_exchange')
@client.register_queue('test_queue_2', :exchange => 'test_exchange')
@client.register_queue('test_queue_3', :exchange => 'test_exchange_2')
- @pub.expects(:bind_queue!).returns(1).times(3)
+ queue = mock("queue")
+ queue.expects(:bind).times(3)
+ @pub.expects(:declare_queue!).returns(queue).times(3)
@pub.send(:bind_queues_for_exchange, 'test_exchange')
@pub.send(:bind_queues_for_exchange, 'test_exchange_2')
end
@@ -312,8 +314,19 @@ module Beetle
test "should not rebind the defined queues for the used exchanges if they already have been bound" do
@client.register_queue('test_queue_1', :exchange => 'test_exchange')
@client.register_queue('test_queue_2', :exchange => 'test_exchange')
- @pub.expects(:bind_queue!).twice
+ queue = mock("queue")
+ queue.expects(:bind).twice
+ @pub.expects(:declare_queue!).returns(queue).twice
+ @pub.send(:bind_queues_for_exchange, 'test_exchange')
@pub.send(:bind_queues_for_exchange, 'test_exchange')
+ end
+
+ test "should declare queues only once even with many bindings" do
+ @client.register_queue('test_queue', :exchange => 'test_exchange')
+ @client.register_binding('test_queue', :exchange => 'test_exchange', :key => 'sir-message-a-lot')
+ queue = mock("queue")
+ queue.expects(:bind).twice
+ @pub.expects(:declare_queue!).returns(queue).once
@pub.send(:bind_queues_for_exchange, 'test_exchange')
end

MODIFY
test/beetle/subscriber_test.rb
test/beetle/subscriber_test.rb
@@ -168,6 +168,17 @@ module Beetle
@sub.send(:bind_queues, %W(x y))
end
+ test "binding queues with many bindings should create it only once" do
+ @client.register_queue(:x, :exchange => 'test_exchange')
+ @client.register_binding(:x, :exchange => 'test_exchange', :key => 'sir-message-a-lot')
+ @client.register_handler(%w(x)){}
+ @sub.stubs(:exchange)
+ queue = mock("queue")
+ queue.expects(:bind).twice
+ @sub.expects(:declare_queue!).returns(queue).once
+ @sub.send(:bind_queues, %W(x))
+ end
+
test "subscribing to queues should subscribe on all queues" do
@client.register_queue(:x)
@client.register_queue(:y)

declare queues only once, even with many bindings (#75)

declare queues only once, even with many bindings

A queue (and its policies) were declared for every binding that
this queue had, which is not needed. This PR makes subscriber or
publisher declare the queue and policies and lets the base bind
directly.