adhoc/issue121_01_conn.rb

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