implement basic throttling functionality
ADD
examples/throttling.rb
+# simple.rb
+# this example shows you a very basic message/handler setup
+#
+# ! check the examples/README.rdoc for information on starting your redis/rabbit !
+#
+# start it with ruby simple.rb
+
+require "rubygems"
+require File.expand_path("../lib/beetle", File.dirname(__FILE__))
+
+# set Beetle log level to info, less noisy than debug
+Beetle.config.logger.level = Logger::INFO
+# override default of 60 seconds
+Beetle.config.throttling_refresh_interval = 5
+
+# setup clients
+client = Beetle::Client.new
+client.register_queue(:test)
+client.register_message(:test)
+client.throttle(:test => 50)
+
+consumer = Beetle::Client.new
+consumer.register_queue(:test)
+consumer.register_message(:test)
+
+# purge the test queue
+client.purge(:test)
+
+# empty the dedup store
+client.deduplication_store.flushdb
+
+# register our handler to the message, check out the message.rb for more stuff you can get from the message object
+messages_received = 0
+consumer.register_handler(:test) do |message|
+ sleep 0.2
+ messages_received += 1
+ puts "Received #{messages_received} messages"
+end
+
+interrupted = false
+
+t = Thread.new do
+ c = 0
+ loop do
+ break if c == 200 || interrupted
+ 100.times do |i|
+ client.publish(:test, (c+=1).to_s)
+ puts "Published message #{c}"
+ sleep 0.1
+ end
+ end
+end
+
+trap('INT') do
+ interrupted = true
+ consumer.stop_listening
+end
+
+# start listening
+# this starts the event machine event loop using EM.run
+# the block passed to listen will be yielded as the last step of the setup process
+consumer.listen do
+ EM.add_periodic_timer(1) do
+ consumer.stop_listening if messages_received >= 200
+ end
+end
+
+t.join

MODIFY
lib/beetle/client.rb
lib/beetle/client.rb
@@ -258,6 +258,25 @@ module Beetle
subscriber.resume_listening(queues)
end
+ # throttle publishing of messages based on queue lengths.
+ #
+ # client.throttle(:foo => 10000, :bar => 10_000)
+ #
+ # throttle publisher to 1 message per second as long as queue foo has more
+ # than 10000 entries or queue bar has more than 100000 entries across all
+ # servers. Queue lenghts are periodically determined during publishing. You
+ # only want to use this feature if you plan to publish a huge amount of
+ # messages to slow consumers so as to not overload the broker or the redis
+ # dedpuplication store.
+ def throttle(throttling_options)
+ publisher.throttle(throttling_options.stringify_keys)
+ end
+
+ # is the publisher currently throttled?
+ def throttled?
+ publisher.throttled?
+ 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/configuration.rb
lib/beetle/configuration.rb
@@ -112,6 +112,9 @@ module Beetle
# decreased parallelism.
attr_accessor :prefetch_count
+ # refresh interval for determining queue lenghts for throttling.
+ attr_accessor :throttling_refresh_interval
+
# directory to store large intermediate files (defaults '/tmp')
attr_accessor :tmpdir
@@ -160,6 +163,7 @@ module Beetle
self.dead_lettering_read_timeout = 3 #3 seconds
self.lazy_queues_enabled = false
+ self.throttling_refresh_interval = 60 # seconds
self.publishing_timeout = 0
self.publisher_connect_timeout = 5

MODIFY
lib/beetle/publisher.rb
lib/beetle/publisher.rb
@@ -9,9 +9,24 @@ module Beetle
@exchanges_with_bound_queues = {}
@dead_servers = {}
@bunnies = {}
+ @throttling_options = {}
+ @next_throttle_refresh = Time.now
+ @throttled = false
at_exit { stop }
end
+ def throttled?
+ @throttled
+ end
+
+ def throttling?
+ !@throttling_options.empty?
+ end
+
+ def throttling_status
+ @throttled ? 'throttled' : 'unthrottled'
+ end
+
# list of exceptions potentially raised by bunny
# these need to be lazy, because qrack exceptions are only defined after a connection has been established
def bunny_exceptions
@@ -29,6 +44,7 @@ module Beetle
exchange_name = opts.delete(:exchange)
opts.delete(:queue)
recycle_dead_servers unless @dead_servers.empty?
+ throttle!
if opts[:redundant]
publish_with_redundancy(exchange_name, message_name, data, opts)
else
@@ -135,6 +151,16 @@ module Beetle
[status, result]
end
+ def throttle(queue_options)
+ @throttling_options = queue_options
+ end
+
+ def throttle!
+ return unless throttling?
+ refresh_throttling!
+ sleep 1 if throttled?
+ end
+
def purge(queue_names) #:nodoc:
each_server do
queue_names.each do |name|
@@ -251,5 +277,30 @@ module Beetle
@exchanges[@server] = {}
@queues[@server] = {}
end
+
+ def refresh_throttling!
+ t = Time.now
+ return if t < @next_throttle_refresh
+ @next_throttle_refresh = t + @client.config.throttling_refresh_interval
+ old_throttled = @throttled
+ @throttled = false
+ @throttling_options.each do |queue_name, max_length|
+ begin
+ len = 0
+ each_server do
+ len += queue(queue_name).status[:message_count]
+ end
+ # logger.debug "Beetle: queue '#{queue_name}' has size #{len}"
+ if len > max_length
+ @throttled = true
+ break
+ end
+ rescue => e
+ logger.warn "Beetle: could not fetch queue length for queue '#{queue_name}': #{e}"
+ end
+ end
+ logger.info "Beetle: publisher #{throttling_status}" if @throttled != old_throttled
+ end
+
end
end

