Sha256: de56d3fd0cc3384832009899227c54657bfea4a252bf650105ba8fec4562e375
Contents?: true
Size: 1.04 KB
Versions: 49
Compression:
Stored size: 1.04 KB
Contents
# frozen_string_literal: true
module Kafka
module Protocol
class CreateTopicsRequest
def initialize(topics:, timeout:)
@topics, @timeout = topics, timeout
end
def api_key
CREATE_TOPICS_API
end
def api_version
0
end
def response_class
Protocol::CreateTopicsResponse
end
def encode(encoder)
encoder.write_array(@topics) do |topic, config|
encoder.write_string(topic)
encoder.write_int32(config.fetch(:num_partitions))
encoder.write_int16(config.fetch(:replication_factor))
# Replica assignments. We don't care.
encoder.write_array([])
encoder.write_array(config.fetch(:config)) do |config_name, config_value|
config_value = config_value.to_s unless config_value.nil?
encoder.write_string(config_name)
encoder.write_string(config_value)
end
end
# Timeout is in ms.
encoder.write_int32(@timeout * 1000)
end
end
end
end
Version data entries
49 entries across 49 versions & 4 rubygems