@@ -32,22 +32,31 @@ require "rubygems" require "qpid" require 'getoptlong' +require 'math/statistics' + +class Array + include Math::Statistics +end + module BunnyPunisher - ITEMS_PER_ITERATION = 1000 - MAX_ITERATIONS = 10 + ITEMS_PER_ITERATION = 100 + MAX_ITERATIONS = 1 EMPTY_HEADER = {} - def self.connect + def self.get_timestamp + "%.10f" % Time.now.to_f + end + + def self.connect(host) spec = "#{Gem.loaded_specs['colinsurprenant-qpid'].full_gem_path}/specs/official-amqp0-8.xml" - host = '127.0.0.1' port = 5672 vhost = '/' user = 'guest' pass = 'guest' - puts("connecting to #{host}:#{port} on vhost #{vhost} with user=#{user}, pass=#{pass}") + #puts("connecting to #{host}:#{port} on vhost #{vhost} with user=#{user}, pass=#{pass}") client = Qpid::Client.new(host, port, Qpid::Spec::Loader.build(spec), vhost) client.start({ "LOGIN" => user, "PASSWORD" => pass }) @@ -58,25 +67,26 @@ module BunnyPunisher end def self.close(channel, client) - puts("closing channel and client") + #puts("closing channel and client") channel.channel_close client.close end def self.qpid_publish(channel, key, data) - c = Qpid::Content.new(EMPTY_HEADER, Marshal.dump(data)) + timestamp = get_timestamp + c = Qpid::Content.new(EMPTY_HEADER, Marshal.dump(timestamp+","+data)) channel.basic_publish(:routing_key => key.to_s, :content => c, :exchange => 'amq.direct') end def self.bench time_start = Time.now yield - puts("sent #{ITEMS_PER_ITERATION} in #{(Time.now - time_start).to_s} sec") + #puts("sent #{ITEMS_PER_ITERATION} in #{(Time.now - time_start).to_s} sec") end def self.publish(channel, key, value) - puts("publishing on queue=#{key}, data=#{value.to_s}") + #puts("publishing on queue=#{key}, bytes=#{value.length + 22} ") (1..MAX_ITERATIONS).each do self.bench do @@ -84,15 +94,16 @@ module BunnyPunisher (1..ITEMS_PER_ITERATION).each { self.qpid_publish(channel, key, value) } self.qpid_publish(channel, key, "end") end - sleep(2) # allow qpid queues to empty + sleep(1) # allow qpid queues to empty end self.qpid_publish(channel, key, "stop") end def self.consume(channel, client, key) - puts("consuming on queue=#{key}") - + #puts("consuming on queue=#{key}") + messages = Array.new + channel.queue_declare(:queue => key) channel.queue_bind(:queue_name => key, :exchange => 'amq.direct') bc = channel.basic_consume(:queue => key) @@ -104,40 +115,62 @@ module BunnyPunisher message = queue.pop(non_block = false) channel.basic_ack(message.delivery_tag) value = Marshal.load(message.content.body) -# puts("received value=#{value.inspect}") + #puts("received value=#{value.inspect}") - case value - when "begin" - i = 0 - time_start = Time.now - when "end" - puts("received #{i - 1} in #{(Time.now - time_start).to_s} sec") - end - - i += 1 - break if value == "stop" + + messages << value + "," + get_timestamp + diff = Array.new + if value.include? "stop" then + messages.each {|m| + t1 = Time.at(m.split(",")[0].to_f) + t2 = Time.at(m.split(",")[2].to_f) + diff << (t2-t1).to_s.to_f + } + + puts("avg=#{diff.avg}, std=#{diff.std}") + break + end end end def self.main opts = GetoptLong.new( [ '--consumer', '-c', GetoptLong::NO_ARGUMENT ], - [ '--publisher', '-p', GetoptLong::NO_ARGUMENT ] + [ '--publisher', '-p', GetoptLong::NO_ARGUMENT ], + [ '--queuename', '-q', GetoptLong::REQUIRED_ARGUMENT], + [ '--host', '-h', GetoptLong::REQUIRED_ARGUMENT], + [ '--bytes', '-b', GetoptLong::REQUIRED_ARGUMENT] ) - mode = :consumer # default + #new default options + host = '192.168.1.1' + mode = :consumer + queue = "test_queue" + bytes = 100 + data = "" + + opts.each do |opt, arg| case opt when '--consumer' mode = :consumer when '--publisher' mode = :publisher + when '--queuename' + queue = arg.to_s + when '--host' + host = arg.to_s + when '--bytes' + if bytes < get_timestamp.length + 1 then + bytes = get_timestamp.length + 1 + end + bytes = arg.to_i - get_timestamp.length + 1 end end - - channel, client = connect - key = "test_queue" - data = "test data" + + channel, client = connect(host) + key = queue + bytes.times do data.concat("a") end case mode when :consumer @@ -150,6 +183,6 @@ module BunnyPunisher end end -puts("starting Qpid Punisher") +#puts("starting Qpid Punisher") BunnyPunisher::main -puts("exiting Qpid Punisher") \ No newline at end of file +#puts("exiting Qpid Punisher") \ No newline at end of file qpid_punisher.rb 6bba970cd1b3cfa2e6ed7f969e8796b9cf0dc795 unknown sheheryar@.office http://github.com/sheysrebellion/bunnypunisher/commit/6160b0d8e57f78bc6e0f361f6f52581b2afd6d6a 6160b0d8e57f78bc6e0f361f6f52581b2afd6d6a 2009-06-24T04:38:06-07:00 2009-06-24T04:38:06-07:00 initial commit 7d8cfa3d6e285f05f96eeb4d867a282574caccf3 unknown sheheryar@.office