Module: Listenable

Extended by:
ActiveSupport::Concern
Defined in:
lib/listenable.rb,
lib/listenable/concern.rb,
lib/listenable/railtie.rb,
lib/listenable/version.rb,
sig/listenable.rbs

Defined Under Namespace

Classes: ConnectionPoolExhausted, Error, Railtie

Constant Summary collapse

CALLBACK_MAP = 
{
  'created' => :after_create,
  'updated' => :after_update,
  'deleted' => :after_destroy
}.freeze
VERSION =

Returns:

  • (String) 
'0.3.2'

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.async_executor ⇒ Object

Thread pool executor for async listeners Auto-scales to 25% of connection pool (very conservative)



58
59
60
61
62
63
64
65
66
 
# File 'lib/listenable.rb', line 58

def async_executor
  @async_executor ||= Concurrent::ThreadPoolExecutor.new(
    min_threads: 0,
    max_threads: default_thread_pool_size,
    max_queue: 10_000,
    fallback_policy: :caller_runs,
    idletime: 60 # Threads idle for 60s are cleaned up
  )
end

Class Method Details

.cleanup! ⇒ Object

Cleanup all subscribers and shutdown thread pool Called on Rails reload to prevent memory leaks



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/listenable.rb', line 70

def cleanup!
  # Unsubscribe all tracked subscribers
  subscribers.each do |subscriber|
    ActiveSupport::Notifications.unsubscribe(subscriber)
  rescue StandardError => e
    Rails.logger&.warn("[Listenable] Failed to unsubscribe: #{e.message}")
  end
  @subscribers = []

  # Clear listener registry for fresh reload
  clear_listeners!

  # Shutdown thread pool gracefully
  shutdown_async_executor!
end

.clear_listeners! ⇒ Object

Clear registry on reload



37
38
39
 
# File 'lib/listenable.rb', line 37

def clear_listeners!
  @listener_classes = Set.new
end

.default_thread_pool_size ⇒ Object

Calculate a safe thread pool size based on connection pool Very conservative: use configurable ratio of pool (default 25%)



48
49
50
51
52
53
54
 
# File 'lib/listenable.rb', line 48

def default_thread_pool_size
  return 2 unless defined?(ActiveRecord::Base)

  pool_size = ActiveRecord::Base.connection_pool.size
  # Use configured ratio of pool, but at least 1 thread, max configured limit
  [[pool_size * max_thread_pool_ratio, 1].max, max_thread_pool_size].min.to_i
end

.listener_classes ⇒ Object

Registry of all listener classes (better than ObjectSpace scan)



27
28
29
 
# File 'lib/listenable.rb', line 27

def listener_classes
  @listener_classes ||= Set.new
end

.register_listener(klass) ⇒ Object

Register a listener class when Listenable is included



32
33
34
 
# File 'lib/listenable.rb', line 32

def register_listener(klass)
  listener_classes.add(klass) if klass.name && !klass.name.empty?
end

.reset_async_executor! ⇒ Object

For testing - immediate shutdown



99
100
101
102
103
104
105
 
# File 'lib/listenable.rb', line 99

def reset_async_executor!
  return unless @async_executor

  @async_executor.shutdown
  @async_executor.kill unless @async_executor.wait_for_termination(5)
  @async_executor = nil
end

.shutdown_async_executor! ⇒ Object

Graceful shutdown of thread pool



87
88
89
90
91
92
93
94
95
96
 
# File 'lib/listenable.rb', line 87

def shutdown_async_executor!
  return unless @async_executor

  @async_executor.shutdown
  unless @async_executor.wait_for_termination(10)
    Rails.logger&.warn('[Listenable] Thread pool shutdown timeout, forcing kill')
    @async_executor.kill
  end
  @async_executor = nil
end

.subscribers ⇒ Object

Track active subscribers to prevent memory leaks on reload



42
43
44
 
# File 'lib/listenable.rb', line 42

def subscribers
  @subscribers ||= []
end