Class: Promenade::Karafka::ConsumerSubscriber

Inherits:
Subscriber
  • Object
show all
Defined in:
lib/promenade/karafka/consumer_subscriber.rb

Instance Method Summary collapse

Instance Method Details

#consumed(event) ⇒ Object 



17
18
19
20
21
22
23
24
25
26
 
# File 'lib/promenade/karafka/consumer_subscriber.rb', line 17

def consumed(event)
  consumer = event.payload[:caller]
  messages = consumer.messages
  batch_processing_duration = convert_milliseconds_to_seconds(event.payload[:time])

  labels = get_labels(consumer)

  Promenade.metric(:karafka_consumer_messages_processed).increment(labels, messages.size)
  Promenade.metric(:karafka_consumer_batch_processing_duration_seconds).observe(labels, batch_processing_duration)
end