Packages
kafka_ex_tc
0.13.0
0.13.0
0.12.1
0.12.1-51-1
0.12.1-50-3
0.12.1-50-2
0.12.1-50-1
0.12.1-49-2
0.12.1-49-1
0.12.1-49-0
0.12.1-48-2
0.12.1-48-1
0.12.1-47-1
0.12.1-46-2
0.12.1-45-2
0.12.1-45-1
0.12.1-44-1
0.12.1-43-1
0.12.1-42-8
0.12.1-42-6
0.12.1-42-5
0.12.1-42-4
0.12.1-42-1
0.12.1-41-2
0.12.1-41-1
0.12.1-40-9
0.12.1-40-8
0.12.1-40-7
0.12.1-40-6
0.12.1-40-5
0.12.1-40-4
0.12.1-40-3
0.12.1-40-2
0.12.1-40-10
0.12.1-40-1
0.12.1-39-2
0.12.1-39-1
0.12.1-36-3
0.12.1-36-2
0.12.1-36-1
0.12.1-34-b
0.12.1-34-a
0.12.1-33-1
0.12.1-32-a1
0.12.1-29-s-3
0.12.1-29-s-2
0.12.1-29-s-1
0.12.1-28-rumble-1
0.12.1-27-2
0.12.1-27-1
0.12.1-26-rumble-2
0.12.1-26-rumble-1
0.12.1-26-dev-1
0.12.1-26-dev.1
0.12.1-26-3
0.12.1-26-2
0.12.1-26-1
0.12.1-25-s-8
0.12.1-25-s-7
0.12.1-25-s-6
0.12.1-25-s-5
0.12.1-25-s-4
0.12.1-25-s-3
0.12.1-25-s-2
0.12.1-25-s-1
0.12.1-25-dev-22
0.12.1-25-dev
0.12.1-24-2
0.12.1-24-1
0.12.1-22-poc-b-7
0.12.1-20-poc-b-6
0.12.1-20-poc-b-5
0.12.1-20-poc-b-4
0.12.1-20-poc-b-3
0.12.1-19-poc-b-3
0.12.1-19-poc-b-2
0.12.1-19-poc-b-1
0.12.1-19-poc-7
0.12.1-19-poc-6
0.12.1-19-poc-4
0.12.1-19-poc-3
0.12.1-19-poc-2
0.12.1-19-poc-1
0.12.1-19-poc
0.12.1-11-debug-1
0.12.1-11-debug
0.12.1-39
0.12.1-38.a
0.12.1-38
0.12.1-37.c
0.12.1-37.b
0.12.1-37.a
0.12.1-37.1
0.12.1-37
0.12.1-35.b
0.12.1-35.a
0.12.1-34
0.12.1-33
0.12.1-32
0.12.1-31
0.12.1-30
0.12.1-29
0.12.1-28
0.12.1-27
0.12.1-26
0.12.1-25
0.12.1-24
0.12.1-23
0.12.1-22.1
0.12.1-22
0.12.1-21
0.12.1-20
0.12.1-19
0.12.1-18
0.12.1-17
0.12.1-16
0.12.1-15
0.12.1-14
0.12.1-13
0.12.1-12
0.12.1-11
0.12.1-10
0.12.1-9
0.12.1-8
0.12.1-7
0.12.1-6
0.12.1-5
0.12.1-4
0.12.1-3
0.12.1-2
0.12.1-1
Kafka client for Elixir/Erlang.
Current section
Files
Jump to
Current section
Files
lib/kafka_ex/protocol/api_versions.ex
defmodule KafkaEx.Protocol.ApiVersions do
alias KafkaEx.Protocol
# the ApiVersions message can also, itself, have different api versions
@default_this_api_version 0
@default_throttle_time 0
@moduledoc """
Implementation of the Kafka ApiVersions request and response APIs
See: https://kafka.apache.org/protocol.html#The_Messages_ApiVersions
Reference implementation in Java:
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsRequest.java
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
"""
defmodule Response do
@moduledoc false
defstruct error_code: nil, api_versions: nil, throttle_time_ms: nil
@type t :: %Response{
error_code: atom,
api_versions: [ApiVersion],
throttle_time_ms: integer
}
end
defmodule ApiVersion do
@moduledoc false
defstruct api_key: nil, min_version: nil, max_version: nil
@type t :: %ApiVersion{
api_key: integer,
min_version: integer,
max_version: integer
}
end
@spec create_request(integer, binary, integer) :: binary
def create_request(
correlation_id,
client_id,
this_api_version \\ @default_this_api_version
)
def create_request(correlation_id, client_id, 1),
do: create_request(correlation_id, client_id, 0)
def create_request(correlation_id, client_id, 2),
do: create_request(correlation_id, client_id, 0)
def create_request(correlation_id, client_id, 0) do
Protocol.create_request(:api_versions, correlation_id, client_id)
end
@spec parse_response(binary, integer) :: Response.t()
def parse_response(binary, this_api_version \\ @default_this_api_version)
def parse_response(
<<_correlation_id::32-signed, error_code::16-signed,
api_versions_count::32-signed, rest::binary>>,
this_api_version
) do
%{
parse_rest_of_response(api_versions_count, rest, this_api_version)
| error_code: Protocol.error(error_code)
}
end
def parse_response(nil, _this_api_version) do
%Response{error_code: :no_response}
end
defp parse_rest_of_response(api_versions_count, data, this_api_version) do
{api_versions, remaining_data} =
Protocol.Common.read_array(
api_versions_count,
data,
&parse_one_api_version/1
)
case {this_api_version, remaining_data} do
{0, ""} ->
%Response{
api_versions: api_versions,
throttle_time_ms: @default_throttle_time
}
{v, <<throttle_time_ms::32-signed>>} when v in [1, 2] ->
%Response{
api_versions: api_versions,
throttle_time_ms: throttle_time_ms
}
end
end
defp parse_one_api_version(
<<api_key::16-signed, min_version::16-signed, max_version::16-signed,
rest::binary>>
) do
{%ApiVersion{
api_key: api_key,
min_version: min_version,
max_version: max_version
}, rest}
end
end