Sha256: ac7bac4939ce2873e28bd313cf9fdbc411110a31b55d5a768d408df24045792b
Contents?: true
Size: 1.6 KB
Versions: 28
Compression:
Stored size: 1.6 KB
Contents
require "thread"
require "gorgon_amq/protocol/client"
require "gorgon_amq/protocol/frame"
module GorgonBunny
# Periodically sends heartbeats, keeping track of the last publishing activity.
#
# @private
class HeartbeatSender
#
# API
#
def initialize(transport, logger)
@transport = transport
@logger = logger
@mutex = Monitor.new
@last_activity_time = Time.now
end
def start(period = 30)
@mutex.synchronize do
# calculate interval as half the given period plus
# some compensation for Ruby's implementation inaccuracy
# (we cannot get at the nanos level the Java client uses, and
# our approach is simplistic). MK.
@interval = [(period / 2) - 1, 0.4].max
@thread = Thread.new(&method(:run))
end
end
def stop
@mutex.synchronize { @thread.exit }
end
def signal_activity!
@last_activity_time = Time.now
end
protected
def run
begin
loop do
self.beat
sleep @interval
end
rescue IOError => ioe
@logger.error "I/O error in the hearbeat sender: #{ioe.message}"
stop
rescue Exception => e
@logger.error "Error in the hearbeat sender: #{e.message}"
stop
end
end
def beat
now = Time.now
if now > (@last_activity_time + @interval)
@logger.debug "Sending a heartbeat, last activity time: #{@last_activity_time}, interval (s): #{@interval}"
@transport.write_without_timeout(GorgonAMQ::Protocol::HeartbeatFrame.encode)
end
end
end
end
Version data entries
28 entries across 28 versions & 1 rubygems