Class: Karafka::Pro::Processing::FiltersApplier
- Inherits:
-
Object
- Object
- Karafka::Pro::Processing::FiltersApplier
- Defined in:
- lib/karafka/pro/processing/filters_applier.rb
Overview
Applier for all filters we want to have. Whether related to limiting messages based on the payload or any other things.
From the outside world perspective, this encapsulates all the filters. This means that this is the API we expose as a single filter, allowing us to control the filtering via many filters easily.
Instance Attribute Summary collapse
-
#filters ⇒ Array
readonly
Registered filters array.
Instance Method Summary collapse
-
#action ⇒ Symbol
Consumer post-filtering action that should be taken.
-
#applied? ⇒ Boolean
Did we filter out any messages during filtering run.
- #apply!(messages) ⇒ Object
-
#cursor ⇒ Karafka::Messages::Message?
The first message we do need to get next time we poll.
-
#initialize(coordinator) ⇒ FiltersApplier
constructor
A new instance of FiltersApplier.
-
#timeout ⇒ Integer
Minimum timeout we need to pause.
Constructor Details
#initialize(coordinator) ⇒ FiltersApplier
Returns a new instance of FiltersApplier.
29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
# File 'lib/karafka/pro/processing/filters_applier.rb', line 29 def initialize(coordinator) # Builds filters out of their factories # We build it that way (providing topic and partition) because there may be a case where # someone wants to have a specific logic that is per topic or partition. Like for example # a case where there is a cache bypassing revocations for topic partition. # # We provide full Karafka routing topic here and not the name only, in case the filter # would be customized based on other topic settings (like VPs, etc) # # This setup allows for biggest flexibility also because topic object holds the reference # to the subscription group and consumer group @filters = coordinator.topic.filtering.factories.map do |factory| factory.call(coordinator.topic, coordinator.partition) end end |
Instance Attribute Details
#filters ⇒ Array (readonly)
Returns registered filters array. Useful if we want to inject internal context aware filters.
26 27 28 |
# File 'lib/karafka/pro/processing/filters_applier.rb', line 26 def filters @filters end |
Instance Method Details
#action ⇒ Symbol
Returns consumer post-filtering action that should be taken.
61 62 63 64 65 66 67 68 69 70 71 72 73 |
# File 'lib/karafka/pro/processing/filters_applier.rb', line 61 def action return :skip unless applied? # The highest priority is on a potential backoff from any of the filters because it is # the less risky (delay and continue later) return :pause if applied.any? { |filter| filter.action == :pause } # If none of the filters wanted to pause, we can check for any that would want to seek # and if there is any, we can go with this strategy return :seek if applied.any? { |filter| filter.action == :seek } :skip end |
#applied? ⇒ Boolean
Returns did we filter out any messages during filtering run.
54 55 56 57 58 |
# File 'lib/karafka/pro/processing/filters_applier.rb', line 54 def applied? return false unless active? !applied.empty? end |
#apply!(messages) ⇒ Object
47 48 49 50 51 |
# File 'lib/karafka/pro/processing/filters_applier.rb', line 47 def apply!() return unless active? @filters.each { |filter| filter.apply!() } end |
#cursor ⇒ Karafka::Messages::Message?
Cursor message can also return the offset in the time format
The first message we do need to get next time we poll. We use the minimum not to jump accidentally by over any.
85 86 87 88 89 |
# File 'lib/karafka/pro/processing/filters_applier.rb', line 85 def cursor return nil unless active? applied.map(&:cursor).compact.min_by(&:offset) end |
#timeout ⇒ Integer
Returns minimum timeout we need to pause. This is the minimum for all the filters to satisfy all of them.
77 78 79 |
# File 'lib/karafka/pro/processing/filters_applier.rb', line 77 def timeout applied.map(&:timeout).compact.min || 0 end |