defmodule Aerospike.Protocol.CDT do @moduledoc false alias Aerospike.Exp alias Aerospike.Protocol.AsmMsg.Operation alias Aerospike.Protocol.MessagePack @particle_blob 4 @context_eval 0xFF @spec list_read_op(String.t(), integer(), [term()], [tuple()] | nil) :: Operation.t() def list_read_op(bin_name, op_code, args, ctx \\ nil) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %Operation{ op_type: Operation.op_cdt_read(), particle_type: @particle_blob, bin_name: bin_name, data: encode_payload(op_code, args, ctx) } end @spec list_modify_op(String.t(), integer(), [term()], [tuple()] | nil) :: Operation.t() def list_modify_op(bin_name, op_code, args, ctx \\ nil) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %Operation{ op_type: Operation.op_cdt_modify(), particle_type: @particle_blob, bin_name: bin_name, data: encode_payload(op_code, args, ctx) } end @spec list_create_op(String.t(), integer(), [term()], [tuple()]) :: Operation.t() def list_create_op(bin_name, flag, args, ctx) when is_binary(bin_name) and is_integer(flag) and is_list(args) and is_list(ctx) do %Operation{ op_type: Operation.op_cdt_modify(), particle_type: @particle_blob, bin_name: bin_name, data: encode_create_payload(flag, args, ctx) } end @spec map_read_op(String.t(), integer(), [term()], [tuple()] | nil) :: Operation.t() def map_read_op(bin_name, op_code, args, ctx \\ nil) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %{list_read_op(bin_name, op_code, args, ctx) | map_cdt: true} end @spec map_modify_op(String.t(), integer(), [term()], [tuple()] | nil) :: Operation.t() def map_modify_op(bin_name, op_code, args, ctx \\ nil) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %{list_modify_op(bin_name, op_code, args, ctx) | map_cdt: true} end @spec map_create_op(String.t(), integer(), [term()], [tuple()]) :: Operation.t() def map_create_op(bin_name, flag, args, ctx) when is_binary(bin_name) and is_integer(flag) and is_list(args) and is_list(ctx) do %{list_create_op(bin_name, flag, args, ctx) | map_cdt: true} end @spec bit_read_op(String.t(), integer(), [term()], [tuple()] | nil) :: Operation.t() def bit_read_op(bin_name, op_code, args, ctx \\ nil) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %Operation{ op_type: Operation.op_bit_read(), particle_type: @particle_blob, bin_name: bin_name, data: encode_payload(op_code, args, ctx) } end @spec bit_modify_op(String.t(), integer(), [term()], [tuple()] | nil) :: Operation.t() def bit_modify_op(bin_name, op_code, args, ctx \\ nil) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %Operation{ op_type: Operation.op_bit_modify(), particle_type: @particle_blob, bin_name: bin_name, data: encode_payload(op_code, args, ctx) } end @spec hll_read_op(String.t(), integer(), [term()]) :: Operation.t() def hll_read_op(bin_name, op_code, args) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %Operation{ op_type: Operation.op_hll_read(), particle_type: @particle_blob, bin_name: bin_name, data: encode_simple_array(op_code, args) } end @spec hll_modify_op(String.t(), integer(), [term()]) :: Operation.t() def hll_modify_op(bin_name, op_code, args) when is_binary(bin_name) and is_integer(op_code) and is_list(args) do %Operation{ op_type: Operation.op_hll_modify(), particle_type: @particle_blob, bin_name: bin_name, data: encode_simple_array(op_code, args) } end @spec cdt_pack_arg(term()) :: term() def cdt_pack_arg(%Exp{wire: wire}), do: {:byte_array, wire} def cdt_pack_arg({:bytes, _} = bytes), do: bytes def cdt_pack_arg(s) when is_binary(s), do: {:particle_string, s} def cdt_pack_arg(list) when is_list(list), do: Enum.map(list, &cdt_pack_arg/1) def cdt_pack_arg(map) when is_map(map), do: Map.new(map, fn {k, v} -> {cdt_pack_arg(k), cdt_pack_arg(v)} end) def cdt_pack_arg(other), do: other @spec encode_ctx([Aerospike.Ctx.step()]) :: binary() def encode_ctx(ctx) when is_list(ctx) do ctx |> ctx_terms() |> MessagePack.pack!() end defp encode_payload(op_code, args, ctx) do inner = [op_code | Enum.map(args, &cdt_pack_arg/1)] case ctx do nil -> MessagePack.pack!(inner) [] -> MessagePack.pack!(inner) steps when is_list(steps) -> MessagePack.pack!([@context_eval, ctx_terms(steps), inner]) end end defp encode_create_payload(flag, args, ctx) do [last | reversed] = Enum.reverse(ctx) ctx = reversed |> Enum.reverse() |> Kernel.++([apply_create_flag(last, flag)]) MessagePack.pack!([@context_eval, ctx_terms(ctx), Enum.map(args, &cdt_pack_arg/1)]) end defp ctx_terms(ctx) do Enum.flat_map(ctx, fn {id, val} -> [id, cdt_pack_arg(val)] end) end defp apply_create_flag({id, val}, flag), do: {Bitwise.bor(id, flag), val} defp encode_simple_array(op_code, args) do MessagePack.pack!([op_code | Enum.map(args, &cdt_pack_arg/1)]) end end