Class: Pgbus::Process::Heartbeat

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/process/heartbeat.rb

Constant Summary collapse

INTERVAL =

seconds

60
ALIVE_THRESHOLD =

5 minutes

300

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(kind:, metadata: {}, on_beat: nil, loop_tick_supplier: nil, metadata_supplier: nil) ⇒ Heartbeat

Returns a new instance of Heartbeat.



14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/pgbus/process/heartbeat.rb', line 14

def initialize(kind:, metadata: {}, on_beat: nil, loop_tick_supplier: nil, metadata_supplier: nil)
  @kind = kind
  @metadata = metadata
  @on_beat = on_beat
  @loop_tick_supplier = loop_tick_supplier
  @metadata_supplier = metadata_supplier
  @timer = nil
  @stopped = false
  # Guards @process_id between the timer thread (beat) and the main
  # thread (stop): TimerTask#shutdown does not wait for an in-flight
  # beat, and a beat that re-registers after deregister_process ran
  # would leave a zombie row (issue #438).
  @mutex = Mutex.new
end

Instance Attribute Details

#process_entry ⇒ Object (readonly)

Returns the value of attribute process_entry.



12
13
14
 
# File 'lib/pgbus/process/heartbeat.rb', line 12

def process_entry
  @process_entry
end

Instance Method Details

#beat ⇒ Object 



41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/pgbus/process/heartbeat.rb', line 41

def beat
  return unless @process_id && !@stopped

  @on_beat&.call
  updates = { last_heartbeat_at: Time.current }
  metadata = beat_metadata
  updates[:metadata] = metadata unless metadata.nil?
  @mutex.synchronize { write_beat(updates) }
rescue StandardError => e
  Pgbus.logger.warn { "[Pgbus] Heartbeat failed: #{e.message}" }
end

#start ⇒ Object 



29
30
31
32
33
 
# File 'lib/pgbus/process/heartbeat.rb', line 29

def start
  register_process
  @timer = Concurrent::TimerTask.new(execution_interval: INTERVAL) { beat }
  @timer.execute
end

#stop ⇒ Object 



35
36
37
38
39
 
# File 'lib/pgbus/process/heartbeat.rb', line 35

def stop
  @stopped = true
  @timer&.shutdown
  @mutex.synchronize { deregister_process }
end