Class: Karafka::Core::Monitoring::Monitor

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/core/monitoring/monitor.rb

Overview

Karafka monitor that can be used to pass through instrumentation calls to selected notifications bus.

It provides abstraction layer that allows us to use both our internal notifications as well as ActiveSupport::Notifications.

Instance Method Summary collapse

Constructor Details

#initialize(notifications_bus, namespace = nil) ⇒ Monitor

Returns a new instance of Monitor.

Parameters:

  • notifications_bus (Object) —

    either our internal notifications bus or ActiveSupport::Notifications

  • namespace (String, nil) (defaults to: nil) —

    namespace for events or nil if no namespace



20
21
22
23
24
 
# File 'lib/karafka/core/monitoring/monitor.rb', line 20

def initialize(notifications_bus, namespace = nil)
  @notifications_bus = notifications_bus
  @namespace = namespace
  @mapped_events = {}
end

Instance Method Details

#instrument(event_id, payload = EMPTY_HASH) ⇒ Object

Passes the instrumentation block (if any) into the notifications bus

Parameters:

  • event_id (String, Symbol) —

    event id

  • payload (Hash) (defaults to: EMPTY_HASH) 


30
31
32
33
34
 
# File 'lib/karafka/core/monitoring/monitor.rb', line 30

def instrument(event_id, payload = EMPTY_HASH, &)
  full_event_name = @mapped_events[event_id] ||= [event_id, @namespace].compact.join('.')

  @notifications_bus.instrument(full_event_name, payload, &)
end

#listeners ⇒ Hash{String => Array}

Note:

Please do not modify this hash. It should be used only for debugging.

Returns hash where keys are events and values are arrays with listeners subscribed to particular events. Since different events may have different listeners, this is returned that way.

Examples:

If you need to get only classes of listeners, you can run following code:

monitor.listeners.map(&:class)

Returns:

  • (Hash{String => Array}) —

    hash where keys are events and values are arrays with listeners subscribed to particular events. Since different events may have different listeners, this is returned that way.



57
58
59
 
# File 'lib/karafka/core/monitoring/monitor.rb', line 57

def listeners
  @notifications_bus.listeners
end

#subscribe ⇒ Object

Allows us to subscribe to the notification bus



37
38
39
 
# File 'lib/karafka/core/monitoring/monitor.rb', line 37

def subscribe(*, &)
  @notifications_bus.subscribe(*, &)
end

#unsubscribe(listener_or_block) ⇒ Object

Allows for removal of whatever was subscribed

Parameters:

  • listener_or_block (Object) —

    object that is subscribed whether this is a listener instance or a block.



45
46
47
 
# File 'lib/karafka/core/monitoring/monitor.rb', line 45

def unsubscribe(listener_or_block)
  @notifications_bus.unsubscribe(listener_or_block)
end