Add a LatestValueCache that can be used to replace Peekable

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/sdk/_internal/_channels.py

@@ -4,6 +4,7 @@
 """General purpose classes for use with channels."""
 
 import abc
+import asyncio
 import typing
 
 from frequenz.channels import Receiver
@@ -24,3 +25,26 @@ def new_receiver(self, maxsize: int = 50) -> Receiver[T]:
         Returns:
             A receiver instance.
         """
+
+
+class LatestValueCache(typing.Generic[T]):
+    """A cache that stores the latest value in a receiver."""
+
+    def __init__(self, receiver: Receiver[T]) -> None:
+        """Create a new cache.
+
+        Args:
+            receiver: The receiver to cache.
+        """
+        self._receiver = receiver
+        self._latest_value: T | None = None
+        self._task = asyncio.create_task(self._run())
+
+    @property
+    def latest_value(self) -> T | None:
+        """Get the latest value in the cache."""
+        return self._latest_value
+
+    async def _run(self) -> None:
+        async for value in self._receiver:
+            self._latest_value = value

src/frequenz/sdk/microgrid/client/_client.py

@@ -425,7 +425,7 @@ def _get_component_data_channel(
         if component_id in self._component_streams:
             return self._component_streams[component_id]
         task_name = f"raw-component-data-{component_id}"
-        chan = Broadcast[_GenericComponentData](task_name)
+        chan = Broadcast[_GenericComponentData](task_name, resend_latest=True)
         self._component_streams[component_id] = chan
 
         self._streaming_tasks[component_id] = asyncio.create_task(