Class: Karafka::Pro::ScheduledMessages::Tracker

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/pro/scheduled_messages/tracker.rb

Overview

Tracks basic state and metrics about schedules to be dispatched

It provides accurate today dispatch taken from daily buffer and estimates for future days

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Tracker

Returns a new instance of Tracker.



27
28
29
30
 
# File 'lib/karafka/pro/scheduled_messages/tracker.rb', line 27

def initialize
  @daily = Hash.new { |h, k| h[k] = 0 }
  @created_at = Time.now.to_i
end

Instance Attribute Details

#daily ⇒ Hash<String, Integer> (readonly)

Returns:

  • (Hash<String, Integer>) 


22
23
24
 
# File 'lib/karafka/pro/scheduled_messages/tracker.rb', line 22

def daily
  @daily
end

#state ⇒ String

Returns current state.

Returns:

  • (String) —

    current state



25
26
27
 
# File 'lib/karafka/pro/scheduled_messages/tracker.rb', line 25

def state
  @state
end

Instance Method Details

#today=(sum) ⇒ Object

Accurate (because coming from daily buffer) number of things to schedule

Parameters:

  • sum (Integer) 


35
36
37
 
# File 'lib/karafka/pro/scheduled_messages/tracker.rb', line 35

def today=(sum)
  @daily[epoch_to_date(@created_at)] = sum
end

#track(message) ⇒ Object

Tracks message dispatch

It is only relevant for future days as for today we use accurate metrics from the daily buffer

Parameters:

  • message (Karafka::Messages::Message) —

    schedule message. Should not be a tombstone message. Tombstone messages cancellations are not tracked because it would drastically increase complexity. For given day we use the accurate counter and for future days we use estimates.



48
49
50
51
52
 
# File 'lib/karafka/pro/scheduled_messages/tracker.rb', line 48

def track(message)
  epoch = message.headers['schedule_target_epoch']

  @daily[epoch_to_date(epoch)] += 1
end