Class: Kafka::Broker
- Inherits:
-
Object
- Object
- Kafka::Broker
- Defined in:
- lib/kafka/broker.rb
Class Method Summary collapse
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.
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:, **) connection = Connection.new(logger: logger, **) new(connection: connection, node_id: node_id, logger: logger) end |
Instance Method Details
#commit_offsets(**options) ⇒ Object
74 75 76 77 78 |
# File 'lib/kafka/broker.rb', line 74 def commit_offsets(**) request = Protocol::OffsetCommitRequest.new(**) @connection.send_request(request) end |
#disconnect ⇒ 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.
42 43 44 45 46 |
# File 'lib/kafka/broker.rb', line 42 def (**) request = Protocol::FetchRequest.new(**) @connection.send_request(request) end |
#fetch_metadata(**options) ⇒ Kafka::Protocol::MetadataResponse
Fetches cluster metadata from the broker.
32 33 34 35 36 |
# File 'lib/kafka/broker.rb', line 32 def (**) request = Protocol::TopicMetadataRequest.new(**) @connection.send_request(request) end |
#fetch_offsets(**options) ⇒ Object
68 69 70 71 72 |
# File 'lib/kafka/broker.rb', line 68 def fetch_offsets(**) request = Protocol::OffsetFetchRequest.new(**) @connection.send_request(request) end |
#find_group_coordinator(**options) ⇒ Object
98 99 100 101 102 |
# File 'lib/kafka/broker.rb', line 98 def find_group_coordinator(**) request = Protocol::GroupCoordinatorRequest.new(**) @connection.send_request(request) end |
#heartbeat(**options) ⇒ Object
104 105 106 107 108 |
# File 'lib/kafka/broker.rb', line 104 def heartbeat(**) request = Protocol::HeartbeatRequest.new(**) @connection.send_request(request) end |
#join_group(**options) ⇒ Object
80 81 82 83 84 |
# File 'lib/kafka/broker.rb', line 80 def join_group(**) request = Protocol::JoinGroupRequest.new(**) @connection.send_request(request) end |
#leave_group(**options) ⇒ Object
92 93 94 95 96 |
# File 'lib/kafka/broker.rb', line 92 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.
52 53 54 55 56 |
# File 'lib/kafka/broker.rb', line 52 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.
62 63 64 65 66 |
# File 'lib/kafka/broker.rb', line 62 def produce(**) request = Protocol::ProduceRequest.new(**) @connection.send_request(request) end |
#sync_group(**options) ⇒ Object
86 87 88 89 90 |
# File 'lib/kafka/broker.rb', line 86 def sync_group(**) request = Protocol::SyncGroupRequest.new(**) @connection.send_request(request) end |
#to_s ⇒ String
19 20 21 |
# File 'lib/kafka/broker.rb', line 19 def to_s "#{@connection} (node_id=#{@node_id.inspect})" end |