Sha256: 90d135f13751f3bf5f22ca3e8724831541b98604ed7dd4e5afc7852a077cd422
Contents?: true
Size: 1.59 KB
Versions: 15
Compression:
Stored size: 1.59 KB
Contents
module Kafka
module Protocol
# A request to fetch messages from a given partition.
#
# ## API Specification
#
# FetchRequest => ReplicaId MaxWaitTime MinBytes [TopicName [Partition FetchOffset MaxBytes]]
# ReplicaId => int32
# MaxWaitTime => int32
# MinBytes => int32
# TopicName => string
# Partition => int32
# FetchOffset => int64
# MaxBytes => int32
#
class FetchRequest
# @param max_wait_time [Integer]
# @param min_bytes [Integer]
# @param topics [Hash]
def initialize(max_wait_time:, min_bytes:, max_bytes:, topics:)
@replica_id = REPLICA_ID
@max_wait_time = max_wait_time
@min_bytes = min_bytes
@max_bytes = max_bytes
@topics = topics
end
def api_key
FETCH_API
end
def api_version
3
end
def response_class
Protocol::FetchResponse
end
def encode(encoder)
encoder.write_int32(@replica_id)
encoder.write_int32(@max_wait_time)
encoder.write_int32(@min_bytes)
encoder.write_int32(@max_bytes)
encoder.write_array(@topics) do |topic, partitions|
encoder.write_string(topic)
encoder.write_array(partitions) do |partition, config|
fetch_offset = config.fetch(:fetch_offset)
max_bytes = config.fetch(:max_bytes)
encoder.write_int32(partition)
encoder.write_int64(fetch_offset)
encoder.write_int32(max_bytes)
end
end
end
end
end
end
Version data entries
15 entries across 15 versions & 1 rubygems