# DO NOT EDIT THIS FILE MANUALLY # This module is automatically generated by running mix task generate_file # every change must be done inside the mix task directly defmodule KlifeProtocol.Messages.<%= module_name %> do @moduledoc """ Kafka protocol <%= module_name %> message Request versions summary: <%= Enum.map(req_versions_comments, fn comment -> """ - #{Enum.join(comment, "\n")} """ end)%> Response versions summary: <%= Enum.map(res_versions_comments, fn comment -> """ - #{Enum.join(comment, "\n")} """ end)%> """ alias KlifeProtocol.Deserializer alias KlifeProtocol.Serializer alias KlifeProtocol.Header @api_key <%= api_key %> <%= case req_header_exceptions do {:fixed, _} -> "" _ -> if req_flexible_version == :none, do: "@min_flexible_version_req :none", else: "@min_flexible_version_req #{req_flexible_version}" end %> <%= case res_header_exceptions do {:fixed, _} -> "" _ -> if res_flexible_version == :none, do: "@min_flexible_version_res :none", else: "@min_flexible_version_res #{res_flexible_version}" end %> @doc """ Receives a map and serialize it to kafka wire format of the given version. Input content fields: <%= Enum.map(req_field_comments, fn {line, depth} -> if depth == 0 do line <> "\n" else base_space = " " extra_space = Enum.map(1..depth, fn _ -> base_space end) |> Enum.join() base_space <> extra_space <> line <> "\n" end end) %> """ def serialize_request(%{headers: headers, content: content}, version) do headers |> Map.put(:request_api_key, @api_key) |> Map.put(:request_api_version, version) |> Header.serialize_request(req_header_version(version)) |> then(&Serializer.execute(content, request_schema(version), &1)) end @doc """ Receive a binary in the kafka wire format and deserialize it into a map. Response content fields: <%= Enum.map(res_field_comments, fn {line, depth} -> if depth == 0 do line <> "\n" else base_space = " " extra_space = Enum.map(1..depth, fn _ -> base_space end) |> Enum.join() base_space <> extra_space <> line <> "\n" end end) %> """ def deserialize_response(data, version, with_header? \\ true) def deserialize_response(data, version, true) do {:ok, {headers, rest_data}} = Header.deserialize_response(data, res_header_version(version)) case Deserializer.execute(rest_data, response_schema(version)) do {:ok, {content, <<>>}} -> {:ok, %{headers: headers, content: content}} {:error, _reason} = err -> err end end def deserialize_response(data, version, false) do case Deserializer.execute(data, response_schema(version)) do {:ok, {content, <<>>}} -> {:ok, %{content: content}} {:error, _reason} = err -> err end end @doc """ Returns the message api key number. """ def api_key(), do: @api_key @doc """ Returns the current max supported version of this message. """ def max_supported_version(), do: <%= request_schemas |> Enum.map(&elem(&1, 0)) |> Enum.reject(&(&1 in version_exceptions)) |> Enum.max() %> @doc """ Returns the current min supported version of this message. """ def min_supported_version(), do: <%= request_schemas |> Enum.map(&elem(&1, 0)) |> Enum.reject(&(&1 in version_exceptions)) |> Enum.min() %> <% default_res_header_code = "if(msg_version >= @min_flexible_version_res, do: 1, else: 0)" default_req_header_code = "if(msg_version >= @min_flexible_version_req, do: 2, else: 1)" %> <%= case req_header_exceptions do {:fixed, v} -> "defp req_header_version(_msg_version), do: #{v}" {:versions, list, v} -> "defp req_header_version(msg_version) do if msg_version in #{inspect(list)}, do: #{v}, else: #{default_req_header_code} end " nil -> "defp req_header_version(msg_version), do: #{default_req_header_code}" end %> <%= case res_header_exceptions do {:fixed, v} -> "defp res_header_version(_msg_version), do: #{v}" {:versions, list, v} -> "defp res_header_version(msg_version) do if msg_version in #{inspect(list)}, do: #{v}, else: #{default_res_header_code} end " nil -> "defp res_header_version(msg_version), do: #{default_res_header_code}" end %> <%= Enum.map(request_schemas, fn {version, schema} -> if version not in version_exceptions do """ def request_schema(#{version}), do: #{inspect(schema, limit: :infinity)} """ else "" end end) %> def request_schema(unkown_version), do: raise("Unknown version #{unkown_version} for message <%= module_name %>") <%= Enum.map(response_schemas, fn {version, schema} -> if version not in version_exceptions do """ def response_schema(#{version}), do: #{inspect(schema, limit: :infinity)} """ else "" end end) %> def response_schema(unkown_version), do: raise("Unknown version #{unkown_version} for message <%= module_name %>") end