Class: Karafka::Pro::Processing::ConsumerGroups::Filters::Delayer

Inherits:
Base
  • Object
show all
Defined in:
lib/karafka/pro/processing/consumer_groups/filters/delayer.rb

Overview

A filter that allows us to delay processing by pausing until time is right.

Constant Summary

Constants included from Actions

Actions::ALL

Instance Attribute Summary

Attributes inherited from Base

#cursor

Instance Method Summary collapse

Methods inherited from Base

#applied?, #mark_as_consumed?, #marking_cursor, #marking_method

Methods included from Actions

pause, #pause?, seek, #seek?, skip, #skip?

Constructor Details

#initialize(delay) ⇒ Delayer 



39
40
41
42
43
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/delayer.rb', line 39

def initialize(delay)
  super()

  @delay = delay
end

Instance Method Details

#action ⇒ Symbol 



79
80
81
82
83
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/delayer.rb', line 79

def action
  return Actions.skip unless applied?

  (timeout <= 0) ? Actions.seek : Actions.pause
end

#apply!(messages) ⇒ Object

Removes too young messages



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/delayer.rb', line 48

def apply!(messages)
  @applied = false
  @cursor = nil

  # Time on message is in seconds with ms precision, so we need to convert the ttl that
  # is in ms to this format
  border = Time.now.utc - (@delay / 1_000.0)

  messages.delete_if do |message|
    too_young = message.timestamp > border

    if too_young
      @applied = true

      @cursor ||= message
    end

    @applied
  end
end

#timeout ⇒ Integer 



70
71
72
73
74
75
76
 
# File 'lib/karafka/pro/processing/consumer_groups/filters/delayer.rb', line 70

def timeout
  return 0 unless @cursor

  timeout = (@delay / 1_000.0) - (::Time.now.utc - @cursor.timestamp)

  (timeout <= 0) ? 0 : timeout * 1_000
end