Module: AMQP::Client
- Includes:
- EM::Deferrable, RightSupport::Log::Mixin
- Defined in:
- lib/right_amqp/amqp/client.rb
Class Method Summary collapse
Instance Method Summary collapse
- #add_channel(mq) ⇒ Object
- #channels ⇒ Object
-
#close(&on_disconnect) ⇒ Object
:stopdoc: def send_data data log ‘send_data’, data super end :startdoc:.
- #connected? ⇒ Boolean
- #connection_completed ⇒ Object
- #connection_status(&blk) ⇒ Object
- #failed ⇒ Object
- #frame_max ⇒ Object
- #init_heartbeat ⇒ Object
- #initialize(opts = {}) ⇒ Object
- #process_frame(frame) ⇒ Object
-
#receive_data(data) ⇒ Object
Catch exceptions that would otherwise cause EM to stop or be in a bad state if a top level EM error handler was setup.
- #reconnect(force = false) ⇒ Object
- #send(data, opts = {}) ⇒ Object
- #unbind ⇒ Object
Class Method Details
.connect(opts = {}) ⇒ Object
298 299 300 301 |
# File 'lib/right_amqp/amqp/client.rb', line 298 def self.connect opts = {} opts = AMQP.settings.merge(opts) EM.connect opts[:host], opts[:port], self, opts end |
.included(base) ⇒ Object
80 81 82 |
# File 'lib/right_amqp/amqp/client.rb', line 80 def self.included(base) base.extend(RightSupport::Log::Mixin::ClassMethods) end |
Instance Method Details
#add_channel(mq) ⇒ Object
185 186 187 188 189 190 |
# File 'lib/right_amqp/amqp/client.rb', line 185 def add_channel mq (@_channel_mutex ||= Mutex.new).synchronize do channels[ key = (channels.keys.max || 0) + 1 ] = mq key end end |
#channels ⇒ Object
192 193 194 |
# File 'lib/right_amqp/amqp/client.rb', line 192 def channels @channels ||= {} end |
#close(&on_disconnect) ⇒ Object
:stopdoc: def send_data data
log 'send_data', data
super
end :startdoc:
237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 |
# File 'lib/right_amqp/amqp/client.rb', line 237 def close &on_disconnect @heartbeat_timer.cancel if @heartbeat_timer @heartbeat_timer = nil if on_disconnect @closing = true @on_disconnect = proc{ on_disconnect.call @closing = false } end callback{ |c| if c.channels.any? c.channels.each do |ch, mq| mq.close end else send Protocol::Connection::Close.new(:reply_code => 200, :reply_text => 'Goodbye', :class_id => 0, :method_id => 0) end } end |
#connected? ⇒ Boolean
170 171 172 |
# File 'lib/right_amqp/amqp/client.rb', line 170 def connected? @connected end |
#connection_completed ⇒ Object
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 |
# File 'lib/right_amqp/amqp/client.rb', line 96 def connection_completed start_tls if @settings[:ssl] log 'connected' # @on_disconnect = proc{ raise Error, 'Disconnected from server' } unless @closing @on_disconnect = method(:disconnected) @reconnecting = false end @connected = true logger.debug("[amqp] Connected to broker #{@settings[:identity]}") @connection_status.call(:connected) if @connection_status @buf = Buffer.new send_data HEADER send_data [1, 1, VERSION_MAJOR, VERSION_MINOR].pack('C4') if heartbeat = @settings[:heartbeat] init_heartbeat if (@settings[:heartbeat] = heartbeat.to_i) > 0 end end |
#connection_status(&blk) ⇒ Object
303 304 305 |
# File 'lib/right_amqp/amqp/client.rb', line 303 def connection_status &blk @connection_status = blk end |
#failed ⇒ Object
307 308 309 310 311 |
# File 'lib/right_amqp/amqp/client.rb', line 307 def failed @connection_status.call(:failed) if @connection_status @has_failed = true close_connection end |
#frame_max ⇒ Object
174 175 176 |
# File 'lib/right_amqp/amqp/client.rb', line 174 def frame_max @frame_max end |
#init_heartbeat ⇒ Object
118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
# File 'lib/right_amqp/amqp/client.rb', line 118 def init_heartbeat # Randomly offset start of heartbeat timer to help separate heartbeat # activity when there are multiple broker connections active EM.add_timer(rand(@settings[:heartbeat])) do begin # While connected, a heartbeat or some other data is expected to be received from # the broker at least every 2 x :heartbeat seconds, otherwise the connection is # assumed to be broken and therefore is closed to cause an automatic reconnect. # While connected, this client will send a heartbeat every :heartbeat # seconds regardless of any other send activity. The RabbitMQ broker will behave # similarly and drop the connection if it does not receive a heartbeat or other # data in time. logger.info("[amqp] Initializing heartbeat for broker #{@settings[:identity]} to #{@settings[:heartbeat]} sec") timeout_factor = 2 @heartbeat_timer.cancel if @heartbeat_timer @heartbeat_timer = EM::PeriodicTimer.new(@settings[:heartbeat]) do begin if connected? now = Time.now if @last_data_received.nil? # This is something of an anomaly; not clear how this condition is reached logger.info("[amqp] Reconnecting to broker #{@settings[:identity]} due to heartbeat timeout " + "with no data having been received") close_connection # which results in an unbind and an automatic reconnect elsif @last_data_received < (now - (@settings[:heartbeat] * timeout_factor)) data_received = (now - @last_data_received).to_i heartbeat_received = (now - @last_heartbeat_received).to_i if @last_heartbeat_received heartbeat_sent = (now - @last_heartbeat_sent).to_i if @last_heartbeat_sent logger.info("[amqp] Reconnecting to broker #{@settings[:identity]} due to heartbeat timeout: " + "last data received #{data_received.inspect} sec ago, " + "last heartbeat received #{heartbeat_received.inspect} sec ago, " + "last heartbeat sent #{heartbeat_sent.inspect} sec ago") close_connection # which results in an unbind and an automatic reconnect else logger.debug("[amqp] Sending heartbeat to broker #{@settings[:identity]}") send AMQP::Frame::Heartbeat.new, :channel => 0 @last_heartbeat_sent = Time.now end else logger.debug("[amqp] Skipping heartbeat check for broker #{@settings[:identity]} because disconnected") end rescue Exception => e logger.error("[amqp] Failed heartbeat check (#{e})\n" + e.backtrace.join("\n")) end end rescue Exception => e logger.error("[amqp] Failed heartbeat initialization (#{e})\n" + e.backtrace.join("\n")) end end end |
#initialize(opts = {}) ⇒ Object
84 85 86 87 88 89 90 91 92 93 94 |
# File 'lib/right_amqp/amqp/client.rb', line 84 def initialize opts = {} @settings = opts extend AMQP.client @on_disconnect ||= proc{ @connection_status.call(:failed) if @connection_status } timeout @settings[:timeout] if @settings[:timeout] errback{ @on_disconnect.call } unless @reconnecting @connected = false end |
#process_frame(frame) ⇒ Object
216 217 218 219 |
# File 'lib/right_amqp/amqp/client.rb', line 216 def process_frame frame # this is a stub meant to be # replaced by the module passed into initialize end |
#receive_data(data) ⇒ Object
Catch exceptions that would otherwise cause EM to stop or be in a bad state if a top level EM error handler was setup. Instead close the connection and leave EM alone. Don’t log an error if the environment variable IGNORE_AMQP_FAILURES is set (used in the enroll script)
201 202 203 204 205 206 207 208 209 210 211 212 213 214 |
# File 'lib/right_amqp/amqp/client.rb', line 201 def receive_data data begin # log 'receive_data', data @buf << data while frame = Frame.parse(@buf) log 'receive', frame process_frame frame end rescue Exception => e logger.exception("[amqp] Failed processing frame, closing connection", e, :trace) unless ENV['IGNORE_AMQP_FAILURES'] failed end end |
#reconnect(force = false) ⇒ Object
262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 |
# File 'lib/right_amqp/amqp/client.rb', line 262 def reconnect force = false if @reconnecting and not force # Wait after first reconnect attempt and in between each subsequent attempt EM.add_timer(@settings[:reconnect_interval] || 5) { reconnect(true) } return end unless @reconnecting @deferred_status = nil initialize(@settings) mqs = @channels @channels = {} mqs.each{ |_,mq| mq.reset } if mqs @reconnecting = true again = @settings[:reconnect_delay] again = again.call if again.is_a?(Proc) if again.is_a?(Numeric) # Wait before making initial reconnect attempt EM.add_timer(again) { reconnect(true) } return elsif ![nil, true].include?(again) raise ::AMQP::Error, "Could not interpret :reconnect_delay => #{again.inspect}; expected nil, true, or Numeric" end end log 'reconnecting' logger.info("[amqp] Attempting to reconnect to broker #{@settings[:identity]}") EM.reconnect(@settings[:host], @settings[:port], self) rescue Exception => e logger.exception("[amqp] Failed to reconnect", e, :trace) failed end |
#send(data, opts = {}) ⇒ Object
221 222 223 224 225 226 227 228 |
# File 'lib/right_amqp/amqp/client.rb', line 221 def send data, opts = {} channel = opts[:channel] ||= 0 data = data.to_frame(channel) unless data.is_a? Frame data.channel = channel log 'send', data send_data data.to_s end |
#unbind ⇒ Object
178 179 180 181 182 183 |
# File 'lib/right_amqp/amqp/client.rb', line 178 def unbind log 'disconnected' @connected = false logger.debug("[amqp] Disconnected from broker #{@settings[:identity]}") EM.next_tick{ @on_disconnect.call } end |