defmodule Aerospike.Protocol.Response do @moduledoc false alias Aerospike.Error alias Aerospike.Key alias Aerospike.Protocol.AsmMsg alias Aerospike.Protocol.AsmMsg.Field alias Aerospike.Protocol.AsmMsg.Operation alias Aerospike.Protocol.AsmMsg.Value alias Aerospike.Protocol.Batch alias Aerospike.Protocol.BatchRead alias Aerospike.Protocol.ResultCode alias Aerospike.Record @spec decode_as_msg(binary()) :: {:ok, AsmMsg.t()} | {:error, Error.t()} def decode_as_msg(body) when is_binary(body) do case AsmMsg.decode(body) do {:ok, _} = ok -> ok {:error, reason} -> {:error, Error.from_result_code(:parse_error, message: "failed to decode AS_MSG reply: #{reason}")} end end @doc """ Parses a successful write or header-only reply into record metadata or an error. """ @spec parse_record_metadata_response(AsmMsg.t()) :: {:ok, Record.metadata()} | {:error, Error.t()} def parse_record_metadata_response(%AsmMsg{} = msg) do case result_atom(msg.result_code) do {:ok, :ok} -> {:ok, record_metadata(msg)} other -> error_from_result(other) end end @doc """ Parses a successful read response into a `%Record{}` or an error. """ @spec parse_record_response(AsmMsg.t(), Key.t()) :: {:ok, Record.t()} | {:error, Error.t()} def parse_record_response(%AsmMsg{} = msg, %Key{} = key) do case result_atom(msg.result_code) do {:ok, :ok} -> with {:ok, bins} <- record_bins_from_operations(msg.operations) do {:ok, %Record{ key: key, bins: bins, generation: msg.generation, ttl: msg.expiration }} end other -> error_from_result(other) end end @doc """ Parses a single-record UDF reply into its returned value or a typed error. """ @spec parse_udf_response(AsmMsg.t()) :: {:ok, term()} | {:error, Error.t()} def parse_udf_response(%AsmMsg{} = msg) do case result_atom(msg.result_code) do {:ok, :ok} -> with {:ok, bins} <- record_bins_from_operations(msg.operations) do extract_udf_value(bins) end {:ok, :udf_bad_response} -> extract_udf_error(msg.operations) other -> error_from_result(other) end end @doc """ Parses a multi-record batch read/header reply into per-record indexed results. """ @spec parse_batch_response(binary(), Aerospike.Command.BatchCommand.NodeRequest.t()) :: {:ok, Batch.Reply.t()} | {:error, Error.t()} def parse_batch_response(body, node_request) when is_binary(body) do Batch.parse_response(body, node_request) end @doc """ Parses a multi-record batch read/header reply into per-record indexed results. """ @spec parse_batch_read_response( binary(), Aerospike.Command.BatchCommand.NodeRequest.t(), keyword() ) :: {:ok, BatchRead.Reply.t()} | {:error, Error.t()} def parse_batch_read_response(body, node_request, opts \\ []) when is_binary(body) and is_list(opts) do BatchRead.parse_response(body, node_request, opts) end @spec record_metadata(AsmMsg.t()) :: Record.metadata() def record_metadata(%AsmMsg{} = msg) do %{generation: msg.generation, ttl: msg.expiration} end @doc """ Extracts the record version from the fields of an AS_MSG response. """ @spec extract_record_version(AsmMsg.t()) :: {:ok, non_neg_integer()} | :none def extract_record_version(%AsmMsg{fields: fields}) do rv_type = Field.type_record_version() case Enum.find(fields, fn field -> field.type == rv_type end) do %Field{data: <>} -> {:ok, version} nil -> :none end end @doc """ Extracts the MRT deadline from the fields of an AS_MSG response. """ @spec extract_mrt_deadline(AsmMsg.t()) :: {:ok, integer()} | :none def extract_mrt_deadline(%AsmMsg{fields: fields}) do dl_type = Field.type_mrt_deadline() case Enum.find(fields, fn field -> field.type == dl_type end) do %Field{data: <>} -> {:ok, deadline} nil -> :none end end defp record_bins_from_operations(operations) do {pairs, count} = decode_record_bin_pairs(operations, [], 0) bins = Map.new(pairs) case map_size(bins) do ^count -> {:ok, bins} _duplicates -> {:ok, repeated_record_bins(pairs)} end end defp decode_record_bin_pairs([], pairs, count), do: {pairs, count} defp decode_record_bin_pairs([%Operation{bin_name: ""} | rest], pairs, count) do decode_record_bin_pairs(rest, pairs, count) end defp decode_record_bin_pairs([%Operation{bin_name: name} = op | rest], pairs, count) do {:ok, value} = Value.decode_value(op.particle_type, op.data) decode_record_bin_pairs(rest, [{name, value} | pairs], count + 1) end defp repeated_record_bins(pairs) do pairs |> Enum.reverse() |> put_repeated_record_bins(%{}, nil) end defp put_repeated_record_bins([], bins, _repeated), do: bins defp put_repeated_record_bins([{name, value} | rest], bins, repeated) do {bins, repeated} = put_operation_value(bins, repeated, name, value) put_repeated_record_bins(rest, bins, repeated) end defp put_operation_value(bins, repeated, name, value) do case Map.fetch(bins, name) do {:ok, existing} -> put_repeated_operation_value(bins, repeated, name, existing, value) :error -> {Map.put(bins, name, value), repeated} end end defp put_repeated_operation_value(bins, repeated, name, existing, value) do if repeated != nil and MapSet.member?(repeated, name) do {Map.put(bins, name, existing ++ [value]), repeated} else {Map.put(bins, name, [existing, value]), put_repeated_name(repeated, name)} end end defp put_repeated_name(nil, name), do: MapSet.new([name]) defp put_repeated_name(repeated, name), do: MapSet.put(repeated, name) defp result_atom(code) when is_integer(code) do ResultCode.from_integer(code) end defp extract_udf_value(%{"SUCCESS" => value}), do: {:ok, value} defp extract_udf_value(%{"FAILURE" => message}) do {:error, Error.from_result_code(:udf_bad_response, message: to_string(message))} end defp extract_udf_value(bins) when map_size(bins) == 0, do: {:ok, nil} defp extract_udf_value(bins), do: {:ok, bins} defp extract_udf_error(operations) do case record_bins_from_operations(operations) do {:ok, %{"FAILURE" => message}} -> {:error, Error.from_result_code(:udf_bad_response, message: to_string(message))} _ -> {:error, Error.from_result_code(:udf_bad_response)} end end defp error_from_result({:ok, error_code}) do {:error, Error.from_result_code(error_code)} end defp error_from_result({:error, unknown}) do {:error, Error.from_result_code(:server_error, message: "unknown result code #{unknown}")} end end