kafka_ex

defmodule KafkaEx.Protocol.OffsetFetch do
  alias KafkaEx.Protocol
  import KafkaEx.Protocol.Common

  @moduledoc """
  Implementation of the Kafka OffsetFetch request and response APIs
  """

  defmodule Request do
    @moduledoc false
    defstruct consumer_group: nil, topic: nil, partition: nil
    @type t :: %Request{
      consumer_group: nil | binary,
      topic: binary,
      partition: integer
    }
  end

  defmodule Response do
    @moduledoc false
    defstruct topic: nil, partitions: []
    @type t :: %Response{topic: binary, partitions: list}

    def last_offset(:topic_not_found) do
      0
    end

    def last_offset(offset_fetch_data) do
      case offset_fetch_data do
        [] -> 0
        _  -> partitions = offset_fetch_data |> hd |> Map.get(:partitions, [])
          case partitions do
            [] -> 0
            _  -> partitions |> hd |> Map.get(:offset, 0)
          end
      end
    end
  end

  def create_request(correlation_id, client_id, offset_fetch_request) do
    KafkaEx.Protocol.create_request(:offset_fetch, correlation_id, client_id) <> << byte_size(offset_fetch_request.consumer_group) :: 16-signed, offset_fetch_request.consumer_group :: binary, 1 :: 32-signed, byte_size(offset_fetch_request.topic) :: 16-signed, offset_fetch_request.topic :: binary, 1 :: 32-signed, offset_fetch_request.partition :: 32 >>
  end

  def parse_response(<< _correlation_id :: 32-signed, topics_size :: 32-signed, topics_data :: binary >>) do
    parse_topics(topics_size, topics_data, __MODULE__)
  end

  def parse_partitions(0, rest, partitions), do: {partitions, rest}
  def parse_partitions(partitions_size, << partition :: 32-signed, offset :: 64-signed, metadata_size :: 16-signed, metadata :: size(metadata_size)-binary, error_code :: 16-signed, rest :: binary >>, partitions) do
    parse_partitions(partitions_size - 1, rest, [%{partition: partition, offset: offset, metadata: metadata, error_code: Protocol.error(error_code)} | partitions])
  end
end