misha_cafex

defmodule Cafex.Protocol.Metadata do
  use Cafex.Protocol, api: :metadata

  defrequest do
    field :topics, [default: []], [binary]
  end

  defresponse do
    field :brokers, [broker]
    field :topics, [topic]

    @type broker :: %{node_id: integer,
                      host: binary,
                      port: 0..65535}
    @type topic :: %{error: Cafex.Protocol.error,
                     name: binary,
                     partitions: [partition]}
    @type partition :: %{error: Cafex.Protocol.error,
                         partition_id: integer,
                         leader: integer,
                         replicas: [integer],
                         isrs: [integer]}
  end

  def encode(%Request{topics: topics}) do
    topics
    |> encode_array(&Cafex.Protocol.encode_string/1)
    |> IO.iodata_to_binary
  end

  @spec decode(binary) :: Response.t
  def decode(data) when is_binary(data) do
    {brokers, rest} = decode_array(data, &parse_broker/1)
    {topics,     _} = decode_array(rest, &parse_topic/1)
    %Response{brokers: brokers, topics: topics}
  end

  defp parse_broker(<< node_id :: 32-signed, host_len :: 16-signed,
                       host :: size(host_len)-binary, port :: 32-signed,
                       rest :: binary >>) do
    {%{node_id: node_id, host: host, port: port}, rest}
  end

  defp parse_topic(<< error_code :: 16-signed, topic_len :: 16-signed,
                      topic :: size(topic_len)-binary, rest :: binary >>) do
    {partitions, rest} = decode_array(rest, &parse_partition/1)
    {%{error: decode_error(error_code), name: topic, partitions: partitions}, rest}
  end

  defp parse_partition(<< error_code :: 16-signed, partition_id :: 32-signed,
                          leader :: 32-signed, rest :: binary >>) do
    {replicas, rest} = decode_array(rest, &parse_int32/1)
    {isrs,     rest} = decode_array(rest, &parse_int32/1)
    {%{error: decode_error(error_code),
       partition_id: partition_id,
       leader: leader,
       replicas: replicas,
       isrs: isrs}, rest}
  end

  defp parse_int32(<< value :: 32-signed, rest :: binary >>), do: {value, rest}
end