Replace channel properties with receiver functions

This allows us to control what happens when a new receiver is created.
In this case we want to send all relevant running state events to each new receiver.
It won't hurt us that other receivers might get them as well as it would just repeat the states they already know of.

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

Author: Marenz

RELEASE_NOTES.md

@@ -6,13 +6,14 @@
 
 ## Upgrading
 
-* The method `Dispatch.running(type: str)` was replaced with the property `Dispatch.started: bool`.
-* The SDK dependency was widened to allow versions up to (excluding) v1.0.0-rc1500
+* Two properties have been replaced by methods that require a type as parameter.
+    * `Dispatcher.lifecycle_events` has been replaced by the method `Dispatcher.new_lifecycle_events_receiver(self, type: str)`.
+    * `Dispatcher.running_status_change` has been replaced by the method `Dispatcher.new_running_state_event_receiver(self, type: str)`.
 
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->
 
 ## Bug Fixes
 
-* Fixed a crash when reading a Dispatch with frequency YEARLY.
+<!-- Here goes notable bug fixes that are worth a special mention or explanation -->

src/frequenz/dispatch/_dispatcher.py

@@ -6,7 +6,7 @@
 import abc
 from typing import Protocol, TypeVar
 
-from frequenz.channels import Broadcast, Receiver
+from frequenz.channels import Receiver
 from frequenz.client.dispatch import Client
 
 from ._dispatch import Dispatch
