defmodule SpawnSdk.Actor do @moduledoc """ Documentation for `Actor`. Actor look like this: defmodule MyActor do use SpawnSdk.Actor, name: "joe", persistent: false, state_type: Io.Eigr.Spawn.Example.MyState, deactivate_timeout: 5_000, snapshot_timeout: 2_000 require Logger alias Io.Eigr.Spawn.Example.{MyState, MyBusinessMessage} defact sum(%MyBusinessMessage{value: value} = data}, %Context{state: state} = ctx) do Logger.info("Received Request...") new_value = (state.value || 0) + value %Value{} |> Value.of(%MyBusinessMessage{value: new_value}, %MyState{value: new_value}) |> Value.reply!() end """ alias SpawnSdk.{ ActorRef, ActorChannel, ActorGroupRef, Context, Value } @type system :: String.t() @type actor :: String.t() @type parent :: actor() @type group :: ActorGroup.t() @type opts :: [ action: String.t() | atom() | nil, data: any() | nil, delay: integer() | nil, metadata: map() | nil, parent: ActorRef.t() | nil, pooled: boolean() | nil, scheduled_to: DateTime.t() | nil, spawn: boolean() | false ] @type action :: String.t() @type context :: Context.t() @type data :: term() @type error :: any() @type value :: Value.t() @callback handle_action({action(), data()}, context()) :: value() | {:reply, value()} | {:error, error()} | {:error, error(), value()} defmodule MalformedActor do defexception message: "Something is wrong with the actor implementation" end @spec channel(ActorChannel.t(), opts()) :: ActorChannel.t() def channel(channel, opts), do: %ActorChannel{channel: channel, opts: opts} @doc """ Creates a reference to an actor so that it can be invoked. The first argument is ActorSystem name. The second argument is a Actor name. The third argument is a keyword list of options: * `spawn` - a boolean indicating whether the actor should be generated or not. Default is **false**. Example: ``` iex(spawn_a@127.0.0.1)1> SpawnSdk.Actor.ref("spawn-system", "joe") %SpawnSdk.ActorRef{system: "spawn-system", name: "joe", opts: []} ``` To invoke an actor using the obtained reference you could do: ``` alias SpawnSdk.Actor my_data = %MyData{value: 1} Actor.ref("spawn-system", "joe") |> Actor.invoke(action: "sum", data: my_data) ``` """ @spec ref(system(), actor(), opts()) :: ActorRef.t() def ref(system, name, opts \\ []), do: %ActorRef{system: system, name: name, opts: opts} @spec group(list(ActorRef), opts()) :: ActorGroupRef.t() def group(actors, opts \\ []) when is_list(actors), do: %ActorGroupRef{actors: actors, opts: opts} @doc """ Spawn an actor and return its reference. The first argument is ActorSystem name. The second argument is a Actor name. The third argument is a parent actor name. Example: ``` iex(spawn_a@127.0.0.1)1> SpawnSdk.Actor.spawn("spawn-system", "joe", "unnamed_actor") %SpawnSdk.ActorRef{system: "spawn-system", name: "joe", opts: []} ``` To invoke an actor using the obtained reference you could do: ``` alias SpawnSdk.Actor my_data = %MyData{value: 1} Actor.spawn("spawn-system", "joe", "unnamed_actor") |> Actor.invoke(action: "sum", data: my_data) ``` """ @spec spawn(system(), actor(), parent(), opts()) :: ActorRef.t() def spawn(system, name, parent, opts \\ []) do :ok = SpawnSdk.spawn_actor(name, system: system, actor: parent) new_options = Keyword.merge(opts, parent: parent, already_spawned: true) %ActorRef{system: system, name: name, opts: new_options} end @doc """ Creates a group of actor reference so that it can be invoked Example: ``` alias SpawnSdk.Actor my_data = %MyData{value: 1} Actor.ref("erlang-system", "joe", action: :sum, data: my_data) |> Actor.to_group("erlang-system", "robert", action: :sum, data: my_data) |> Actor.multi() ``` """ @spec to_group(ActorRef.t(), opts()) :: ActorGroupRef.t() def to_group(%ActorRef{} = first, opts) do %ActorGroupRef{actors: [first], opts: opts} end @doc """ Creates a group of actor reference so that it can be invoked Example: ``` alias SpawnSdk.Actor my_data = %MyData{value: 1} Actor.ref("erlang-system", "joe") |> Actor.to_group("erlang-system", "robert") |> Actor.to_group("erlang-system", "mike") |> Actor.to_group("eigr-system", "adriano") |> Actor.to_group("eigr-system", "marcel") |> Actor.to_group("spawn-elixir-system", "elias") |> Actor.multi(action: :sum, data: my_data) ``` """ @spec to_group(ActorRef.t() | ActorGroupRef.t(), system(), actor(), opts()) :: ActorGroupRef.t() def to_group(ref, system, actor, opts \\ []) def to_group(%ActorRef{} = first, system, actor, opts) when is_nil(system) and is_nil(actor) do %ActorGroupRef{actors: [first], opts: opts} end def to_group(%ActorRef{} = first, system, actor, opts) do last = %ActorRef{system: system, name: actor, opts: opts} %ActorGroupRef{actors: [first] ++ last, opts: opts} end def to_group(%ActorGroupRef{actors: first} = ref, system, actor, opts) when is_list(first) do last = %ActorRef{system: system, name: actor, opts: opts} %ActorGroupRef{ref | actors: first ++ last} end @doc """ Sends a assynchronous message to group of actors. Example: ``` alias SpawnSdk.Actor Actor.ref("spawn-system", "joe") |> Actor.cast(action: "sum", data: %MyData{value: 1}) ``` """ @spec cast(ActorRef.t(), opts()) :: {:ok, :async} def cast(%ActorRef{system: system, name: actor, opts: actor_opts} = _ref, opts) do spawn? = Keyword.get(actor_opts, :spawn, false) || Keyword.get(opts, :spawn, false) already_spawned? = Keyword.get(actor_opts, :already_spawned) new_opts = Keyword.merge(opts, system: system, async: true) case spawn? and not already_spawned? do true -> %ActorRef{system: _parent_system, name: parent_name} = _parent = Keyword.get(actor_opts, :parent) || Keyword.get(opts, :parent) new_opts = Keyword.merge(new_opts, actor: parent_name) SpawnSdk.invoke(actor, new_opts) false -> SpawnSdk.invoke(actor, new_opts) end end @doc """ Sends a message to the actor and returns the result. Example: ``` alias SpawnSdk.Actor Actor.ref("spawn-system", "joe") |> Actor.invoke(action: "sum", data: %MyData{value: 1}) ``` """ @spec invoke(ActorRef.t(), opts()) :: {:ok, any()} | {:error, any()} def invoke(%ActorRef{system: system, name: actor, opts: actor_opts} = _ref, opts) do spawn? = Keyword.get(actor_opts, :spawn, false) || Keyword.get(opts, :spawn, false) already_spawned? = Keyword.get(actor_opts, :already_spawned) new_opts = Keyword.merge(opts, system: system, async: false) case spawn? and not already_spawned? do true -> %ActorRef{system: _parent_system, name: parent_name} = _parent_name = Keyword.get(actor_opts, :parent) || Keyword.get(opts, :parent) new_opts = Keyword.merge(new_opts, actor: parent_name) SpawnSdk.invoke(actor, new_opts) false -> SpawnSdk.invoke(actor, new_opts) end end @doc """ Invokes a group of actors and returns all results. Example: ``` alias SpawnSdk.Actor my_data = %MyData{value: 1} Actor.ref("erlang-system", "joe", action: :sum, data: my_data) |> Actor.to_group("erlang-system", "robert", action: :sum, data: my_data) |> Actor.to_group("erlang-system", "mike", action: :sum, data: my_data) |> Actor.to_group("eigr-system", "adriano", action: :sum, data: my_data) |> Actor.to_group("eigr-system", "marcel", action: :sum, data: my_data) |> Actor.to_group("spawn-elixir-system", "elias"", action: "calc", data: my_data) |> Actor.multi() ``` """ @spec multi(ActorGroupRef.t()) :: list(any()) | {:error, any()} def multi(%ActorGroupRef{actors: actors} = _ref) do tasks = Enum.map(actors, fn %ActorRef{opts: actor_opts} = actor -> Task.async(fn -> invoke(actor, actor_opts) end) end) Task.await_many(tasks) end @spec pub(ActorChannel.t(), opts()) :: :ok def pub(%ActorChannel{} = _channel, _opts) do # TODO: implement this in Proxy too? end defmacro __using__(opts) do quote bind_quoted: [opts: opts] do alias SpawnSdk.{ Context, Flow.Broadcast, Flow.Pipe, Flow.Forward, Flow.SideEffect, Value } import SpawnSdk.Actor use SpawnSdk.Defact import SpawnSdk.System.SpawnSystem, only: [ invoke: 2, register: 2, spawn_actor: 2 ] Module.register_attribute(__MODULE__, :actor_opts, persist: true) Module.put_attribute(__MODULE__, :actor_opts, opts) @behaviour SpawnSdk.Actor @before_compile SpawnSdk.Actor end end defmacro __before_compile__(_a) do opts = Module.get_attribute(__CALLER__.module, :actor_opts) actions = Module.get_attribute(__CALLER__.module, :defact_exports) actor_name = Keyword.get(opts, :name, Atom.to_string(__CALLER__.module)) actor_kind = Keyword.get(opts, :kind, :NAMED) |> SpawnSdk.System.SpawnSystem.decode_kind() channel_group = Keyword.get(opts, :channels, []) min_pool_size = Keyword.get(opts, :min_pool_size, 1) max_pool_size = Keyword.get(opts, :max_pool_size, 0) state_type = Keyword.get(opts, :state_type, :json) stateful = Keyword.get(opts, :stateful, true) tags = Keyword.get(opts, :tags, nil) init? = Enum.any?(actions, fn {action, _} -> String.downcase(action) == "init" end) setup? = Enum.any?(actions, fn {action, _} -> String.downcase(action) == "setup" end) if init? and setup? do raise SpawnSdk.Actor.MalformedActor, "You can't have both init and setup actions implemented, they are reserved and do the same thing" end Enum.each(channel_group, fn {topic, action} -> if String.contains?(topic, "*") do raise SpawnSdk.Actor.MalformedActor, "Channels doesn't support wildcard subscriptions" end if "#{action}" not in Enum.map(actions, fn {action, _} -> "#{action}" end) do raise SpawnSdk.Actor.MalformedActor, "Channel action #{action} is not implemented in current Actor." end topic when is_binary(topic) -> if String.contains?(topic, "*") do raise SpawnSdk.Actor.MalformedActor, "Channels doesn't support wildcard subscriptions" end if "receive" not in Enum.map(actions, fn {action, _} -> "#{action}" end) do raise SpawnSdk.Actor.MalformedActor, "Found a registered topic but no action named receive was found." end _ -> raise SpawnSdk.Actor.MalformedActor, "Channels is not invalid, use list(topic) or list({topic, action})" end) if stateful and !Code.ensure_loaded?(Statestores.Supervisor) do raise SpawnSdk.Actor.MalformedActor, """ ArgumentError. You need to add :spawn_statestores to your dependency if you are going to use persistent actors. Otherwise, set `stateful: false` in your Actor attributes """ end if state_type == nil and stateful do raise SpawnSdk.Actor.MalformedActor, "ArgumentError. State type is mandatory if stateful is true" end if stateful and actor_kind == :POOLED do raise SpawnSdk.Actor.MalformedActor, """ Pooled Actors cannot be stateful. Please set stateful attribute to false to be able to register Actor #{actor_name} """ end deactivate_timeout = Keyword.get(opts, :deactivate_timeout, 10_000) snapshot_timeout = Keyword.get(opts, :snapshot_timeout, 2_000) quote do def __meta__(:actions) do unquote(actions) |> Enum.filter(fn {_action, %{timer: timer}} -> is_nil(timer) end) |> Enum.map(fn {action, %{timer: timer}} -> action end) end def __meta__(:timers) do unquote(actions) |> Enum.reject(fn {_action, %{timer: timer}} -> is_nil(timer) end) |> Enum.map(fn {action, %{timer: timer}} -> {action, timer} end) end def __meta__(:channel_group), do: unquote(channel_group) def __meta__(:name), do: unquote(actor_name) def __meta__(:kind), do: unquote(actor_kind) def __meta__(:stateful), do: unquote(stateful) def __meta__(:state_type), do: unquote(state_type) def __meta__(:min_pool_size), do: unquote(min_pool_size) def __meta__(:max_pool_size), do: unquote(max_pool_size) def __meta__(:snapshot_timeout), do: unquote(snapshot_timeout) def __meta__(:deactivate_timeout), do: unquote(deactivate_timeout) def __meta__(:tags), do: Map.new(unquote(tags) || %{}) end end end