Make actor_factory function async for more flexibility

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

src/frequenz/dispatch/_actor_dispatcher.py

@@ -7,7 +7,7 @@
 import logging
 from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Awaitable
 
 from frequenz.channels import Broadcast, Receiver
 from frequenz.client.dispatch.types import TargetComponents
@@ -136,7 +136,9 @@ async def main():
 
     def __init__(
         self,
-        actor_factory: Callable[[DispatchInfo, Receiver[DispatchInfo]], Actor],
+        actor_factory: Callable[
+            [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
+        ],
         running_status_receiver: Receiver[Dispatch],
         dispatch_identity: Callable[[Dispatch], int] | None = None,
     ) -> None:
@@ -189,7 +191,7 @@ async def _start_actor(self, dispatch: Dispatch) -> None:
         else:
             try:
                 _logger.info("Starting actor for dispatch type %r", dispatch.type)
-                actor = self._actor_factory(
+                actor = await self._actor_factory(
                     dispatch_update,
                     self._updates_channel.new_receiver(limit=1, warn_on_overflow=False),
                 )

src/frequenz/dispatch/_dispatcher.py

@@ -8,7 +8,7 @@
 import asyncio
 import logging
 from asyncio import Event
-from typing import Callable
+from typing import Awaitable, Callable
 
 from frequenz.channels import Receiver
 from frequenz.client.dispatch import Client
@@ -236,7 +236,9 @@ async def start_dispatching(
         self,
         dispatch_type: str,
         *,
-        actor_factory: Callable[[DispatchInfo, Receiver[DispatchInfo]], Actor],
+        actor_factory: Callable[
+            [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
+        ],
         merge_strategy: MergeStrategy | None = None,
     ) -> None:
         """Manage actors for a given dispatch type.

tests/test_mananging_actor.py

@@ -78,6 +78,14 @@ async def _run(self) -> None:
         while True:
             await asyncio.sleep(1)
 
+    @classmethod
+    async def create(
+        cls, initial_dispatch: DispatchInfo, receiver: Receiver[DispatchInfo]
+    ) -> "MockActor":
+        """Create a new actor."""
+        actor = cls(initial_dispatch, receiver)
+        return actor
+
 
 @dataclass
 class TestEnv:
@@ -101,7 +109,7 @@ async def test_env() -> AsyncIterator[TestEnv]:
     channel = Broadcast[Dispatch](name="dispatch ready test channel")
 
     actors_service = ActorDispatcher(
-        actor_factory=MockActor,
+        actor_factory=MockActor.create,
         running_status_receiver=channel.new_receiver(),
         dispatch_identity=lambda dispatch: dispatch.id,
     )
@@ -296,14 +304,14 @@ async def new_mock_receiver(
         ):
             await dispatcher.start_dispatching(
                 dispatch_type="MANAGE_TEST",
-                actor_factory=MockActor,
+                actor_factory=MockActor.create,
                 merge_strategy=strategy,
             )
 
         # pylint: disable=protected-access
         assert "MANAGE_TEST" in dispatcher._actor_dispatchers
         actor_manager = dispatcher._actor_dispatchers["MANAGE_TEST"]
-        assert actor_manager._actor_factory == MockActor
+        assert actor_manager._actor_factory == MockActor.create
 
         dispatch = Dispatch(
             replace(