Packages

kafka_ex_tc

0.12.1-11-debug
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.

HexDocs

Current section

Files

Jump to
Readme
130 Versions
3 Dependencies
Dependants
Files
Activity
kafka_ex_tc lib kafka_ex protocol join_group.ex
Raw

lib/kafka_ex/protocol/join_group.ex 

defmodule KafkaEx.Protocol.JoinGroup do

  import KafkaEx.Protocol.Common



  @moduledoc """

  Implementation of the Kafka JoinGroup request and response APIs

  """



  @protocol_type "consumer"

  @strategy_name "assign"

  @metadata_version 0



  defmodule Request do

    @moduledoc false

    defstruct member_id: nil, group_name: nil, topics: nil, session_timeout: nil



    @type t :: %Request{

            member_id: binary,

            group_name: binary,

            topics: [binary],

            session_timeout: integer

          }

  end



  defmodule Response do

    @moduledoc false

    defstruct error_code: nil,

              generation_id: 0,

              leader_id: nil,

              member_id: nil,

              members: []



    @type t ::

            %Response{

              error_code: atom | integer,

              generation_id: integer,

              leader_id: binary,

              member_id: binary,

              members: [binary]

            }

            | {:error, atom}



    def leader?(%__MODULE__{member_id: member_id, leader_id: leader_id}) do

      member_id == leader_id

    end

  end



  @spec create_request(integer, binary, Request.t()) :: binary

  def create_request(correlation_id, client_id, %Request{} = join_group_req) do

    metadata =

      <<@metadata_version::16-signed, length(join_group_req.topics)::32-signed,

        topic_data(join_group_req.topics)::binary, 0::32-signed>>



    KafkaEx.Protocol.create_request(:join_group, correlation_id, client_id) <>

      <<

        byte_size(join_group_req.group_name)::16-signed,

        join_group_req.group_name::binary,

        join_group_req.session_timeout::32-signed,

        byte_size(join_group_req.member_id)::16-signed,

        join_group_req.member_id::binary,

        byte_size(@protocol_type)::16-signed,

        @protocol_type::binary,

        # We always have just one GroupProtocl

        1::32-signed,

        byte_size(@strategy_name)::16-signed,

        @strategy_name::binary,

        byte_size(metadata)::32-signed,

        metadata::binary

      >>

  end



  @spec parse_response(binary) :: Response.t()

  def parse_response(

        <<_correlation_id::32-signed, error_code::16-signed,

          generation_id::32-signed, protocol_len::16-signed,

          _protocol::size(protocol_len)-binary, leader_len::16-signed,

          leader::size(leader_len)-binary, member_id_len::16-signed,

          member_id::size(member_id_len)-binary, members_size::32-signed,

          rest::binary>>

      ) do

    members = parse_members(members_size, rest, [])



    %Response{

      error_code: KafkaEx.Protocol.error(error_code),

      generation_id: generation_id,

      leader_id: leader,

      member_id: member_id,

      members: members

    }

  end



  defp parse_members(0, <<>>, members), do: members



  defp parse_members(

         size,

         <<member_len::16-signed, member::size(member_len)-binary,

           meta_len::32-signed, _metadata::size(meta_len)-binary,

           rest::binary>>,

         members

       ) do

    parse_members(size - 1, rest, [member | members])

  end

end