adhoc/issue121_02.rb

   1 # -*- encoding: utf-8 -*-
   2
   3 require 'stomp'
   4
   5 # Focus on this gem's capabilities.
   6 # require 'memory_profiler'
   7 require 'memory-profiler'
   8
   9 if Kernel.respond_to?(:require_relative)
  10   require_relative("stomp_adhoc_common")
  11   require_relative("payload_generator")
  12 else
  13   $LOAD_PATH << File.dirname(__FILE__)
  14   require "stomp_adhoc_common"
  15   require("payload_generator")
  16 end
  17 include Stomp11Common
  18
  19 # Round 2 of testing around issue #121.
  20
  21 class Issue121Examp02
  22
  23   attr_reader :client, :session
  24
  25   # Initialize.
  26   def initialize(topic = false)
  27     @client, @session, @topic = nil, nil, topic
  28     @nmsgs = nmsgs()
  29     @queue = make_destination("issue121/test_02")
  30     @id = "issue121_02"
  31     @block = cli_block()
  32     #
  33     @cmin, @cmax = 1292, 67782 # From the issue discussion
  34     PayloadGenerator::initialize(min= @cmin, max= @cmax)
  35     @ffmts = "%16.6f"
  36     #
  37     mps = 5.6 # see issue discussion
  38     @to, @nmts, @nts, @umps = 0.0, Time.now.to_f, @nmsgs, mps
  39     @tslt = 1.0 / @umps
  40   end # initialize
  41
  42   # Startup
  43   def start
  44     #
  45     client_hdrs = {"accept-version" => "1.1,1.2",
  46       "host" => virt_host,
  47     }
  48     #
  49     client_hash = { :hosts => [
  50         {:login => login(), :passcode => passcode(), :host => host(), :port => port()},
  51       ],
  52       :connect_headers => client_hdrs,
  53     }
  54     #
  55     @client = Stomp::Client.new(client_hash)
  56     puts "START: Client Connect complete"
  57     raise "START: Connection failed!!" unless @client.open?
  58     raise "START: Unexpected protocol level!!" if @client.protocol() == Stomp::SPL_10
  59     cf = @client.connection_frame()
  60     puts "START: Connection frame\n#{cf}"
  61     raise "START: Connect error!!: #{cf.body}" if @client.connection_frame().command == Stomp::CMD_ERROR
  62     @session = @client.connection_frame().headers['session']
  63     puts "START: Queue/Topic Name: #{@queue}"
  64     puts "START: Session: #{@session}"
  65     puts "START: NMSGS: #{@nmsgs}"
  66     puts "START: Block: #{@block}"
  67     puts "START: Wanted Messages Per Second: #{@umps}"
  68     puts "START: Sleep Time: #{@tslt}"
  69     $stdout.flush
  70   end # start
  71
  72   #
  73   def shutdown
  74     @client.close
  75     #
  76     te = Time.now.to_f
  77     et = te - @nmts
  78     avgsz = @to / @nts
  79     mps = @nts.to_f / et
  80     #
  81     fet = sprintf(@ffmts, et)
  82     favgsz = sprintf(@ffmts, avgsz)
  83     fmps = sprintf(@ffmts, mps)
  84     #
  85     sep = "=" * 72
  86     puts sep
  87     puts "\tNumber of payloads generated: #{@nts}"
  88     puts "\tMin Length: #{@cmin}, Max Length: #{@cmax}"
  89     puts "\tAVG_SIZE: #{favgsz}, ELAPS_SEC: #{fet}(seconds)"
  90     puts "\tNMSGS_PER_SEC: #{fmps}"
  91     puts sep
  92     #
  93     puts "SHUT: Shutdown complete"
  94     $stdout.flush
  95   end # shutdown
  96
  97   # pub
  98   def publish
  99     m = "Message: "
 100     nm = 0
 101
 102     @nmsgs.times do |n|
 103       nm += 1
 104       puts "PUB: NEXT MESSAGE NUMBER: #{nm}"; $stdout.flush
 105       mo = PayloadGenerator::payload()
 106       @to += mo.bytesize()
 107       hs = {:session => @session}
 108
 109       if @block
 110         ip = false
 111         @client.publish(@queue,
 112           mo,
 113           hs) {|m|
 114             puts "PUB: HAVE_RECEIPT:\nID: #{m.headers['receipt-id']}"
 115             $stdout.flush
 116             ip = m
 117         }
 118         sleep 0.01 until ip
 119       else
 120         @client.publish(@queue, mo, hs)
 121       end # if @block
 122
 123       puts "PUB: start user sleep"
 124       sleep @tslt # see issue discussion
 125       puts "PUB: end user sleep"
 126       $stdout.flush
 127     end # @nmsgs.times do
 128
 129     puts "PUB: end of publish"
 130     $stdout.flush
 131   end # publish
 132
 133 end # class
 134
 135 #
 136 # :limit => is max number of classes to report on
 137 MemoryProfiler::start_daemon( :limit=>25, :delay=>10, :marshal_size=>true, :sort_by=>:absdelta )
 138 #
 139 1.times do |i|
 140   rpt  = MemoryProfiler.start( :limit=> 25 ) do
 141     e = Issue121Examp02.new
 142     e.start
 143     e.publish
 144     # No subscribes here, just publish
 145     # See discussion in issue #121
 146     e.shutdown
 147   end
 148   puts MemoryProfiler.format(rpt)
 149   sleep 1
 150 end
 151 #
 152 MemoryProfiler::stop_daemon