Improvement the Protocol

apps/actors/lib/actors.ex

@@ -9,7 +9,13 @@ defmodule Actors do
 
   alias Actors.Registry.ActorRegistry
 
-  alias Eigr.Functions.Protocol.Actors.{Actor, ActorSystem, Registry}
+  alias Eigr.Functions.Protocol.Actors.{
+    Actor,
+    ActorId,
+    ActorConfiguration,
+    ActorSystem,
+    Registry
+  }
 
   alias Eigr.Functions.Protocol.{
     InvocationRequest,
@@ -27,12 +33,12 @@ defmodule Actors do
           service_info: %ServiceInfo{} = _service_info,
           actor_system:
             %ActorSystem{name: _name, registry: %Registry{actors: actors} = _registry} =
-              actor_system
+              _actor_system
         } = _registration
       ) do
     ActorRegistry.register(actors)
 
-    with :ok <- create_actors(actor_system, actors) do
+    with :ok <- create_actors(actors) do
       proxy_info =
         ProxyInfo.new(
           protocol_major_version: 1,
@@ -84,18 +90,16 @@ defmodule Actors do
 
   def invoke(
         %InvocationRequest{
-          actor: %Actor{} = actor,
-          system: %ActorSystem{} = system,
+          actor_id: %ActorId{} = actor,
           async: type
         } = request
       ) do
-    invoke(type, system, actor, request)
+    invoke(type, actor, request)
   end
 
   defp invoke(
          false,
-         %ActorSystem{name: system_name} = system,
-         %Actor{name: actor_name} = actor,
+         %ActorId{name: actor_name, actor_system: system_name} = _actor_id,
          request
        ) do
     case Spawn.Cluster.Node.Registry.lookup(Actors.Actor.Entity, actor_name) do
@@ -105,9 +109,10 @@ defmodule Actors do
         ActorEntity.invoke(actor_name, request)
 
       _ ->
-        with {:ok, %{node: node, actor: _registered_actor}} <-
+        with {:ok, %{node: node, actor: registered_actor}} <-
                ActorRegistry.lookup(system_name, actor_name),
-             _pid <- Node.spawn(node, __MODULE__, :try_reactivate_actor, [system, actor]) do
+             _pid <-
+               Node.spawn(node, __MODULE__, :try_reactivate_actor, [registered_actor]) do
           Process.sleep(1)
           {:ok, response_body} = ActorEntity.invoke(actor_name, request)
 
@@ -133,8 +138,7 @@ defmodule Actors do
 
   defp invoke(
          true,
-         %ActorSystem{name: system_name} = system,
-         %Actor{name: actor_name} = actor,
+         %ActorId{name: actor_name, actor_system: system_name} = _actor_id,
          request
        ) do
     case Spawn.Cluster.Node.Registry.lookup(Actors.Actor.Entity, actor_name) do
@@ -144,9 +148,10 @@ defmodule Actors do
         ActorEntity.invoke(actor_name, request)
 
       _ ->
-        with {:ok, %{node: node, actor: _registered_actor}} <-
+        with {:ok, %{node: node, actor: registered_actor}} <-
                ActorRegistry.lookup(system_name, actor_name),
-             _pid <- Node.spawn(node, __MODULE__, :try_reactivate_actor, [system, actor]) do
+             _pid <-
+               Node.spawn(node, __MODULE__, :try_reactivate_actor, [registered_actor]) do
           Process.sleep(1)
           {:ok, response_body} = ActorEntity.invoke_async(actor_name, request)
 
@@ -170,8 +175,8 @@ defmodule Actors do
     end
   end
 
-  def try_reactivate_actor(%ActorSystem{} = system, %Actor{name: name} = actor) do
-    case ActorEntitySupervisor.lookup_or_create_actor(system, actor) do
+  def try_reactivate_actor(%Actor{actor_id: %ActorId{name: name}} = actor) do
+    case ActorEntitySupervisor.lookup_or_create_actor(actor) do
       {:ok, pid} ->
         Logger.debug("Actor #{name} reactivated. PID: #{inspect(pid)}")
         {:ok, pid}
@@ -182,29 +187,17 @@ defmodule Actors do
     end
   end
 
-  # To lookup all actors
-  def try_reactivate_actor(nil, %Actor{name: name} = actor) do
-    case ActorEntitySupervisor.lookup_or_create_actor(nil, actor) do
-      {:ok, pid} ->
-        Logger.debug("Actor #{name} reactivated. PID: #{inspect(pid)}")
-        {:ok, pid}
-
-      reason ->
-        Logger.error("Failed to reactivate actor #{name}: #{inspect(reason)}")
-        {:error, reason}
-    end
-  end
-
-  defp create_actors(actor_system, actors) do
+  defp create_actors(actors) do
     actors
     |> Flow.from_enumerable(
       min_demand: @activate_actors_min_demand,
       max_demand: @activate_actors_max_demand
     )
+    |> Flow.filter(&is_singleton?/1)
     |> Flow.map(fn {actor_name, actor} ->
       Logger.debug("Registering #{actor_name} #{inspect(actor)} on Node: #{inspect(Node.self())}")
 
-      {time, result} = :timer.tc(&lookup_actor/3, [actor_system, actor_name, actor])
+      {time, result} = :timer.tc(&lookup_actor/2, [actor_name, actor])
 
       Logger.info(
         "Registered and Activated the #{actor_name} on Node #{inspect(Node.self())} in #{inspect(time)}ms"
@@ -215,8 +208,17 @@ defmodule Actors do
     |> Flow.run()
   end
 
-  defp lookup_actor(actor_system, actor_name, actor) do
-    case ActorEntitySupervisor.lookup_or_create_actor(actor_system, actor) do
+  defp is_singleton?(
+         {_actor_name,
+          %Actor{configuration: %ActorConfiguration{dispatcher: SINGLETON_DISPATCHER}}}
+       ),
+       do: true
+
+  defp is_singleton?({_actor_name, %Actor{configuration: %ActorConfiguration{dispatcher: _}}}),
+    do: false
+
+  defp lookup_actor(actor_name, actor) do
+    case ActorEntitySupervisor.lookup_or_create_actor(actor) do
       {:ok, pid} ->
         {:ok, pid}