Module: Karafka::Pro::Processing::ConsumerGroups::PeriodicJob::Consumer

Defined in:
lib/karafka/pro/processing/consumer_groups/periodic_job/consumer.rb

Overview

Consumer extra methods useful only when periodic jobs are in use

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(consumer_singleton_class) ⇒ Object

Defines an empty #tick method if not present

We define it that way due to our injection strategy flow.

Parameters:

  • consumer_singleton_class (Karafka::BaseConsumer) —

    consumer singleton class that is being enriched with periodic jobs API



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/karafka/pro/processing/consumer_groups/periodic_job/consumer.rb', line 46

def included(consumer_singleton_class)
  # Do not define empty tick method on consumer if it already exists
  # We only define it when it does not exist to have empty periodic ticking
  #
  # We need to check both cases (public and private) since user is not expected to
  # have this method public
  return if consumer_singleton_class.method_defined?(:tick)
  return if consumer_singleton_class.private_instance_methods.include?(:tick)

  # Create empty ticking method
  consumer_singleton_class.class_eval do
    def tick
    end
  end
end

Instance Method Details

#on_before_schedule_tick ⇒ Object

Runs the on-schedule tick periodic operations This method is an alias but is part of the naming convention used for other flows, this is why we do not reference the handle_before_schedule_tick directly



66
67
68
 
# File 'lib/karafka/pro/processing/consumer_groups/periodic_job/consumer.rb', line 66

def on_before_schedule_tick
  handle_before_schedule_tick
end

#on_tick ⇒ Object

Used by the executor to trigger consumer tick



72
73
74
75
76
77
78
79
80
81
 
# File 'lib/karafka/pro/processing/consumer_groups/periodic_job/consumer.rb', line 72

def on_tick
  handle_tick
rescue => e
  Karafka.monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    type: "consumer.tick.error"
  )
end