Make Merge public again

The `Merge` receiver was made private to try to hide the implementation
details to the user, but it was a bad idea because the `Merge` receiver
has more methods than a plain `Receiver`, in particular `stop()`, which
users need to use for guaranteed cleanup.

This also means now the `merge()` function returns the `Merge` instance
instead of an abstract `Receiver`.

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

Author: llucax

src/frequenz/channels/__init__.py

@@ -77,7 +77,7 @@
 from ._anycast import Anycast
 from ._broadcast import Broadcast
 from ._exceptions import ChannelClosedError, ChannelError, Error
-from ._merge import merge
+from ._merge import Merger, merge
 from ._receiver import Receiver, ReceiverError, ReceiverStoppedError
 from ._select import (
     Selected,
@@ -95,6 +95,7 @@
     "ChannelClosedError",
     "ChannelError",
     "Error",
+    "Merger",
     "Receiver",
     "ReceiverError",
     "ReceiverStoppedError",

src/frequenz/channels/_merge.py

@@ -3,6 +3,8 @@
 
 """Merge messages coming from channels into a single stream."""
 
+from __future__ import annotations
+
 import asyncio
 import itertools
 from collections import deque
@@ -13,7 +15,7 @@
 _T = TypeVar("_T")
 
 
-def merge(*receivers: Receiver[_T]) -> Receiver[_T]:
+def merge(*receivers: Receiver[_T]) -> Merger[_T]:
     """Merge messages coming from multiple receivers into a single stream.
 
     Example:
@@ -46,19 +48,19 @@ def merge(*receivers: Receiver[_T]) -> Receiver[_T]:
     if not receivers:
         raise ValueError("At least one receiver must be provided")
 
-    # This is just a small optimization to avoid creating a merge receiver when it is
-    # not really needed.
-    if len(receivers) == 1:
-        return receivers[0]
+    return Merger(*receivers, name="merge")
 
-    return _Merge(*receivers, name="merge")
 
+class Merger(Receiver[_T]):
+    """A receiver that merges messages coming from multiple receivers into a single stream.
 
-class _Merge(Receiver[_T]):
-    """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:
-        """Create a `_Merge` instance.
+        """Create a `Merger` instance.
 
         Args:
             *receivers: The receivers to merge.
@@ -82,7 +84,7 @@ def __del__(self) -> None:
                 task.cancel()
 
     async def stop(self) -> None:
-        """Stop the `_Merge` instance and cleanup any pending tasks."""
+        """Stop the `Merger` instance and cleanup any pending tasks."""
         for task in self._pending:
             task.cancel()
         await asyncio.gather(*self._pending, return_exceptions=True)

tests/test_merge.py

@@ -3,22 +3,12 @@
 
 """Tests for the merge implementation."""
 
-from unittest import mock
-
 import pytest
 
-from frequenz.channels import Receiver, merge
+from frequenz.channels import merge
 
 
 async def test_empty() -> None:
     """Ensure merge() raises an exception when no receivers are provided."""
     with pytest.raises(ValueError, match="At least one receiver must be provided"):
         merge()
-
-
-async def test_one() -> None:
-    """Ensure merge() returns the same receiver when only one is provided."""
-    receiver = mock.MagicMock(spec=Receiver[int])
-
-    merge_receiver: Receiver[int] = merge(receiver)
-    assert merge_receiver is receiver