defmodule Aerospike.Command.BatchUdf do @moduledoc false alias Aerospike.BatchResult alias Aerospike.Command.Batch, as: MixedBatch alias Aerospike.Command.BatchCommand.Entry alias Aerospike.Error alias Aerospike.Key alias Aerospike.Policy @type option :: {:timeout, non_neg_integer()} | {:socket_timeout, non_neg_integer()} @type result :: {:ok, [BatchResult.t()]} | {:error, Error.t()} | {:error, :cluster_not_ready} @spec execute(GenServer.server(), [Key.t()], String.t(), String.t(), list(), [option()]) :: result() def execute(tender, keys, package, function, args, opts \\ []) def execute(_tender, [], package, function, args, opts) when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do with {:ok, _parent_policy} <- Policy.batch(Policy.batch_parent_opts(opts)), {:ok, _write_policy} <- Policy.batch_record_write(Policy.batch_record_opts(opts)) do {:ok, []} end end def execute(tender, keys, package, function, args, opts) when is_list(keys) and is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do with :ok <- validate_keys(keys), {:ok, _parent_policy} <- Policy.batch(Policy.batch_parent_opts(opts)), {:ok, write_policy} <- Policy.batch_record_write(Policy.batch_record_opts(opts)), {:ok, results} <- MixedBatch.execute( tender, entries(keys, package, function, args, write_policy), Policy.batch_parent_opts(opts) ) do {:ok, MixedBatch.to_public_results(results)} end end defp entries(keys, package, function, args, %Policy.BatchWrite{} = policy) do keys |> Enum.with_index() |> Enum.map(fn {key, index} -> %Entry{ index: index, key: key, kind: :udf, dispatch: :write, payload: Map.merge(payload(policy), %{package: package, function: function, args: args}) } end) end defp payload(%Policy.BatchWrite{} = policy) do Map.take(policy, [ :ttl, :generation, :generation_policy, :filter, :exists, :commit_level, :durable_delete, :respond_per_op, :send_key, :read_mode_ap, :read_mode_sc, :read_touch_ttl_percent ]) end defp validate_keys(keys) do if Enum.all?(keys, &match?(%Key{}, &1)) do :ok else invalid_argument("Aerospike.batch_udf/6 expects a list of %Aerospike.Key{} values") end end defp invalid_argument(message) do {:error, Error.from_result_code(:invalid_argument, message: message)} end end