defmodule Aerospike.Protocol.AsmMsg do @moduledoc false import Bitwise alias Aerospike.Protocol.AsmMsg.Field alias Aerospike.Protocol.AsmMsg.Operation @msg_remaining_header_size 22 # Info1 flags @info1_read 0x01 @info1_get_all 0x02 @info1_short_query 0x04 @info1_batch 0x08 @info1_nobindata 0x20 @info1_read_mode_ap_all 0x40 @info1_compress_response 0x80 # Info2 flags @info2_write 0x01 @info2_delete 0x02 @info2_generation 0x04 @info2_generation_gt 0x08 @info2_durable_delete 0x10 @info2_create_only 0x20 @info2_relax_ap_long_query 0x40 @info2_respond_all_ops 0x80 # Info3 flags @info3_last 0x01 @info3_commit_master 0x02 @info3_partition_done 0x04 @info3_update_only 0x08 @info3_create_or_replace 0x10 @info3_replace_only 0x20 @info3_sc_read_type 0x40 @info3_sc_read_relax 0x80 # Info4/MRT flags @info4_mrt_verify_read 0x01 @info4_mrt_roll_forward 0x02 @info4_mrt_roll_back 0x04 @info4_mrt_on_locking_only 0x10 defstruct info1: 0, info2: 0, info3: 0, info4: 0, result_code: 0, generation: 0, expiration: 0, timeout: 0, fields: [], operations: [] @type t :: %__MODULE__{ info1: non_neg_integer(), info2: non_neg_integer(), info3: non_neg_integer(), info4: non_neg_integer(), result_code: non_neg_integer(), generation: non_neg_integer(), expiration: non_neg_integer(), timeout: integer(), fields: [Field.t()], operations: [Operation.t()] } # Flag accessor functions @doc "Returns the INFO1_READ flag value." @spec info1_read() :: 0x01 def info1_read, do: @info1_read @doc "Returns the INFO1_GET_ALL flag value." @spec info1_get_all() :: 0x02 def info1_get_all, do: @info1_get_all @doc "Returns the INFO1_SHORT_QUERY flag value." @spec info1_short_query() :: 0x04 def info1_short_query, do: @info1_short_query @doc "Returns the INFO1_BATCH flag value." @spec info1_batch() :: 0x08 def info1_batch, do: @info1_batch @doc "Returns the INFO1_NOBINDATA flag value." @spec info1_nobindata() :: 0x20 def info1_nobindata, do: @info1_nobindata @doc "Returns the INFO1_READ_MODE_AP_ALL flag value." @spec info1_read_mode_ap_all() :: 0x40 def info1_read_mode_ap_all, do: @info1_read_mode_ap_all @doc "Returns the INFO1_COMPRESS_RESPONSE flag value." @spec info1_compress_response() :: 0x80 def info1_compress_response, do: @info1_compress_response @doc "Returns the INFO2_WRITE flag value." @spec info2_write() :: 0x01 def info2_write, do: @info2_write @doc "Returns the INFO2_DELETE flag value." @spec info2_delete() :: 0x02 def info2_delete, do: @info2_delete @doc "Returns the INFO2_GENERATION flag value." @spec info2_generation() :: 0x04 def info2_generation, do: @info2_generation @doc "Returns the INFO2_GENERATION_GT flag value." @spec info2_generation_gt() :: 0x08 def info2_generation_gt, do: @info2_generation_gt @doc "Returns the INFO2_DURABLE_DELETE flag value." @spec info2_durable_delete() :: 0x10 def info2_durable_delete, do: @info2_durable_delete @doc "Returns the INFO2_CREATE_ONLY flag value." @spec info2_create_only() :: 0x20 def info2_create_only, do: @info2_create_only @doc "Returns the INFO2_RELAX_AP_LONG_QUERY flag value." @spec info2_relax_ap_long_query() :: 0x40 def info2_relax_ap_long_query, do: @info2_relax_ap_long_query @doc "Returns the INFO2_RESPOND_ALL_OPS flag value." @spec info2_respond_all_ops() :: 0x80 def info2_respond_all_ops, do: @info2_respond_all_ops @doc "Returns the INFO3_LAST flag value." @spec info3_last() :: 0x01 def info3_last, do: @info3_last @doc "Returns the INFO3_COMMIT_MASTER flag value." @spec info3_commit_master() :: 0x02 def info3_commit_master, do: @info3_commit_master @doc "Returns the INFO3_PARTITION_DONE flag value." @spec info3_partition_done() :: 0x04 def info3_partition_done, do: @info3_partition_done @doc "Returns the INFO3_UPDATE_ONLY flag value." @spec info3_update_only() :: 0x08 def info3_update_only, do: @info3_update_only @doc "Returns the INFO3_CREATE_OR_REPLACE flag value." @spec info3_create_or_replace() :: 0x10 def info3_create_or_replace, do: @info3_create_or_replace @doc "Returns the INFO3_REPLACE_ONLY flag value." @spec info3_replace_only() :: 0x20 def info3_replace_only, do: @info3_replace_only @doc "Returns the INFO3_SC_READ_TYPE flag value." @spec info3_sc_read_type() :: 0x40 def info3_sc_read_type, do: @info3_sc_read_type @doc "Returns the INFO3_SC_READ_RELAX flag value." @spec info3_sc_read_relax() :: 0x80 def info3_sc_read_relax, do: @info3_sc_read_relax @doc "Returns the INFO4_MRT_VERIFY_READ flag value." @spec info4_mrt_verify_read() :: 0x01 def info4_mrt_verify_read, do: @info4_mrt_verify_read @doc "Returns the INFO4_MRT_ROLL_FORWARD flag value." @spec info4_mrt_roll_forward() :: 0x02 def info4_mrt_roll_forward, do: @info4_mrt_roll_forward @doc "Returns the INFO4_MRT_ROLL_BACK flag value." @spec info4_mrt_roll_back() :: 0x04 def info4_mrt_roll_back, do: @info4_mrt_roll_back @doc "Returns the INFO4_MRT_ON_LOCKING_ONLY flag value." @spec info4_mrt_on_locking_only() :: 0x10 def info4_mrt_on_locking_only, do: @info4_mrt_on_locking_only @doc """ Encodes an AsmMsg struct into binary format. The encoding produces: - 22-byte AS_MSG header - Encoded fields (field_count fields) - Encoded operations (operation_count operations) Note: This does NOT include the 8-byte protocol header. ## Examples iex> msg = %Aerospike.Protocol.AsmMsg{info1: 1, info2: 0, info3: 0, info4: 0} iex> encoded = Aerospike.Protocol.AsmMsg.encode(msg) iex> byte_size(encoded) 22 """ @spec encode(t()) :: binary() def encode(%__MODULE__{} = msg) do field_count = length(msg.fields) op_count = length(msg.operations) header = <<@msg_remaining_header_size::8, msg.info1::8, msg.info2::8, msg.info3::8, msg.info4::8, msg.result_code::8, msg.generation::32-big, msg.expiration::32-big, msg.timeout::32-big-signed, field_count::16-big, op_count::16-big>> fields_binary = Enum.map_join(msg.fields, &Field.encode/1) ops_binary = Enum.map_join(msg.operations, &Operation.encode/1) header <> fields_binary <> ops_binary end @doc """ Decodes a binary AS_MSG (without the 8-byte protocol header) into an AsmMsg struct. Returns `{:ok, msg}` on success or `{:error, reason}` on failure. ## Examples iex> binary = <<22, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0>> iex> {:ok, msg} = Aerospike.Protocol.AsmMsg.decode(binary) iex> msg.info1 1 """ @spec decode(binary()) :: {:ok, t()} | {:error, atom()} def decode(<< @msg_remaining_header_size::8, info1::8, info2::8, info3::8, info4::8, result_code::8, generation::32-big, expiration::32-big, timeout::32-big-signed, field_count::16-big, op_count::16-big, rest::binary >>) do with {:ok, fields, rest} <- decode_fields(rest, field_count), {:ok, operations, _rest} <- decode_operations(rest, op_count) do {:ok, %__MODULE__{ info1: info1, info2: info2, info3: info3, info4: info4, result_code: result_code, generation: generation, expiration: expiration, timeout: timeout, fields: fields, operations: operations }} end end def decode(<>) when header_size != @msg_remaining_header_size do {:error, :invalid_header_size} end def decode(_), do: {:error, :incomplete_header} defp decode_fields(binary, 0), do: {:ok, [], binary} defp decode_fields(binary, count) do decode_fields(binary, count, []) end defp decode_fields(binary, 0, acc), do: {:ok, Enum.reverse(acc), binary} defp decode_fields(binary, count, acc) do case Field.decode(binary) do {:ok, field, rest} -> decode_fields(rest, count - 1, [field | acc]) {:error, _} = error -> error end end defp decode_operations(binary, 0), do: {:ok, [], binary} defp decode_operations(binary, count) do decode_operations(binary, count, []) end defp decode_operations(binary, 0, acc), do: {:ok, Enum.reverse(acc), binary} defp decode_operations(binary, count, acc) do case Operation.decode(binary) do {:ok, op, rest} -> decode_operations(rest, count - 1, [op | acc]) {:error, _} = error -> error end end @doc """ Creates a simple read command (GET). """ @spec read_command(binary(), binary(), binary()) :: t() def read_command(namespace, set, digest) do %__MODULE__{ info1: @info1_read ||| @info1_get_all, fields: [ Field.namespace(namespace), Field.set(set), Field.digest(digest) ] } end @doc """ Creates a simple write command (PUT). """ @spec write_command(binary(), binary(), binary(), [Operation.t()]) :: t() def write_command(namespace, set, digest, operations) do %__MODULE__{ info2: @info2_write, fields: [ Field.namespace(namespace), Field.set(set), Field.digest(digest) ], operations: operations } end @doc """ Creates a delete command. """ @spec delete_command(binary(), binary(), binary()) :: t() def delete_command(namespace, set, digest) do %__MODULE__{ info2: @info2_write ||| @info2_delete, fields: [ Field.namespace(namespace), Field.set(set), Field.digest(digest) ] } end @doc """ Creates an exists command (read metadata only, no bin data). Uses `INFO1_READ | INFO1_NOBINDATA` to check record existence without fetching bins. """ @spec exists_command(binary(), binary(), binary()) :: t() def exists_command(namespace, set, digest) do %__MODULE__{ info1: @info1_read ||| @info1_nobindata, fields: [ Field.namespace(namespace), Field.set(set), Field.digest(digest) ] } end @doc """ Creates a touch command (refresh TTL without changing bins). Uses `INFO2_WRITE` with a single touch operation to refresh the record's TTL. """ @spec touch_command(binary(), binary(), binary()) :: t() def touch_command(namespace, set, digest) do %__MODULE__{ info2: @info2_write, fields: [ Field.namespace(namespace), Field.set(set), Field.digest(digest) ], operations: [Operation.touch()] } end end