Class: LS4::HeartbeatClientService

Inherits:
Service show all
Defined in:
lib/ls4/service/heartbeat.rb

Direct Known Subclasses

HeartbeatMemberService

Instance Method Summary collapse

Methods inherited from Service

init

Methods included from EventBus::SingletonMixin

#ebus_bind!, #ebus_connect, extended

Methods included from EventBus::BusMixin

#ebus_all_slots, #ebus_disconnect!

Methods included from EventBus::DeclarerBase::Methods

#connect, #ebus_all_slots, #ebus_call_log, #ebus_call_slots, #ebus_signal_error, #ebus_signal_log, #ebus_signal_slots

Methods included from EventBus::DeclarerBase

#call_slot, #signal_slot

Constructor Details

#initialize ⇒ HeartbeatClientService

Returns a new instance of HeartbeatClientService.



72
73
74
 
# File 'lib/ls4/service/heartbeat.rb', line 72

def initialize
  @heartbeat_nid = nil
end

Instance Method Details

#do_heartbeat_blocking ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/ls4/service/heartbeat.rb', line 127

def do_heartbeat_blocking
  sync_hash = SyncBus.get_hash
  begin
    res = @cs.call(:heartbeat, @heartbeat_nid, sync_hash)
    hbres = HeartbeatResponse.new.from_msgpack(res)
    if hbres.sync_needed
      ProcessBus.submit {
        SyncBus.try_sync
      }
    end
  rescue
    $log.error "heartbeat error: #{$!}"
    $log.error_backtrace $!.backtrace
  end
  nil
end

#heartbeat_blocking! ⇒ Object 



144
145
146
 
# File 'lib/ls4/service/heartbeat.rb', line 144

def heartbeat_blocking!
  do_heartbeat_blocking
end

#run ⇒ Object

def get_cs_session ProcessBus.get_session(ConfigBus.get_cs_address) end

def on_timer do_heartbeat end

def do_heartbeat sync_hash = SyncBus.get_hash get_cs_session.callback(:heartbeat, @heartbeat_nid, sync_hash) do |future| begin hbres = HeartbeatResponse.new.from_msgpack(future.get) ack_heartbeat(hbres) rescue $log.error “heartbeat error: #$!” $log.error_backtrace $!.backtrace end end end

def ack_heartbeat(hbres) if hbres.sync_needed SyncBus.try_sync end end

ebus_connect :ProcessBus, :on_timer

def heartbeat_blocking! do_heartbeat.join end



110
111
112
113
114
115
116
117
118
119
 
# File 'lib/ls4/service/heartbeat.rb', line 110

def run
  @cs = MessagePack::RPC::Client.new(*ConfigBus.get_cs_address)
  @end = false
  @thread = Thread.new do
    while !@end
      sleep 1
      do_heartbeat_blocking
    end
  end
end

#shutdown ⇒ Object 



121
122
123
124
125
 
# File 'lib/ls4/service/heartbeat.rb', line 121

def shutdown
  @end = true
  #@thread.join
  @cs.close
end