Class: Rdkafka::Consumer::TopicPartitionList

Inherits:
Object
  • Object
show all
Defined in:
lib/rdkafka/consumer/topic_partition_list.rb

Overview

A list of topics with their partition information

Instance Method Summary collapse

Constructor Details

#initialize(data = nil) ⇒ TopicPartitionList

Create a topic partition list.

Parameters:

  • data (Hash{String => nil,Partition}) (defaults to: nil) —

    The topic and partition data or nil to create an empty list



12
13
14
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 12

def initialize(data = nil)
  @data = data || {}
end

Instance Method Details

#==(other) ⇒ Boolean

Check equality with another TopicPartitionList

Parameters:

Returns:

  • (Boolean) 


92
93
94
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 92

def ==(other)
  to_h == other.to_h
end

#add_topic(topic, partitions = nil) ⇒ nil

Add a topic with optionally partitions to the list. Calling this method multiple times for the same topic will overwrite the previous configuraton.

Examples:

Add a topic with unassigned partitions

tpl.add_topic("topic")

Add a topic with assigned partitions

tpl.add_topic("topic", (0..8))

Add a topic with all topics up to a count

tpl.add_topic("topic", 9)

Parameters:

  • topic (String) —

    The topic's name

  • partitions (Array<Integer>, Range<Integer>, Integer) (defaults to: nil) —

    The topic's partitions or partition count

Returns:

  • (nil) 


52
53
54
55
56
57
58
59
60
61
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 52

def add_topic(topic, partitions = nil)
  if partitions.nil?
    @data[topic.to_s] = nil
  else
    if partitions.is_a? Integer
      partitions = (0..partitions - 1)
    end
    @data[topic.to_s] = partitions.map { |p| Partition.new(p, nil, Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR) }
  end
end

#add_topic_and_partitions_with_offsets(topic, partitions_with_offsets) ⇒ nil

Add a topic with partitions and offsets set to the list Calling this method multiple times for the same topic will overwrite the previous configuraton.

Parameters:

  • topic (String) —

    The topic's name

  • partitions_with_offsets (Hash{Integer => Integer}, Array<Consumer::Partition>) —

    The topic's partitions and offsets (Hash) or partitions with offsets and metadata (Array)

Returns:

  • (nil) 


70
71
72
73
74
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 70

def add_topic_and_partitions_with_offsets(topic, partitions_with_offsets)
  @data[topic.to_s] = partitions_with_offsets.map do |p, o|
    p.is_a?(Partition) ? p : Partition.new(p, o)
  end
end

#count ⇒ Integer

Number of items in the list

Returns:

  • (Integer) 


18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 18

def count
  i = 0
  @data.each do |_topic, partitions|
    i += if partitions
      partitions.count
    else
      1
    end
  end
  i
end

#empty? ⇒ Boolean

Whether this list is empty

Returns:

  • (Boolean) 


32
33
34
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 32

def empty?
  @data.empty?
end

#to_h ⇒ Hash{String => Array<Partition>,nil}

Return a Hash with the topics as keys and and an array of partition information as the value if present.

Returns:



79
80
81
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 79

def to_h
  @data
end

#to_s ⇒ String

Human readable representation of this list.

Returns:

  • (String) 


85
86
87
 
# File 'lib/rdkafka/consumer/topic_partition_list.rb', line 85

def to_s
  "<TopicPartitionList: #{to_h}>"
end