Class: Kafka::Broker
- Inherits:
-
Object
- Object
- Kafka::Broker
- Defined in:
- lib/kafka/broker.rb
Instance Method Summary collapse
- #commit_offsets(**options) ⇒ Object
- #disconnect ⇒ nil
-
#fetch_messages(**options) ⇒ Kafka::Protocol::FetchResponse
Fetches messages from a specified topic and partition.
-
#fetch_metadata(**options) ⇒ Kafka::Protocol::MetadataResponse
Fetches cluster metadata from the broker.
- #fetch_offsets(**options) ⇒ Object
- #find_group_coordinator(**options) ⇒ Object
- #heartbeat(**options) ⇒ Object
-
#initialize(connection:, node_id: nil, logger:) ⇒ Broker
constructor
A new instance of Broker.
- #join_group(**options) ⇒ Object
- #leave_group(**options) ⇒ Object
-
#list_offsets(**options) ⇒ Kafka::Protocol::ListOffsetResponse
Lists the offset of the specified topics and partitions.
-
#produce(**options) ⇒ Kafka::Protocol::ProduceResponse
Produces a set of messages to the broker.
- #sync_group(**options) ⇒ Object
- #to_s ⇒ String
Constructor Details
#initialize(connection:, node_id: nil, logger:) ⇒ Broker
Returns a new instance of Broker.
7 8 9 10 11 |
# File 'lib/kafka/broker.rb', line 7 def initialize(connection:, node_id: nil, logger:) @connection = connection @node_id = node_id @logger = logger end |
Instance Method Details
#commit_offsets(**options) ⇒ Object
69 70 71 72 73 |
# File 'lib/kafka/broker.rb', line 69 def commit_offsets(**) request = Protocol::OffsetCommitRequest.new(**) @connection.send_request(request) end |
#disconnect ⇒ nil
19 20 21 |
# File 'lib/kafka/broker.rb', line 19 def disconnect @connection.close end |
#fetch_messages(**options) ⇒ Kafka::Protocol::FetchResponse
Fetches messages from a specified topic and partition.
37 38 39 40 41 |
# File 'lib/kafka/broker.rb', line 37 def (**) request = Protocol::FetchRequest.new(**) @connection.send_request(request) end |
#fetch_metadata(**options) ⇒ Kafka::Protocol::MetadataResponse
Fetches cluster metadata from the broker.
27 28 29 30 31 |
# File 'lib/kafka/broker.rb', line 27 def (**) request = Protocol::TopicMetadataRequest.new(**) @connection.send_request(request) end |
#fetch_offsets(**options) ⇒ Object
63 64 65 66 67 |
# File 'lib/kafka/broker.rb', line 63 def fetch_offsets(**) request = Protocol::OffsetFetchRequest.new(**) @connection.send_request(request) end |
#find_group_coordinator(**options) ⇒ Object
93 94 95 96 97 |
# File 'lib/kafka/broker.rb', line 93 def find_group_coordinator(**) request = Protocol::GroupCoordinatorRequest.new(**) @connection.send_request(request) end |
#heartbeat(**options) ⇒ Object
99 100 101 102 103 |
# File 'lib/kafka/broker.rb', line 99 def heartbeat(**) request = Protocol::HeartbeatRequest.new(**) @connection.send_request(request) end |
#join_group(**options) ⇒ Object
75 76 77 78 79 |
# File 'lib/kafka/broker.rb', line 75 def join_group(**) request = Protocol::JoinGroupRequest.new(**) @connection.send_request(request) end |
#leave_group(**options) ⇒ Object
87 88 89 90 91 |
# File 'lib/kafka/broker.rb', line 87 def leave_group(**) request = Protocol::LeaveGroupRequest.new(**) @connection.send_request(request) end |
#list_offsets(**options) ⇒ Kafka::Protocol::ListOffsetResponse
Lists the offset of the specified topics and partitions.
47 48 49 50 51 |
# File 'lib/kafka/broker.rb', line 47 def list_offsets(**) request = Protocol::ListOffsetRequest.new(**) @connection.send_request(request) end |
#produce(**options) ⇒ Kafka::Protocol::ProduceResponse
Produces a set of messages to the broker.
57 58 59 60 61 |
# File 'lib/kafka/broker.rb', line 57 def produce(**) request = Protocol::ProduceRequest.new(**) @connection.send_request(request) end |
#sync_group(**options) ⇒ Object
81 82 83 84 85 |
# File 'lib/kafka/broker.rb', line 81 def sync_group(**) request = Protocol::SyncGroupRequest.new(**) @connection.send_request(request) end |
#to_s ⇒ String
14 15 16 |
# File 'lib/kafka/broker.rb', line 14 def to_s "#{@connection} (node_id=#{@node_id.inspect})" end |