Turn dispatch actor into background service

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

Author: Marenz

src/frequenz/dispatch/actor.py src/frequenz/dispatch/_bg_service.pysrc/frequenz/dispatch/actor.py renamed to src/frequenz/dispatch/_bg_service.py

@@ -3,6 +3,7 @@
 
 """The dispatch actor."""
 
+import asyncio
 import logging
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
@@ -13,7 +14,7 @@
 from frequenz.channels.timer import SkipMissedAndResync, Timer
 from frequenz.client.dispatch import Client
 from frequenz.client.dispatch.types import Event
-from frequenz.sdk.actor import Actor
+from frequenz.sdk.actor import BackgroundService
 
 from ._dispatch import Dispatch
 from ._event import Created, Deleted, DispatchEvent, Updated
@@ -23,13 +24,11 @@
 
 
 # pylint: disable=too-many-instance-attributes
-class DispatchingActor(Actor):
-    """Dispatch actor.
+class DispatchBackgroundService(BackgroundService):
+    """Dispatch background service.
 
-    This actor is responsible for handling dispatches for a microgrid.
-
-    This means staying in sync with the API and scheduling
-    dispatches as necessary.
+    This service is responsible for managing dispatches and scheduling them
+    based on their start and stop times.
     """
 
     @dataclass(order=True)
@@ -52,7 +51,7 @@ def __init__(
         microgrid_id: int,
         client: Client,
     ) -> None:
-        """Initialize the actor.
+        """Initialize the background service.
 
         Args:
             microgrid_id: The microgrid ID to handle dispatches for.
@@ -81,7 +80,7 @@ def __init__(
         Interval is chosen arbitrarily, as it will be reset on the first event.
         """
 
-        self._scheduled_events: list["DispatchingActor.QueueItem"] = []
+        self._scheduled_events: list["DispatchBackgroundService.QueueItem"] = []
         """The scheduled events, sorted by time.
 
         Each event is a tuple of the scheduled time and the dispatch.
@@ -130,9 +129,16 @@ async def new_running_state_event_receiver(self, type: str) -> Receiver[Dispatch
 
     # pylint: enable=redefined-builtin
 
+    def start(self) -> None:
+        """Start the background service."""
+        self._tasks.add(asyncio.create_task(self._run()))
+
     async def _run(self) -> None:
-        """Run the actor."""
-        _logger.info("Starting dispatch actor for microgrid %s", self._microgrid_id)
+        """Run the background service."""
+        _logger.info(
+            "Starting dispatching background service for microgrid %s",
+            self._microgrid_id,
+        )
 
         # Initial fetch
         await self._fetch()

src/frequenz/dispatch/_dispatcher.py

@@ -7,9 +7,9 @@
 from frequenz.channels import Receiver
 from frequenz.client.dispatch import Client
 
+from ._bg_service import DispatchBackgroundService
 from ._dispatch import Dispatch
 from ._event import DispatchEvent
-from .actor import DispatchingActor
 
 
 class Dispatcher:
@@ -172,14 +172,14 @@ def __init__(
             key: The key to access the service.
         """
         self._client = Client(server_url=server_url, key=key)
-        self._actor = DispatchingActor(
+        self._bg_service = DispatchBackgroundService(
             microgrid_id,
             self._client,
         )
 
     async def start(self) -> None:
-        """Start the actor."""
-        self._actor.start()
+        """Start the local dispatch service."""
+        self._bg_service.start()
 
     @property
     def client(self) -> Client:
@@ -197,7 +197,7 @@ def new_lifecycle_events_receiver(
         Returns:
             A new receiver for new dispatches.
         """
-        return self._actor.new_lifecycle_events_receiver(type)
+        return self._bg_service.new_lifecycle_events_receiver(dispatch_type)
 
     async def new_running_state_event_receiver(
         self, dispatch_type: str
@@ -234,4 +234,4 @@ async def new_running_state_event_receiver(
         Returns:
             A new receiver for dispatches whose running status changed.
         """
-        return await self._actor.new_running_state_event_receiver(type)
+        return await self._bg_service.new_running_state_event_receiver(dispatch_type)

tests/test_frequenz_dispatch.py

@@ -19,7 +19,7 @@
 from pytest import fixture
 
 from frequenz.dispatch import Created, Deleted, Dispatch, DispatchEvent, Updated
-from frequenz.dispatch.actor import DispatchingActor
+from frequenz.dispatch._bg_service import DispatchBackgroundService
 
 
 @fixture
@@ -48,7 +48,7 @@ def _now() -> datetime:
 class ActorTestEnv:
     """Test environment for the actor."""
 
-    actor: DispatchingActor
+    actor: DispatchBackgroundService
     """The actor under test."""
     lifecycle_events: Receiver[DispatchEvent]
     """The receiver for updated dispatches."""
@@ -66,7 +66,7 @@ async def actor_env() -> AsyncIterator[ActorTestEnv]:
     microgrid_id = randint(1, 100)
     client = FakeClient()
 
-    actor = DispatchingActor(
+    actor = DispatchBackgroundService(
         microgrid_id=microgrid_id,
         client=client,
     )

tests/test_mananging_actor.py

@@ -18,7 +18,7 @@
 from pytest import fixture
 
 from frequenz.dispatch import Dispatch, DispatchManagingActor, DispatchUpdate
-from frequenz.dispatch.actor import DispatchingActor
+from frequenz.dispatch._bg_service import DispatchBackgroundService
 
 
 @fixture
@@ -142,14 +142,16 @@ def test_heapq_dispatch_compare(test_env: TestEnv) -> None:
     until_time = now + timedelta(minutes=5)
 
     # Create the heap
-    scheduled_events: list[DispatchingActor.QueueItem] = []
+    scheduled_events: list[DispatchBackgroundService.QueueItem] = []
 
     # Push two events with the same 'until' time onto the heap
     heapq.heappush(
-        scheduled_events, DispatchingActor.QueueItem(until_time, Dispatch(dispatch1))
+        scheduled_events,
+        DispatchBackgroundService.QueueItem(until_time, Dispatch(dispatch1)),
     )
     heapq.heappush(
-        scheduled_events, DispatchingActor.QueueItem(until_time, Dispatch(dispatch2))
+        scheduled_events,
+        DispatchBackgroundService.QueueItem(until_time, Dispatch(dispatch2)),
     )