Class: Kafka::Broker

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/broker.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection:, node_id: nil, logger:) ⇒ Broker

Returns a new instance of Broker.



12
13
14
15
16
 
# File 'lib/kafka/broker.rb', line 12

def initialize(connection:, node_id: nil, logger:)
  @connection = connection
  @node_id = node_id
  @logger = logger
end

Class Method Details

.connect(node_id: nil, logger:, **options) ⇒ Object 



7
8
9
10
 
# File 'lib/kafka/broker.rb', line 7

def self.connect(node_id: nil, logger:, **options)
  connection = Connection.new(logger: logger, **options)
  new(connection: connection, node_id: node_id, logger: logger)
end

Instance Method Details

#disconnect ⇒ nil

Returns:

  • (nil) 


24
25
26
 
# File 'lib/kafka/broker.rb', line 24

def disconnect
  @connection.close
end

#fetch_messages(**options) ⇒ Kafka::Protocol::FetchResponse

Fetches messages from a specified topic and partition.

Parameters:

  • max_wait_time (Integer)
  • min_bytes (Integer)
  • topics (Hash)

Returns:



42
43
44
45
46
 
# File 'lib/kafka/broker.rb', line 42

def fetch_messages(**options)
  request = Protocol::FetchRequest.new(**options)

  @connection.send_request(request)
end

#fetch_metadata(**options) ⇒ Kafka::Protocol::MetadataResponse

Fetches cluster metadata from the broker.

Parameters:

  • topics (Array<String>)

Returns:



32
33
34
35
36
 
# File 'lib/kafka/broker.rb', line 32

def fetch_metadata(**options)
  request = Protocol::TopicMetadataRequest.new(**options)

  @connection.send_request(request)
end

#list_offsets(**options) ⇒ Kafka::Protocol::ListOffsetResponse

Lists the offset of the specified topics and partitions.

Parameters:

  • topics (Hash)

Returns:



52
53
54
55
56
 
# File 'lib/kafka/broker.rb', line 52

def list_offsets(**options)
  request = Protocol::ListOffsetRequest.new(**options)

  @connection.send_request(request)
end

#produce(**options) ⇒ Kafka::Protocol::ProduceResponse

Produces a set of messages to the broker.

Parameters:

  • required_acks (Integer)
  • timeout (Integer)
  • messages_for_topics (Hash)

Returns:



62
63
64
65
66
 
# File 'lib/kafka/broker.rb', line 62

def produce(**options)
  request = Protocol::ProduceRequest.new(**options)

  @connection.send_request(request)
end

#to_s ⇒ String

Returns:

  • (String) 


19
20
21
 
# File 'lib/kafka/broker.rb', line 19

def to_s
  "#{@connection} (node_id=#{@node_id.inspect})"
end