misha_cafex

defmodule Cafex.Protocol.SyncGroup do
  use Cafex.Protocol, api: :sync_group

  defrequest do
    field :group_id, binary
    field :generation_id, integer
    field :member_id, binary
    field :group_assignment, [{member_id, member_assignment}]
    @type member_id :: binary
    @type member_assignment :: {version :: integer, [partition_assignment], user_data :: binary}
    @type partition_assignment :: {topic, [partition]}
    @type topic :: binary
    @type partition :: integer
  end

  defresponse do
    field :error, Cafex.Protocol.error
    field :member_assignment, Request.member_assignment
  end

  def encode(%{group_id: group_id,
              generation_id: generation_id,
              member_id: member_id,
              group_assignment: assignment}) do
    [encode_string(group_id),
     << generation_id :: 32-signed >>,
     encode_string(member_id),
     encode_array(assignment, &encode_member_assignment/1)]
    |> IO.iodata_to_binary
  end

  def decode(<< error_code :: 16-signed,
                assignment_size :: 32-signed,
                assignment :: size(assignment_size)-binary>>) do

    error = decode_error(error_code)
    assignment = case error do
      :no_error -> parse_assignment(assignment)
      _ -> nil
    end
    %Response{error: error, member_assignment: assignment}
  end

  def encode_member_assignment({member_id, assignment}) do
    [encode_string(member_id), encode_assignment(assignment)]
  end
end