Sha256: fbd178a57bfd55b927876aa7a2c7d6ac76a2a71ed16d6f34b5da8581de7e43bd
Contents?: true
Size: 1.83 KB
Versions: 33
Compression:
Stored size: 1.83 KB
Contents
# frozen_string_literal: true
module Kafka
module Protocol
# A request to fetch messages from a given partition.
#
# ## API Specification
#
# FetchRequest => ReplicaId MaxWaitTime MinBytes MaxBytes IsolationLevel [TopicName [Partition FetchOffset MaxBytes]]
# ReplicaId => int32
# MaxWaitTime => int32
# MinBytes => int32
# MaxBytes => int32
# IsolationLevel => int8
# TopicName => string
# Partition => int32
# FetchOffset => int64
# MaxBytes => int32
#
class FetchRequest
ISOLATION_READ_UNCOMMITTED = 0
ISOLATION_READ_COMMITTED = 1
# @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
4
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_int8(ISOLATION_READ_COMMITTED)
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
33 entries across 33 versions & 4 rubygems