@@ -197,16 +197,10 @@ def __init__(
             server_url: The URL of the dispatch service.
             key: The key to access the service.
         """
-        self._running_state_channel = Broadcast[Dispatch](name="running_state_change")
-        self._lifecycle_events_channel = Broadcast[DispatchEvent](
-            name="lifecycle_events"
-        )
         self._client = Client(server_url=server_url, key=key)
         self._actor = DispatchingActor(
             microgrid_id,
             self._client,
-            self._lifecycle_events_channel.new_sender(),
-            self._running_state_channel.new_sender(),
         )
 
     async def start(self) -> None:
@@ -218,18 +212,19 @@ def client(self) -> Client:
         """Return the client."""
         return self._client
 
-    @property
-    def lifecycle_events(self) -> ReceiverFetcher[DispatchEvent]:
-        """Return new, updated or deleted dispatches receiver fetcher.
+    def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
+        """Return new, updated or deleted dispatches receiver.
+
+        Args:
+            type: The type of the dispatch to listen for.
 
         Returns:
             A new receiver for new dispatches.
         """
-        return self._lifecycle_events_channel
+        return self._actor.new_lifecycle_events_receiver(type)
 
-    @property
-    def running_status_change(self) -> ReceiverFetcher[Dispatch]:
-        """Return running status change receiver fetcher.
+    async def new_running_state_event_receiver(self, type: str) -> Receiver[Dispatch]:
+        """Return running state event receiver.
 
         This receiver will receive a message whenever the current running
         status of a dispatch changes.
@@ -242,7 +237,7 @@ def running_status_change(self) -> ReceiverFetcher[Dispatch]:
         then a message will be sent.
 
         In other words: Any change that is expected to make an actor start, stop
-        or reconfigure itself with new parameters causes a message to be
+        or adjust itself according to new dispatch options causes a message to be
         sent.
 
         A non-exhaustive list of possible changes that will cause a message to be sent:
@@ -255,7 +250,10 @@ def running_status_change(self) -> ReceiverFetcher[Dispatch]:
          - The payload changed
          - The dispatch was deleted
 
+        Args:
+            type: The type of the dispatch to listen for.
+
         Returns:
             A new receiver for dispatches whose running status changed.
         """
-        return self._running_state_channel
+        return await self._actor.new_running_state_event_receiver(type)

src/frequenz/dispatch/actor.py

@@ -9,7 +9,7 @@
 from heapq import heappop, heappush
 
 import grpc.aio
-from frequenz.channels import Sender, select, selected_from
+from frequenz.channels import Broadcast, Receiver, select, selected_from
 from frequenz.channels.timer import SkipMissedAndResync, Timer
 from frequenz.client.dispatch import Client
 from frequenz.client.dispatch.types import Event
@@ -50,24 +50,28 @@ def __init__(
         self,
         microgrid_id: int,
         client: Client,
-        lifecycle_updates_sender: Sender[DispatchEvent],
-        running_state_change_sender: Sender[Dispatch],
     ) -> None:
         """Initialize the actor.
 
         Args:
             microgrid_id: The microgrid ID to handle dispatches for.
             client: The client to use for fetching dispatches.
-            lifecycle_updates_sender: A sender for dispatch lifecycle events.
-            running_state_change_sender: A sender for dispatch running state changes.
         """
         super().__init__(name="dispatch")
 
         self._client = client
         self._dispatches: dict[int, Dispatch] = {}
         self._microgrid_id = microgrid_id
-        self._lifecycle_updates_sender = lifecycle_updates_sender
-        self._running_state_change_sender = running_state_change_sender
+
+        self._lifecycle_events_channel = Broadcast[DispatchEvent](
+            name="lifecycle_events"
+        )
+        self._lifecycle_events_tx = self._lifecycle_events_channel.new_sender()
+        self._running_state_status_channel = Broadcast[Dispatch](
+            name="running_state_status"
+        )
+
+        self._running_state_status_tx = self._running_state_status_channel.new_sender()
         self._next_event_timer = Timer(
             timedelta(seconds=100), SkipMissedAndResync(), auto_start=False
         )
@@ -84,6 +88,44 @@ def __init__(
         always at index 0.
         """
 
+    def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
+        """Create a new receiver for lifecycle events.
+
+        Args:
+            type: The type of events to receive.
+
+        Returns:
+            A new receiver for lifecycle events.
+        """
+        return self._lifecycle_events_channel.new_receiver().filter(
+            lambda event: event.dispatch.type == type
+        )
+
+    async def new_running_state_event_receiver(self, type: str) -> Receiver[Dispatch]:
+        """Create a new receiver for running state events.
+
+        Args:
+            type: The type of events to receive.
+
+        Returns:
+            A new receiver for running state status.
+        """
+        # Find all matching dispatches based on the type and collect them
+        dispatches = [
+            dispatch for dispatch in self._dispatches.values() if dispatch.type == type
+        ]
+
+        # Create receiver with enough capacity to hold all matching dispatches
+        receiver = self._running_state_status_channel.new_receiver(
+            limit=max(1, len(dispatches))
+        ).filter(lambda dispatch: dispatch.type == type)
+
+        # Send all matching dispatches to the receiver
+        for dispatch in dispatches:
+            await self._send_running_state_change(dispatch)
+
+        return receiver
+
     async def _run(self) -> None:
         """Run the actor."""
         _logger.info("Starting dispatch actor for microgrid %s", self._microgrid_id)
@@ -111,24 +153,18 @@ async def _run(self) -> None:
                     case Event.CREATED:
                         self._dispatches[dispatch.id] = dispatch
                         await self._update_dispatch_schedule_and_notify(dispatch, None)
-                        await self._lifecycle_updates_sender.send(
-                            Created(dispatch=dispatch)
-                        )
+                        await self._lifecycle_events_tx.send(Created(dispatch=dispatch))
                     case Event.UPDATED:
                         await self._update_dispatch_schedule_and_notify(
                             dispatch, self._dispatches[dispatch.id]
                         )
                         self._dispatches[dispatch.id] = dispatch
-                        await self._lifecycle_updates_sender.send(
-                            Updated(dispatch=dispatch)
-                        )
+                        await self._lifecycle_events_tx.send(Updated(dispatch=dispatch))
                     case Event.DELETED:
                         self._dispatches.pop(dispatch.id)
                         await self._update_dispatch_schedule_and_notify(None, dispatch)
 
-                        await self._lifecycle_updates_sender.send(
-                            Deleted(dispatch=dispatch)
-                        )
+                        await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
 
     async def _execute_scheduled_event(self, dispatch: Dispatch) -> None:
         """Execute a scheduled event.
@@ -170,17 +206,13 @@ async def _fetch(self) -> None:
                     if not old_dispatch:
                         _logger.info("New dispatch: %s", dispatch)
                         await self._update_dispatch_schedule_and_notify(dispatch, None)
-                        await self._lifecycle_updates_sender.send(
-                            Created(dispatch=dispatch)
-                        )
+                        await self._lifecycle_events_tx.send(Created(dispatch=dispatch))
                     elif dispatch.update_time != old_dispatch.update_time:
                         _logger.info("Updated dispatch: %s", dispatch)
                         await self._update_dispatch_schedule_and_notify(
                             dispatch, old_dispatch
                         )
-                        await self._lifecycle_updates_sender.send(
-                            Updated(dispatch=dispatch)
-                        )
+                        await self._lifecycle_events_tx.send(Updated(dispatch=dispatch))
 
         except grpc.aio.AioRpcError as error:
             _logger.error("Error fetching dispatches: %s", error)
@@ -189,13 +221,13 @@ async def _fetch(self) -> None:
 
         for dispatch in old_dispatches.values():
             _logger.info("Deleted dispatch: %s", dispatch)
-            await self._lifecycle_updates_sender.send(Deleted(dispatch=dispatch))
+            await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
             await self._update_dispatch_schedule_and_notify(None, dispatch)
 
             # Set deleted only here as it influences the result of dispatch.started
             # which is used in above in _running_state_change
             dispatch._set_deleted()  # pylint: disable=protected-access
-            await self._lifecycle_updates_sender.send(Deleted(dispatch=dispatch))
+            await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
 
     async def _update_dispatch_schedule_and_notify(
         self, dispatch: Dispatch | None, old_dispatch: Dispatch | None
@@ -359,4 +391,4 @@ async def _send_running_state_change(self, dispatch: Dispatch) -> None:
         Args:
             dispatch: The dispatch that changed.
         """
-        await self._running_state_change_sender.send(dispatch)
+        await self._running_state_status_tx.send(dispatch)