1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
# -*- encoding: utf-8 -*-
require 'stomp'
# Focus on this gem's capabilities.
# require 'memory_profiler'
require 'memory-profiler'
if Kernel.respond_to?(:require_relative)
require_relative("stomp_adhoc_common")
else
$LOAD_PATH << File.dirname(__FILE__)
require "stomp_adhoc_common"
end
include Stomp11Common
# Initial testing around issue #121.
class Issue121Examp01
attr_reader :client, :session
# Initialize.
def initialize(topic = false)
@client, @session, @topic = nil, nil, topic
@nmsgs = nmsgs()
@queue = make_destination("issue121/test_01")
@id = "issue121_01"
@block = cli_block()
end # initialize
# Startup
def start
#
client_hdrs = {"accept-version" => "1.1,1.2",
"host" => virt_host,
}
#
client_hash = { :hosts => [
{:login => login(), :passcode => passcode(), :host => host(), :port => port()},
],
:connect_headers => client_hdrs,
}
#
@client = Stomp::Client.new(client_hash)
puts "START: Client Connect complete"
raise "START: Connection failed!!" unless @client.open?
raise "START: Unexpected protocol level!!" if @client.protocol() == Stomp::SPL_10
cf = @client.connection_frame()
puts "START: Connection frame\n#{cf}"
raise "START: Connect error!!: #{cf.body}" if @client.connection_frame().command == Stomp::CMD_ERROR
@session = @client.connection_frame().headers['session']
puts "START: Queue/Topic Name: #{@queue}"
puts "START: Session: #{@session}"
puts "START: Block: #{@block}"
$stdout.flush
end # start
#
def shutdown
@client.close
puts "SHUT: Shutdown complete"
end # shutdown
# pub
def publish
m = "Message: "
nm = 0
@nmsgs.times do |n|
nm += 1
puts "PUB: NEXT MESSAGE NUMBER: #{nm}"
mo = "#{m} #{n}"
puts "PUB: PAYLOAD: #{mo}"
hs = {:session => @session}
if @block
ip = false
@client.publish(@queue,
mo,
hs) {|m|
puts "PUB: HAVE_RECEIPT:\nID: #{m.headers['receipt-id']}"
ip = m
}
sleep 0.01 until ip
else
@client.publish(@queue, mo, hs)
end # if @block
end # @nmsgs.times do
end # publish
# sub
def subscribe
puts "SUB: Subscribe starts For: #{@queue}"
rmc, done = 0, false
sh = {:id => "#{@id}", :ack => "auto"}
@client.subscribe(@queue, sh) {|m|
rmc += 1
rm = m
puts "SUB: HAVE_MESSAGE:\n#{rm}"
if rmc >= @nmsgs
puts "SUB: Subscribe is ending for #{@queue}"
done = true
Thread.done
end
}
sleep 0.01 until done
puts "SUB: Receives Done For: #{@queue}"
end # subscribe
end # class
#
# puts "BEG: Memory Profiler Version is: #{MemoryProfiler::VERSION}"
MemoryProfiler::start_daemon( :limit=>5, :delay=>10, :marshal_size=>true, :sort_by=>:absdelta )
#
5.times do |i|
rpt = MemoryProfiler.start( :limit=>10 ) do
e = Issue121Examp01.new
e.start
e.publish
e.subscribe
e.shutdown
end
puts MemoryProfiler.format(rpt)
sleep 1
end
#
MemoryProfiler::stop_daemon
|