MODIFY
test/beetle/client_test.rb
test/beetle/client_test.rb
@@ -430,6 +430,18 @@ module Beetle
client.trace(["test"])
end
+ test "delegates setting throttling options to the publisher instance and stringifies the queue names" do
+ client = Client.new
+ client.send(:publisher).expects(:throttle).with("test" => 1)
+ client.throttle(:test => 1)
+ end
+
+ test "delegates querying the throttling status to the publisher instance" do
+ client = Client.new
+ client.send(:publisher).expects(:throttled?)
+ client.throttled?
+ end
+
def message_stub_for_tracing
header_stub = stub_everything("header")
header_stub.stubs(:method).returns(stub_everything("method"))

MODIFY
test/beetle/publisher_test.rb
test/beetle/publisher_test.rb
@@ -80,6 +80,7 @@ module Beetle
@pub.send(:mark_server_dead)
publishing = sequence('publishing')
@pub.expects(:recycle_dead_servers).in_sequence(publishing)
+ @pub.expects(:throttle!).in_sequence(publishing)
@pub.expects(:publish_with_failover).with("mama-exchange", "mama", @data, @opts).in_sequence(publishing)
@pub.publish("mama", @data)
end
@@ -89,6 +90,7 @@ module Beetle
@pub.send(:mark_server_dead)
publishing = sequence('publishing')
@pub.expects(:recycle_dead_servers).in_sequence(publishing)
+ @pub.expects(:throttle!).in_sequence(publishing)
@pub.expects(:publish_with_redundancy).with("mama-exchange", "mama", @data, @opts.merge(:redundant => true)).in_sequence(publishing)
@pub.publish("mama", @data, :redundant => true)
end
@@ -339,6 +341,57 @@ module Beetle
@pub.setup_queues_and_policies(["queue"])
end
+ test "reports whether it has been throttled" do
+ assert !@pub.throttled?
+ @pub.instance_variable_set :@throttled, true
+ assert @pub.throttled?
+ end
+
+ test "sets throttling options" do
+ h = { "x" => 1, "y" => 2}
+ @pub.throttle(h)
+ assert_equal h, @pub.instance_variable_get(:@throttling_options)
+ end
+
+ test "throttle! sleeps appropriately when refreshing throttling" do
+ @pub.expects(:throttling?).returns(true).twice
+ @pub.expects(:refresh_throttling!).twice
+ @pub.expects(:throttled?).returns(true)
+ @pub.expects(:sleep).with(1)
+ @pub.throttle!
+ @pub.expects(:throttled?).returns(false)
+ @pub.expects(:sleep).never
+ @pub.throttle!
+ end
+
+ test "returns a throttling status" do
+ assert_equal 'unthrottled', @pub.throttling_status
+ @pub.instance_variable_set :@throttled, true
+ assert_equal 'throttled', @pub.throttling_status
+ end
+
+ test "refresh_throttling! does not recompute values when refresh interval has not passed" do
+ @pub.instance_variable_set :@next_throttle_refresh, Time.now + 1
+ options = {}
+ @pub.throttle(options)
+ options.expects(:each).never
+ @pub.__send__ :refresh_throttling!
+ end
+
+ test "refresh_throttling! throttles when queue length exceeds limit" do
+ assert !@pub.throttled?
+ @pub.instance_variable_set :@next_throttle_refresh, Time.now - 1
+ options = { "test" => 100 }
+ @pub.throttle(options)
+ @pub.expects(:each_server).yields
+ q = mock("queue")
+ q.expects(:status).returns(:message_count => 500)
+ @pub.expects(:queue).returns(q)
+ @pub.logger.expects(:info)
+ @pub.__send__ :refresh_throttling!
+ assert @pub.throttled?
+ end
+
end
class PublisherExchangeManagementTest < Minitest::Test

implement basic throttling functionality