Class: Karafka::Persistence::Controller

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/persistence/controller.rb

Overview

Module used to provide a persistent cache across batch requests for a given topic and partition to store some additional details when the persistent mode for a given topic is turned on

Constant Summary collapse

PERSISTENCE_SCOPE =

Thread.current scope under which we store controllers data

:controllers

Class Method Summary collapse

Class Method Details

.all ⇒ Hash

Returns current thread persistence scope hash with all the controllers.

Returns:

  • (Hash) —

    current thread persistence scope hash with all the controllers



16
17
18
 
# File 'lib/karafka/persistence/controller.rb', line 16

def all
  Thread.current[PERSISTENCE_SCOPE] ||= {}
end

.fetch(topic, partition) ⇒ Karafka::BaseController

Used to build (if block given) and/or fetch a current controller instance that will be

used to process messages from a given topic and partition

Parameters:

  • topic (Karafka::Routing::Topic) —

    topic instance for which we might cache

  • partition (Integer) —

    number of partition for which we want to cache

Returns:



25
26
27
28
29
30
31
32
33
34
 
# File 'lib/karafka/persistence/controller.rb', line 25

def fetch(topic, partition)
  all[topic.id] ||= {}

  # We always store a current instance
  if topic.persistent
    all[topic.id][partition] ||= yield
  else
    all[topic.id][partition] = yield
  end
end