misha_cafex

defmodule Cafex.Protocol.JoinGroup do
  use Cafex.Protocol, api: :join_group

  defrequest do
    field :group_id, binary
    field :session_timeout, integer
    field :member_id, binary
    field :protocol_type, binary
    field :group_protocols, [group_protocol]

    @type group_protocol :: {name :: binary, protocol_metadata}
    @type protocol_metadata :: {version :: integer, subscription :: [topic], user_data :: binary}
    @type topic :: binary
  end

  defresponse do
    field :error, Cafex.Protocol.error
    field :generation_id, binary
    field :group_protocol, binary
    field :leader_id, binary
    field :member_id, binary
    field :members, [member]

    @type member :: {id :: binary, metadata :: Request.protocol_metadata}
  end

  def encode(%{group_id: group_id,
               session_timeout: timeout,
               member_id: member_id,
               protocol_type: protocol_type,
               group_protocols: protocols}) do
    [encode_string(group_id),
     << timeout :: 32-signed >>,
     encode_string(member_id),
     encode_string(protocol_type),
     encode_array(protocols, &encode_group_protocol/1)]
    |> IO.iodata_to_binary
  end

  def decode(<< error_code :: 16-signed,
                generation_id :: 32-signed,
                group_protocol_len :: 16-signed,
                group_protocol :: size(group_protocol_len)-binary,
                leader_id_len :: 16-signed,
                leader_id :: size(leader_id_len)-binary,
                member_id_len :: 16-signed,
                member_id :: size(member_id_len)-binary,
                rest :: binary >>) do
    {members, _} = decode_array(rest, &parse_member/1)
    %Response{error: decode_error(error_code),
              generation_id: generation_id,
              group_protocol: group_protocol,
              leader_id: leader_id,
              member_id: member_id,
              members: members}
  end

  defp encode_group_protocol({name, metadata}) do
    [encode_string(name),
     encode_group_protocol_metadata(metadata)]
  end

  defp parse_member(<< member_id_len :: 16-signed,
                       member_id :: size(member_id_len)-binary,
                       metadata_len :: 32-signed,
                       metadata :: size(metadata_len)-binary,
                       rest :: binary>>) do
    metadata = parse_group_protocol_metadata(metadata)
    {{member_id, metadata}, rest}
  end
end