Class: Takagi::EventBus::ObserverCleanup

Inherits:
Object
  • Object
show all
Defined in:
lib/takagi/event_bus/observer_cleanup.rb,
sig/takagi/event_bus/observer_cleanup.rbs

Overview

Background thread for stale observer cleanup

Periodically cleans up stale observers from the ObserveRegistry. Observers are considered stale if they haven't received notifications for longer than max_age seconds.

Examples:

cleanup = ObserverCleanup.new(interval: 60, max_age: 600)
cleanup.start
# ... cleanup runs in background ...
cleanup.stop

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(interval: 60, max_age: 600) ⇒ ObserverCleanup

Initialize observer cleanup

Parameters:

  • interval (Integer) (defaults to: 60) —

    Cleanup interval in seconds (default: 60)

  • max_age (Integer) (defaults to: 600) —

    Max observer age in seconds (default: 600)

  • interval: (::Integer) (defaults to: 60)
  • max_age: (::Integer) (defaults to: 600) 


22
23
24
25
26
27
28
29
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 22

def initialize(interval: 60, max_age: 600)
  @interval = interval
  @max_age = max_age
  @running = false
  @thread = nil
  @mutex = Mutex.new
  @stats = { runs: 0, cleaned: 0, errors: 0 }
end

Instance Attribute Details

#interval ⇒ Object (readonly)

Returns the value of attribute interval.

Returns:

  • (Object) 


17
18
19
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 17

def interval
  @interval
end

#max_age ⇒ Object (readonly)

Returns the value of attribute max_age.

Returns:

  • (Object) 


17
18
19
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 17

def max_age
  @max_age
end

Instance Method Details

#cleanup_now ⇒ Object

Force a cleanup run (for testing)

Returns:

  • (Object) 


71
72
73
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 71

def cleanup_now
  cleanup_stale_observers
end

#cleanup_stale_observers ⇒ Object

Cleanup stale observers from ObserveRegistry

Returns:

  • (Object) 


91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 91

def cleanup_stale_observers
  @mutex.synchronize { @stats[:runs] += 1 }

  cleaned_count =
    if defined?(Takagi::ObserveRegistry)
      Takagi::ObserveRegistry.cleanup_stale_observers(max_age: @max_age)
    else
      0
    end

  @mutex.synchronize { @stats[:cleaned] += cleaned_count }

  Takagi.logger.debug "Observer cleanup completed (run ##{@stats[:runs]})"
  Takagi.logger.info "Cleaned up #{cleaned_count} stale observers" if cleaned_count.positive?

  cleaned_count
end

#run_cleanup_loop ⇒ Object

Main cleanup loop

Returns:

  • (Object) 


78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 78

def run_cleanup_loop
  while @running
    begin
      sleep @interval
      cleanup_stale_observers if @running
    rescue StandardError => e
      @mutex.synchronize { @stats[:errors] += 1 }
      Takagi.logger.error "Observer cleanup error: #{e.class} - #{e.message}"
    end
  end
end

#running? ⇒ Boolean

Check if cleanup is running

Returns:

  • (Boolean) 


60
61
62
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 60

def running?
  @mutex.synchronize { @running }
end

#start ⇒ nil, untyped

Start the cleanup thread

Returns:

  • (nil, untyped) 


32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 32

def start
  @mutex.synchronize do
    return if @running

    @running = true
    @thread = Thread.new { run_cleanup_loop }
    @thread.name = 'ObserverCleanup'
  end

  Takagi.logger.info "Observer cleanup started (interval: #{@interval}s, max_age: #{@max_age}s)"
end

#stats ⇒ Hash

Get cleanup statistics

Returns:

  • (Hash) —

    Statistics



66
67
68
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 66

def stats
  @mutex.synchronize { @stats.dup }
end

#stop ⇒ nil, untyped

Stop the cleanup thread

Returns:

  • (nil, untyped) 


45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/takagi/event_bus/observer_cleanup.rb', line 45

def stop
  @mutex.synchronize do
    return unless @running

    @running = false
    @thread&.kill
    @thread&.join(5) # Wait up to 5 seconds
    @thread = nil
  end

  Takagi.logger.info 'Observer cleanup stopped'
end