Add stop method for MergeNamed and Merge channels

Both channels stores async tasks that listen on the receivers.
If they are removed without stopping the receivers tasks,
then user gets error: Task was destroyed but it is pending!

`stop` method should be called when channel is no longer needed.
Destructor can't be async, so we do that with extra method `stop`.

Signed-off-by: ela-kotulska-frequenz <[email protected]>

Author: ela-kotulska-frequenz

RELEASE_NOTES.md

@@ -2,16 +2,10 @@
 
 ## Summary
 
-<!-- Here goes a general summary of what this release is about -->
-
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including if there are any depractions and what they should be replaced with -->
-
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* Add method to stop `Merge` and `MergeNamed`.
 
 ## Bug Fixes
-
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->

src/frequenz/channels/util/_merge.py

@@ -24,6 +24,9 @@ class Merge(Receiver[T]):
             # do something with msg
             pass
         ```
+
+        When `merge` is no longer needed, then it should be stopped using
+        `self.stop()` method. This will cleanup any internal pending async tasks.
     """
 
     def __init__(self, *args: Receiver[T]) -> None:
@@ -44,6 +47,13 @@ def __del__(self) -> None:
         for task in self._pending:
             task.cancel()
 
+    async def stop(self) -> None:
+        """Stop the `Merge` instance and cleanup any pending tasks."""
+        for task in self._pending:
+            task.cancel()
+        await asyncio.gather(*self._pending, return_exceptions=True)
+        self._pending = set()
+
     async def ready(self) -> None:
         """Wait until the receiver is ready with a value.
 

src/frequenz/channels/util/_merge_named.py

@@ -11,7 +11,11 @@
 
 
 class MergeNamed(Receiver[Tuple[str, T]]):
-    """Merge messages coming from multiple named channels into a single stream."""
+    """Merge messages coming from multiple named channels into a single stream.
+
+    When `MergeNamed` is no longer needed, then it should be stopped using
+    `self.stop()` method. This will cleanup any internal pending async tasks.
+    """
 
     def __init__(self, **kwargs: Receiver[T]) -> None:
         """Create a `MergeNamed` instance.
@@ -31,6 +35,13 @@ def __del__(self) -> None:
         for task in self._pending:
             task.cancel()
 
+    async def stop(self) -> None:
+        """Stop the `MergeNamed` instance and cleanup any pending tasks."""
+        for task in self._pending:
+            task.cancel()
+        await asyncio.gather(*self._pending, return_exceptions=True)
+        self._pending = set()
+
     async def ready(self) -> None:
         """Wait until there's a message in any of the channels.