class ActiveMessaging::MessageReceiver
Attributes
connection[RW]
pause[RW]
poller[RW]
Public Class Methods
new(poller, connection, pause=1)
click to toggle source
# File lib/activemessaging/threaded_poller.rb, line 156 def initialize(poller, connection, pause=1) logger.debug("MessageReceiver initialize: poller:#{poller}, connection:#{connection}, pause:#{pause}") raise "No connection found for '#{poller.connection}'" unless connection self.poller = poller self.connection = connection self.pause = pause end
Public Instance Methods
inspect()
click to toggle source
# File lib/activemessaging/threaded_poller.rb, line 187 def inspect "#<MessageReceiver #{to_s}>" end
logger()
click to toggle source
# File lib/activemessaging/threaded_poller.rb, line 195 def logger; ::ActiveMessaging.logger; end
receive(worker)
click to toggle source
# File lib/activemessaging/threaded_poller.rb, line 166 def receive(worker) return unless poller.running # logger.debug("***** MessageReceiver calling receive") message = self.connection.receive(worker.options) # logger.debug("***** MessageReceiver receive returned") if message logger.debug("ActiveMessaging::MessageReceiver.receive: message:'#{message.inspect}'") poller.dispatch!(message, worker) else if (!poller || !poller.alive? || !poller.running) logger.debug("ActiveMessaging::MessageReceiver.receive: terminate") self.terminate end logger.debug("ActiveMessaging::MessageReceiver.receive: no message for worker #{worker.object_id}, retry in #{pause} sec") after(pause) { receive(worker) } end end
to_s()
click to toggle source
# File lib/activemessaging/threaded_poller.rb, line 191 def to_s @str ||= "#{Process.pid}-#{Thread.current.object_id}:#{self.object_id}" end