expose publisher method to setup queues/policies ahead of use
expose publisher method to setup queues/policies ahead of use
Any encountered error will be turned into a warning instead. In the
worst case this means an application can still boot and retry the
setup on use later on.
MODIFY
lib/beetle/client.rb
lib/beetle/client.rb
@@ -280,6 +280,13 @@ module Beetle
publisher.throttled?
end
+ # set up queues and policies for all configured queues. Otherwise this will
+ # happen on first use of an exchange, which can be undesired for latency
+ # sensitive endpoints. Only needs to be called once.
+ def setup_queues_and_policies
+ publisher.setup_queues_and_policies
+ end
+
# traces queues without consuming them. useful for debugging message flow.
def trace(queue_names=self.queues.keys, tracer=nil, &block)
queues_to_trace = self.queues.slice(*queue_names)

MODIFY
lib/beetle/publisher.rb
lib/beetle/publisher.rb
@@ -169,10 +169,14 @@ module Beetle
end
end
- def setup_queues_and_policies(queue_names) #:nodoc:
+ def setup_queues_and_policies
each_server do
- queue_names.each do |name|
- queue(name, create_policies: true)
+ begin
+ @client.queues.keys.each do |name|
+ queue(name)
+ end
+ rescue => e
+ logger.warn "Beetle: failed setting up queues and policies on #{@server}: #{e}"
end
end
end

MODIFY
test/beetle/client_test.rb
test/beetle/client_test.rb
@@ -282,6 +282,12 @@ module Beetle
client.stop_publishing
end
+ test "should delegate setup_queues_and_policies to the publisher instance" do
+ client = Client.new
+ client.send(:publisher).expects(:setup_queues_and_policies)
+ client.setup_queues_and_policies
+ end
+
test "should delegate queue purging to the publisher instance" do
client = Client.new
client.register_queue(:queue)

MODIFY
test/beetle/publisher_test.rb
test/beetle/publisher_test.rb
@@ -355,14 +355,34 @@ module Beetle
end
test "setting up queues and policies should iterate over all servers" do
- @pub.servers = %w(a b)
- queue = mock("queue")
+ client = Client.new
+ client.register_queue("queue")
+ pub = Publisher.new(client)
+ pub.servers = %w(a b)
+
s = sequence("setup")
- @pub.expects(:set_current_server).with("a").in_sequence(s)
- @pub.expects(:queue).with("queue", :create_policies => true).returns(queue).in_sequence(s)
- @pub.expects(:set_current_server).with("b").in_sequence(s)
- @pub.expects(:queue).with("queue", :create_policies => true).returns(queue).in_sequence(s)
- @pub.setup_queues_and_policies(["queue"])
+ pub.expects(:set_current_server).with("a").in_sequence(s)
+ pub.expects(:queue).with(client.config.beetle_policy_updates_queue_name).in_sequence(s)
+ pub.expects(:queue).with("queue").in_sequence(s)
+ pub.expects(:set_current_server).with("b").in_sequence(s)
+ pub.expects(:queue).with(client.config.beetle_policy_updates_queue_name).in_sequence(s)
+ pub.expects(:queue).with("queue").in_sequence(s)
+
+ pub.setup_queues_and_policies()
+ end
+
+ test "setting up queues and policies should handle ephemeral errors" do
+ client = Client.new
+ pub = Publisher.new(client)
+ client.register_queue("queue")
+ pub.servers = %w(a b)
+ pub.stubs(:queue).raises(StandardError)
+
+ s = sequence("setup")
+ pub.expects(:set_current_server).with("a").in_sequence(s)
+ pub.expects(:set_current_server).with("b").in_sequence(s)
+
+ pub.setup_queues_and_policies()
end
test "reports whether it has been throttled" do

expose publisher method to setup queues/policies ahead of use (#76)

expose publisher method to setup queues/policies ahead of use

Any encountered error will be turned into a warning instead. In the
worst case this means an application can still boot and retry the
setup on use later on.