Packages
kafka_ex_tc
0.12.1
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/offset.ex
defmodule KafkaEx.Protocol.Offset do
alias KafkaEx.Protocol
@moduledoc """
Implementation of the Kafka Offset request and response APIs
"""
defmodule Request do
@moduledoc false
defstruct replica_id: -1,
topic_name: nil,
partition: nil,
time: -1,
max_number_of_offsets: 1
@type t :: %Request{
replica_id: integer,
topic_name: binary,
partition: integer,
time: integer,
max_number_of_offsets: integer
}
end
defmodule Response do
@moduledoc false
defstruct topic: nil, partition_offsets: []
@type t :: %Response{topic: binary, partition_offsets: list}
def extract_offset([%__MODULE__{partition_offsets: [%{offset: [offset]}]}]),
do: offset
def extract_offset([%__MODULE__{partition_offsets: [%{offset: []}]}]), do: 0
end
def create_request(correlation_id, client_id, topic, partition, time) do
KafkaEx.Protocol.create_request(:offset, correlation_id, client_id) <>
<<-1::32-signed, 1::32-signed, byte_size(topic)::16-signed, topic::binary,
1::32-signed, partition::32-signed, parse_time(time)::64, 1::32>>
end
def parse_response(
<<_correlation_id::32-signed, num_topics::32-signed, rest::binary>>
),
do: parse_topics(num_topics, rest)
@spec parse_time(:latest | :earliest | :calendar.datetime()) :: integer
def parse_time(:latest), do: -1
def parse_time(:earliest), do: -2
def parse_time(time) do
current_time_in_seconds = :calendar.datetime_to_gregorian_seconds(time)
unix_epoch_in_seconds =
:calendar.datetime_to_gregorian_seconds({{1970, 1, 1}, {0, 0, 0}})
(current_time_in_seconds - unix_epoch_in_seconds) * 1000
end
defp parse_topics(0, _), do: []
defp parse_topics(
topics_size,
<<topic_size::16-signed, topic::size(topic_size)-binary,
partitions_size::32-signed, rest::binary>>
) do
{partitions, topics_data} = parse_partitions(partitions_size, rest)
[
%Response{topic: topic, partition_offsets: partitions}
| parse_topics(topics_size - 1, topics_data)
]
end
defp parse_partitions(partitions_size, rest, partitions \\ [])
defp parse_partitions(0, rest, partitions), do: {partitions, rest}
defp parse_partitions(
partitions_size,
<<partition::32-signed, error_code::16-signed, offsets_size::32-signed,
rest::binary>>,
partitions
) do
{offsets, rest} = parse_offsets(offsets_size, rest)
parse_partitions(partitions_size - 1, rest, [
%{
partition: partition,
error_code: Protocol.error(error_code),
offset: offsets
}
| partitions
])
end
defp parse_offsets(offsets_size, rest, offsets \\ [])
defp parse_offsets(0, rest, offsets), do: {Enum.reverse(offsets), rest}
defp parse_offsets(offsets_size, <<offset::64-signed, rest::binary>>, offsets) do
parse_offsets(offsets_size - 1, rest, [offset | offsets])
end
end