Contents

require "thread"
require "amq/protocol/client"
require "amq/protocol/frame"

module Bunny
  class HeartbeatSender

    #
    # API
    #

    def initialize(transport)
      @transport = transport
      @mutex     = Mutex.new

      @last_activity_time = Time.now
    end

    def start(period = 30)
      @mutex.synchronize do
        @period = period

        @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 (@period / 2)
        end
      rescue IOError => ioe
        # ignored
      rescue Exception => e
        puts e.message
      end
    end

    def beat
      now = Time.now

      if now > (@last_activity_time + @period)
        @transport.send_raw(AMQ::Protocol::HeartbeatFrame.encode)
      end
    end
  end
end

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
bunny-0.9.0.pre3	lib/bunny/heartbeat_sender.rb
bunny-0.9.0.pre2	lib/bunny/heartbeat_sender.rb
bunny-0.9.0.pre1	lib/bunny/heartbeat_sender.rb