Why even classes

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

Author: Marenz

src/frequenz/dispatch/_bg_service.py

@@ -114,13 +114,6 @@ def __init__(
         )
 
         self._running_state_status_tx = self._running_state_status_channel.new_sender()
-        self._next_event_timer = Timer(
-            timedelta(seconds=100), SkipMissedAndResync(), auto_start=False
-        )
-        """The timer to schedule the next event.
-
-        Interval is chosen arbitrarily, as it will be reset on the first event.
-        """
 
         self._scheduled_events: list["DispatchScheduler.QueueItem"] = []
         """The scheduled events, sorted by time.
@@ -235,19 +228,20 @@ async def _run(self) -> None:
             self._microgrid_id,
         )
 
-        # Initial fetch
-        await self._fetch()
-
-        stream = self._client.stream(microgrid_id=self._microgrid_id)
-
         # Streaming updates
-        with closing(self._next_event_timer) as next_event_timer:
+        with closing(
+            Timer(timedelta(seconds=100), SkipMissedAndResync(), auto_start=False)
+        ) as next_event_timer:
+            # Initial fetch
+            await self._fetch(next_event_timer)
+            stream = self._client.stream(microgrid_id=self._microgrid_id)
+
             async for selected in select(next_event_timer, stream):
                 if selected_from(selected, next_event_timer):
                     if not self._scheduled_events:
                         continue
                     await self._execute_scheduled_event(
-                        heappop(self._scheduled_events).dispatch
+                        heappop(self._scheduled_events).dispatch, next_event_timer
                     )
                 elif selected_from(selected, stream):
                     _logger.debug("Received dispatch event: %s", selected.message)
@@ -256,14 +250,16 @@ async def _run(self) -> None:
                         case Event.CREATED:
                             self._dispatches[dispatch.id] = dispatch
                             await self._update_dispatch_schedule_and_notify(
-                                dispatch, None
+                                dispatch, None, next_event_timer
                             )
                             await self._lifecycle_events_tx.send(
                                 Created(dispatch=dispatch)
                             )
                         case Event.UPDATED:
                             await self._update_dispatch_schedule_and_notify(
-                                dispatch, self._dispatches[dispatch.id]
+                                dispatch,
+                                self._dispatches[dispatch.id],
+                                next_event_timer,
                             )
                             self._dispatches[dispatch.id] = dispatch
                             await self._lifecycle_events_tx.send(
@@ -272,18 +268,19 @@ async def _run(self) -> None:
                         case Event.DELETED:
                             self._dispatches.pop(dispatch.id)
                             await self._update_dispatch_schedule_and_notify(
-                                None, dispatch
+                                None, dispatch, next_event_timer
                             )
 
                             await self._lifecycle_events_tx.send(
                                 Deleted(dispatch=dispatch)
                             )
 
-    async def _execute_scheduled_event(self, dispatch: Dispatch) -> None:
+    async def _execute_scheduled_event(self, dispatch: Dispatch, timer: Timer) -> None:
         """Execute a scheduled event.
 
         Args:
             dispatch: The dispatch to execute.
+            timer: The timer to use for scheduling the next event.
         """
         _logger.debug("Executing scheduled event: %s (%s)", dispatch, dispatch.started)
         await self._send_running_state_change(dispatch)
@@ -298,9 +295,9 @@ async def _execute_scheduled_event(self, dispatch: Dispatch) -> None:
         else:
             self._schedule_start(dispatch)
 
-        self._update_timer()
+        self._update_timer(timer)
 
-    async def _fetch(self) -> None:
+    async def _fetch(self, timer: Timer) -> None:
         """Fetch all relevant dispatches using list.
 
         This is used for the initial fetch and for re-fetching all dispatches
@@ -321,12 +318,14 @@ async def _fetch(self) -> None:
                     old_dispatch = old_dispatches.pop(dispatch.id, None)
                     if not old_dispatch:
                         _logger.debug("New dispatch: %s", dispatch)
-                        await self._update_dispatch_schedule_and_notify(dispatch, None)
+                        await self._update_dispatch_schedule_and_notify(
+                            dispatch, None, timer
+                        )
                         await self._lifecycle_events_tx.send(Created(dispatch=dispatch))
                     elif dispatch.update_time != old_dispatch.update_time:
                         _logger.debug("Updated dispatch: %s", dispatch)
                         await self._update_dispatch_schedule_and_notify(
-                            dispatch, old_dispatch
+                            dispatch, old_dispatch, timer
                         )
                         await self._lifecycle_events_tx.send(Updated(dispatch=dispatch))
 
@@ -340,7 +339,7 @@ async def _fetch(self) -> None:
         for dispatch in old_dispatches.values():
             _logger.debug("Deleted dispatch: %s", dispatch)
             await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
-            await self._update_dispatch_schedule_and_notify(None, dispatch)
+            await self._update_dispatch_schedule_and_notify(None, dispatch, timer)
 
             # Set deleted only here as it influences the result of dispatch.started
             # which is used in above in _running_state_change
@@ -350,7 +349,7 @@ async def _fetch(self) -> None:
         self._initial_fetch_event.set()
 
     async def _update_dispatch_schedule_and_notify(
-        self, dispatch: Dispatch | None, old_dispatch: Dispatch | None
+        self, dispatch: Dispatch | None, old_dispatch: Dispatch | None, timer: Timer
     ) -> None:
         """Update the schedule for a dispatch.
 
@@ -408,13 +407,13 @@ async def _update_dispatch_schedule_and_notify(
                 self._schedule_start(dispatch)
 
         # We modified the schedule, so we need to reset the timer
-        self._update_timer()
+        self._update_timer(timer)
 
-    def _update_timer(self) -> None:
+    def _update_timer(self, timer: Timer) -> None:
         """Update the timer to the next event."""
         if self._scheduled_events:
             due_at: datetime = self._scheduled_events[0].time
-            self._next_event_timer.reset(interval=due_at - datetime.now(timezone.utc))
+            timer.reset(interval=due_at - datetime.now(timezone.utc))
             _logger.debug("Next event scheduled at %s", self._scheduled_events[0].time)
 
     def _remove_scheduled(self, dispatch: Dispatch) -> bool:

tests/test_frequenz_dispatch.py

@@ -461,7 +461,6 @@ async def test_dispatch_new_but_finished(
     test_env.client.set_dispatches(test_env.microgrid_id, [finished_dispatch])
     await test_env.service.stop()
     test_env.service.start()
-    fake_time.shift(timedelta(seconds=1))
     await asyncio.sleep(1)
 
     test_env = replace(
@@ -473,6 +472,7 @@ async def test_dispatch_new_but_finished(
             )
         ),
     )
+    fake_time.shift(timedelta(seconds=1))
     await asyncio.sleep(1)
 
     # Process the lifecycle event caused by the old dispatch at startup