Make merge and Merger covariant

When merging, we should allow containers of more specific types to be
mixed with less specific types as long as the result of the merge uses
the less specific type, so we use a covariant type variable for it.

We also use the new public type variables.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/channels/_merge.py

@@ -52,14 +52,13 @@
 import asyncio
 import itertools
 from collections import deque
-from typing import Any, TypeVar
+from typing import Any
 
+from ._generic import ReceiverMessageT_co
 from ._receiver import Receiver, ReceiverStoppedError
 
-_T = TypeVar("_T")
 
-
-def merge(*receivers: Receiver[_T]) -> Merger[_T]:
+def merge(*receivers: Receiver[ReceiverMessageT_co]) -> Merger[ReceiverMessageT_co]:
     """Merge messages coming from multiple receivers into a single stream.
 
     Example:
@@ -95,31 +94,33 @@ def merge(*receivers: Receiver[_T]) -> Merger[_T]:
     return Merger(*receivers, name="merge")
 
 
-class Merger(Receiver[_T]):
+class Merger(Receiver[ReceiverMessageT_co]):
     """A receiver that merges messages coming from multiple receivers into a single stream.
 
     Tip:
         Please consider using the more idiomatic [`merge()`][frequenz.channels.merge]
         function instead of creating a `Merger` instance directly.
     """
 
-    def __init__(self, *receivers: Receiver[_T], name: str | None) -> None:
+    def __init__(
+        self, *receivers: Receiver[ReceiverMessageT_co], name: str | None
+    ) -> None:
         """Initialize this merger.
 
         Args:
             *receivers: The receivers to merge.
             name: The name of the receiver. Used to create the string representation
                 of the receiver.
         """
-        self._receivers: dict[str, Receiver[_T]] = {
+        self._receivers: dict[str, Receiver[ReceiverMessageT_co]] = {
             str(id): recv for id, recv in enumerate(receivers)
         }
         self._name: str = name if name is not None else type(self).__name__
         self._pending: set[asyncio.Task[Any]] = {
             asyncio.create_task(anext(recv), name=name)
             for name, recv in self._receivers.items()
         }
-        self._results: deque[_T] = deque(maxlen=len(self._receivers))
+        self._results: deque[ReceiverMessageT_co] = deque(maxlen=len(self._receivers))
 
     def __del__(self) -> None:
         """Finalize this merger."""
@@ -170,7 +171,7 @@ async def ready(self) -> bool:
                     asyncio.create_task(anext(self._receivers[name]), name=name)
                 )
 
-    def consume(self) -> _T:
+    def consume(self) -> ReceiverMessageT_co:
         """Return the latest message once `ready` is complete.
 
         Returns: