Add a ReceiverFetcherWith implementation

This is a `ReceiverFetcher` which can be used to modify newly created
receivers, for example by applying a `map` or a `filter` on them,
before returning them.

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

Author: shsms

src/frequenz/sdk/_internal/_channels.py

@@ -8,10 +8,12 @@
 import typing
 
 from frequenz.channels import Receiver
+from typing_extensions import override
 
 from ._asyncio import cancel_and_await
 
 T_co = typing.TypeVar("T_co", covariant=True)
+U_co = typing.TypeVar("U_co", covariant=True)
 
 
 class ReceiverFetcher(typing.Generic[T_co], typing.Protocol):
@@ -29,6 +31,37 @@ def new_receiver(self, *, limit: int = 50) -> Receiver[T_co]:
         """
 
 
+class ReceiverFetcherWith(ReceiverFetcher[U_co], typing.Generic[T_co, U_co]):
+    """A receiver fetcher that can manipulate receivers before returning them."""
+
+    def __init__(
+        self,
+        fetcher: ReceiverFetcher[T_co],
+        manipulator: typing.Callable[[Receiver[T_co]], Receiver[U_co]],
+    ) -> None:
+        """Initialize this instance.
+
+        Args:
+            fetcher: The underlying fetcher to get receivers from.
+            manipulator: The method to be applied on new receivers before returning
+                them.
+        """
+        self._fetcher = fetcher
+        self._manipulator = manipulator
+
+    @override
+    def new_receiver(self, *, limit: int = 50) -> Receiver[U_co]:
+        """Get a receiver from the channel.
+
+        Args:
+            limit: The maximum size of the receiver.
+
+        Returns:
+            A receiver instance.
+        """
+        return self._manipulator(self._fetcher.new_receiver(limit=limit))
+
+
 class _Sentinel:
     """A sentinel to denote that no value has been received yet."""