Class: Karafka::Routing::Topics

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Enumerable
Defined in:
lib/karafka/routing/topics.rb

Overview

Abstraction layer on top of groups of topics

Instance Method Summary collapse

Constructor Details

#initialize(topics_array) ⇒ Topics

Returns a new instance of Topics.

Parameters:



13
14
15
 
# File 'lib/karafka/routing/topics.rb', line 13

def initialize(topics_array)
  @accumulator = topics_array.dup
end

Instance Method Details

#<<(topic) ⇒ self

Note:

Topics can be appended to this collection at runtime from one thread while other threads iterate the same collection (for example during routing lookups). Swapping the @accumulator reference atomically (rather than appending in place) guarantees a concurrent #each always traverses a complete, immutable snapshot: it either sees the newly added topic or it does not, but never a torn array. This holds on every Ruby runtime, not only on MRI's GIL. The in-place mutators above (map!, sort_by!, reverse!) only run while routes are built at boot, single-threaded, so they do not need the same treatment.

Adds a topic using copy-on-write: we publish a brand-new accumulator array instead of mutating the existing one in place.

Parameters:

Returns:

  • (self) 


32
33
34
35
 
# File 'lib/karafka/routing/topics.rb', line 32

def <<(topic)
  @accumulator += [topic]
  self
end

#delete_if ⇒ Object

Allows us to remove elements from the topics

Block to decide what to delete



45
46
47
 
# File 'lib/karafka/routing/topics.rb', line 45

def delete_if(&)
  @accumulator.delete_if(&)
end

#each ⇒ Object

Yields each topic



38
39
40
 
# File 'lib/karafka/routing/topics.rb', line 38

def each(&)
  @accumulator.each(&)
end

#find(topic_name) ⇒ Karafka::Routing::Topic

Finds topic by its name

Parameters:

  • topic_name (String) —

    topic name

Returns:

Raises:



55
56
57
58
 
# File 'lib/karafka/routing/topics.rb', line 55

def find(topic_name)
  @accumulator.find { |topic| topic.name == topic_name } ||
    raise(Karafka::Errors::TopicNotFoundError, topic_name)
end