6 attr_reader :connection
9 def on_connecting(params=nil)
10 Log.info("TCP Connection attempt %d to %s" % [params[:cur_conattempts], stomp_url(params)])
14 def on_connected(params=nil)
15 Log.info("Conncted to #{stomp_url(params)}")
19 def on_disconnect(params=nil)
20 Log.info("Disconnected from #{stomp_url(params)}")
24 def on_connectfail(params=nil)
25 Log.info("TCP Connection to #{stomp_url(params)} failed on attempt #{params[:cur_conattempts]}")
29 def on_miscerr(params, errstr)
30 Log.error("Unexpected error on connection #{stomp_url(params)}: #{errstr}")
34 def on_ssl_connecting(params)
35 Log.info("Estblishing SSL session with #{stomp_url(params)}")
39 def on_ssl_connected(params)
40 Log.info("SSL session established with #{stomp_url(params)}")
44 def on_ssl_connectfail(params)
45 Log.error("SSL session creation with #{stomp_url(params)} failed: #{params[:ssl_exception]}")
48 # Stomp 1.1+ - heart beat read (receive) failed.
49 def on_hbread_fail(params, ticker_data)
50 Log.error("Heartbeat read failed from '%s': %s" % [stomp_url(params), ticker_data.inspect])
54 # Stomp 1.1+ - heart beat send (transmit) failed.
55 def on_hbwrite_fail(params, ticker_data)
56 Log.error("Heartbeat write failed from '%s': %s" % [stomp_url(params), ticker_data.inspect])
60 # Log heart beat fires
61 def on_hbfire(params, srind, curt)
64 Log.debug("Received heartbeat from %s: %s, %s" % [stomp_url(params), srind, curt])
66 Log.debug("Publishing heartbeat to %s: %s, %s" % [stomp_url(params), srind, curt])
72 "%s://%s@%s:%d" % [ params[:cur_ssl] ? "stomp+ssl" : "stomp", params[:cur_login], params[:cur_host], params[:cur_port]]
77 @config = Config.instance
82 # Connects to the RabbitMQ middleware
83 def connect(connector = ::Stomp::Connection)
85 Log.debug("Already connection, not re-initializing connection")
90 @base64 = get_bool_option("rabbitmq.base64", "false")
92 pools = @config.pluginconf["rabbitmq.pool.size"].to_i
95 1.upto(pools) do |poolnum|
98 host[:host] = get_option("rabbitmq.pool.#{poolnum}.host")
99 host[:port] = get_option("rabbitmq.pool.#{poolnum}.port", 61613).to_i
100 host[:login] = get_env_or_option("STOMP_USER", "rabbitmq.pool.#{poolnum}.user")
101 host[:passcode] = get_env_or_option("STOMP_PASSWORD", "rabbitmq.pool.#{poolnum}.password")
102 host[:ssl] = get_bool_option("rabbitmq.pool.#{poolnum}.ssl", "false")
103 host[:ssl] = ssl_parameters(poolnum, get_bool_option("rabbitmq.pool.#{poolnum}.ssl.fallback", "false")) if host[:ssl]
105 Log.debug("Adding #{host[:host]}:#{host[:port]} to the connection pool")
109 raise "No hosts found for the RabbitMQ connection pool" if hosts.size == 0
111 connection = {:hosts => hosts}
113 # Various STOMP gem options, defaults here matches defaults for 1.1.6 the meaning of
114 # these can be guessed, the documentation isn't clear
115 connection[:initial_reconnect_delay] = Float(get_option("rabbitmq.initial_reconnect_delay", 0.01))
116 connection[:max_reconnect_delay] = Float(get_option("rabbitmq.max_reconnect_delay", 30.0))
117 connection[:use_exponential_back_off] = get_bool_option("rabbitmq.use_exponential_back_off", "true")
118 connection[:back_off_multiplier] = Integer(get_option("rabbitmq.back_off_multiplier", 2))
119 connection[:max_reconnect_attempts] = Integer(get_option("rabbitmq.max_reconnect_attempts", 0))
120 connection[:randomize] = get_bool_option("rabbitmq.randomize", "false")
121 connection[:backup] = get_bool_option("rabbitmq.backup", "false")
123 connection[:timeout] = Integer(get_option("rabbitmq.timeout", -1))
124 connection[:connect_timeout] = Integer(get_option("rabbitmq.connect_timeout", 30))
125 connection[:reliable] = true
126 connection[:max_hbrlck_fails] = Integer(get_option("rabbitmq.max_hbrlck_fails", 2))
127 connection[:max_hbread_fails] = Integer(get_option("rabbitmq.max_hbread_fails", 2))
129 connection[:connect_headers] = connection_headers
131 connection[:logger] = EventLogger.new
133 @connection = connector.new(connection)
134 rescue Exception => e
135 raise("Could not connect to RabbitMQ Server: #{e}")
139 def connection_headers
140 headers = {:"accept-version" => "1.0"}
142 heartbeat_interval = Integer(get_option("rabbitmq.heartbeat_interval", 0))
143 stomp_1_0_fallback = get_bool_option("rabbitmq.stomp_1_0_fallback", true)
145 headers[:host] = get_option("rabbitmq.vhost", "/")
147 if heartbeat_interval > 0
148 unless Util.versioncmp(stomp_version, "1.2.10") >= 0
149 raise("Setting STOMP 1.1 properties like heartbeat intervals require at least version 1.2.10 of the STOMP gem")
152 if heartbeat_interval < 30
153 Log.warn("Connection heartbeat is set to %d, forcing to minimum value of 30s")
154 heartbeat_interval = 30
157 heartbeat_interval = heartbeat_interval * 1000
158 headers[:"heart-beat"] = "%d,%d" % [heartbeat_interval + 500, heartbeat_interval - 500]
160 if stomp_1_0_fallback
161 headers[:"accept-version"] = "1.1,1.0"
163 headers[:"accept-version"] = "1.1"
166 Log.warn("Connecting without STOMP 1.1 heartbeats, consider setting plugin.rabbitmq.heartbeat_interval")
173 ::Stomp::Version::STRING
176 # Sets the SSL paramaters for a specific connection
177 def ssl_parameters(poolnum, fallback)
178 params = {:cert_file => get_cert_file(poolnum),
179 :key_file => get_key_file(poolnum),
180 :ts_files => get_option("rabbitmq.pool.#{poolnum}.ssl.ca", false)}
182 raise "cert, key and ca has to be supplied for verified SSL mode" unless params[:cert_file] && params[:key_file] && params[:ts_files]
184 raise "Cannot find certificate file #{params[:cert_file]}" unless File.exist?(params[:cert_file])
185 raise "Cannot find key file #{params[:key_file]}" unless File.exist?(params[:key_file])
187 params[:ts_files].split(",").each do |ca|
188 raise "Cannot find CA file #{ca}" unless File.exist?(ca)
192 Stomp::SSLParams.new(params)
194 raise "Stomp gem >= 1.2.2 is needed"
197 rescue Exception => e
199 Log.warn("Failed to set full SSL verified mode, falling back to unverified: #{e.class}: #{e}")
202 Log.error("Failed to set full SSL verified mode: #{e.class}: #{e}")
207 # Returns the name of the private key file used by RabbitMQ
208 # Will first check if an environment variable MCOLLECTIVE_RABBITMQ_POOLX_SSL_KEY exists,
209 # where X is the RabbitMQ pool number.
210 # If the environment variable doesn't exist, it will try and load the value from the config.
211 def get_key_file(poolnum)
212 ENV["MCOLLECTIVE_RABBITMQ_POOL%s_SSL_KEY" % poolnum] || get_option("rabbitmq.pool.#{poolnum}.ssl.key", false)
215 # Returns the name of the certificate file used by RabbitMQ
216 # Will first check if an environment variable MCOLLECTIVE_RABBITMQ_POOLX_SSL_CERT exists,
217 # where X is the RabbitMQ pool number.
218 # If the environment variable doesn't exist, it will try and load the value from the config.
219 def get_cert_file(poolnum)
220 ENV["MCOLLECTIVE_RABBITMQ_POOL%s_SSL_CERT" % poolnum] || get_option("rabbitmq.pool.#{poolnum}.ssl.cert", false)
223 # Receives a message from the RabbitMQ connection
225 Log.debug("Waiting for a message from RabbitMQ")
227 # When the Stomp library > 1.2.0 is mid reconnecting due to its reliable connection
228 # handling it sets the connection to closed. If we happen to be receiving at just
229 # that time we will get an exception warning about the closed connection so handling
230 # that here with a sleep and a retry.
232 msg = @connection.receive
233 rescue ::Stomp::Error::NoCurrentConnection
238 raise "Received a processing error from RabbitMQ: '%s'" % msg.body.chomp if msg.body =~ /Processing error/
240 Message.new(msg.body, msg, :base64 => @base64, :headers => msg.headers)
243 # Sends a message to the RabbitMQ connection
245 msg.base64_encode! if @base64
247 if msg.type == :direct_request
248 msg.discovered_hosts.each do |node|
249 target = target_for(msg, node)
251 Log.debug("Sending a direct message to RabbitMQ target '#{target[:name]}' with headers '#{target[:headers].inspect}'")
253 @connection.publish(target[:name], msg.payload, target[:headers])
256 target = target_for(msg)
258 Log.debug("Sending a broadcast message to RabbitMQ target '#{target[:name]}' with headers '#{target[:headers].inspect}'")
260 @connection.publish(target[:name], msg.payload, target[:headers])
264 def target_for(msg, node=nil)
265 if msg.type == :reply
266 target = {:name => msg.request.headers["reply-to"], :headers => {}, :id => ""}
268 elsif [:request, :direct_request].include?(msg.type)
269 target = make_target(msg.agent, msg.type, msg.collective, msg.reply_to, node)
272 raise "Don't now how to create a target for message type #{msg.type}"
276 # marks messages as valid for ttl + 10 seconds, we do this here
277 # rather than in make_target as this should only be set on publish
278 target[:headers]["expiration"] = ((msg.ttl + 10) * 1000).to_s
283 def make_target(agent, type, collective, reply_to=nil, node=nil)
284 raise("Unknown target type #{type}") unless [:directed, :broadcast, :reply, :request, :direct_request].include?(type)
285 raise("Unknown collective '#{collective}' known collectives are '#{@config.collectives.join ', '}'") unless @config.collectives.include?(collective)
287 target = {:name => "", :headers => {}, :id => nil}
289 if get_bool_option("rabbitmq.use_reply_exchange", false)
290 reply_path = "/exchange/mcollective_reply/%s_%s" % [ @config.identity, $$ ]
292 reply_path = "/temp-queue/mcollective_reply_%s" % agent
295 when :reply # receiving replies on a temp queue
296 target[:name] = reply_path
297 target[:id] = "mcollective_%s_replies" % agent
299 when :broadcast, :request # publishing a request to all nodes with an agent
300 target[:name] = "/exchange/%s_broadcast/%s" % [collective, agent]
302 target[:headers]["reply-to"] = reply_to
304 target[:headers]["reply-to"] = reply_path
306 target[:id] = "%s_broadcast_%s" % [collective, agent]
308 when :direct_request # a request to a specific node
309 raise "Directed requests need to have a node identity" unless node
311 target[:name] = "/exchange/%s_directed/%s" % [ collective, node]
312 target[:headers]["reply-to"] = reply_path
314 when :directed # subscribing to directed messages
315 target[:name] = "/exchange/%s_directed/%s" % [ collective, @config.identity ]
316 target[:id] = "%s_%s_directed_to_identity" % [ collective, @config.identity ]
322 # Subscribe to a topic or queue
323 def subscribe(agent, type, collective)
324 return if type == :reply
326 source = make_target(agent, type, collective)
328 unless @subscriptions.include?(source[:id])
329 Log.debug("Subscribing to #{source[:name]} with headers #{source[:headers].inspect.chomp}")
330 @connection.subscribe(source[:name], source[:headers], source[:id])
331 @subscriptions << source[:id]
333 rescue ::Stomp::Error::DuplicateSubscription
334 Log.error("Received subscription request for #{source.inspect.chomp} but already had a matching subscription, ignoring")
337 # Subscribe to a topic or queue
338 def unsubscribe(agent, type, collective)
339 return if type == :reply
341 source = make_target(agent, type, collective)
343 Log.debug("Unsubscribing from #{source[:name]}")
344 @connection.unsubscribe(source[:name], source[:headers], source[:id])
345 @subscriptions.delete(source[:id])
348 # Disconnects from the RabbitMQ connection
350 Log.debug("Disconnecting from RabbitMQ")
351 @connection.disconnect
355 # looks in the environment first then in the config file
356 # for a specific option, accepts an optional default.
358 # raises an exception when it cant find a value anywhere
359 def get_env_or_option(env, opt, default=nil)
360 return ENV[env] if ENV.include?(env)
361 return @config.pluginconf[opt] if @config.pluginconf.include?(opt)
362 return default if default
364 raise("No #{env} environment or plugin.#{opt} configuration option given")
367 # looks for a config option, accepts an optional default
369 # raises an exception when it cant find a value anywhere
370 def get_option(opt, default=nil)
371 return @config.pluginconf[opt] if @config.pluginconf.include?(opt)
372 return default unless default.nil?
374 raise("No plugin.#{opt} configuration option given")
377 # looks up a boolean value in the config
378 def get_bool_option(val, default)
379 Util.str_to_bool(@config.pluginconf.fetch(val, default))
385 # vi:tabstop=4:expandtab:ai