Repository: frequenz-floss/frequenz-channels-python (language: python)

Issue:
Remove `MergeNamed`
### What's needed?

We already have 2 ways to join multiple receivers, `Merge` and `select`. `MergeNamed` is a mixture of both, so it is redundant.

### Proposed solution

Remove `MergeNamed`.

### Use cases

_No response_

### Alternatives and workarounds

_No response_

### Additional context

_No response_
Expose `Merge` as a `merge()` function
### What's needed?

There is not much use in having `Merge` as a class from a user point of view, users normally only care about merging some channels, so having a class just makes the syntax more weird for most use cases that will be `async for msg in merge(...)`.

### Proposed solution

Make `Merge` private (`_Merge`) and create a new function that just does `return _Merge(...)`.

### Use cases

_No response_

### Alternatives and workarounds

_No response_

### Additional context

_No response_


Discussion / hints:

Fix patch:
diff --git a/.github/ISSUE_TEMPLATE/bug.yml b/.github/ISSUE_TEMPLATE/bug.yml
index bd567007..4f7724b0 100644
--- a/.github/ISSUE_TEMPLATE/bug.yml
+++ b/.github/ISSUE_TEMPLATE/bug.yml
@@ -52,7 +52,7 @@ body:
         - Build script, CI, dependencies, etc. (part:tooling)
         - Channels, `Broadcast`, `Anycast`, etc. (part:channels)
         - Select (part:select)
-        - Utility receivers, `Merge`, etc. (part:receivers)
+        - Utility receivers, `Event`, `FileWatcher`, `Timer`, etc. (part:receivers)
     validations:
       required: true
   - type: textarea
diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
index fd869421..9e8ae4c4 100644
--- a/RELEASE_NOTES.md
+++ b/RELEASE_NOTES.md
@@ -53,8 +53,6 @@
 
 * The following symbols were moved to the top-level `frequenz.channels` package:
 
-  - `Merge`
-  - `MergeNamed`
   - `Selected`
   - `SelectError`
   - `SelectErrorGroup`
@@ -68,6 +66,14 @@
 
   This channel was removed as it is not recommended practice and was a niche use case. If you need to use it, you can set up two channels or copy the `Bidirectional` class from the previous version to your project.
 
+* `Merge`
+
+  Replaced by the new `merge()` function. When replacing `Merge` with `merge()` please keep in mind that this new function will raise a `ValueError` if no receivers are passed to it.
+
+* `MergeNamed`
+
+  This class was redundant, use either the new `merge()` function or `select()` instead.
+
 * `Peekable`
 
   This class was removed because it was merely a shortcut to a receiver that caches the last value received. It did not fit the channel abstraction well and was infrequently used.
@@ -92,6 +98,8 @@
 
 ## New Features
 
+* A new `merge()` function was added to replace `Merge`.
+
 * `Anycast`
 
   - The following new read-only properties were added:
@@ -126,14 +134,6 @@
 
   - A more useful implementation of `__str__ and `__repr__` were added.
 
-* `Merge`
-
-  - A more useful implementation of `__str__ and `__repr__` were added.
-
-* `MergeNamed`
-
-  - A more useful implementation of `__str__ and `__repr__` were added.
-
 * `Peekable`
 
   - A more useful implementation of `__str__ and `__repr__` were added.
diff --git a/src/frequenz/channels/__init__.py b/src/frequenz/channels/__init__.py
index dbf60751..a4f17426 100644
--- a/src/frequenz/channels/__init__.py
+++ b/src/frequenz/channels/__init__.py
@@ -26,11 +26,10 @@
 
 Utilities to work with channels:
 
-* [Merge][frequenz.channels.Merge] and [MergeNamed][frequenz.channels.MergeNamed]:
-  [Receivers][frequenz.channels.Receiver] that merge messages coming from multiple
-  receivers into a single stream.
+* [merge][frequenz.channels.merge]: Merge messages coming from multiple receivers into
+  a single stream.
 
-* [select][frequenz.channels.select]:  Iterate over the values of all
+* [select][frequenz.channels.select]: Iterate over the values of all
   [receivers][frequenz.channels.Receiver] as new values become available.
 
 Exception classes:
@@ -78,8 +77,7 @@
 from ._anycast import Anycast
 from ._broadcast import Broadcast
 from ._exceptions import ChannelClosedError, ChannelError, Error
-from ._merge import Merge
-from ._merge_named import MergeNamed
+from ._merge import merge
 from ._receiver import Receiver, ReceiverError, ReceiverStoppedError
 from ._select import (
     Selected,
@@ -97,8 +95,6 @@
     "ChannelClosedError",
     "ChannelError",
     "Error",
-    "Merge",
-    "MergeNamed",
     "Receiver",
     "ReceiverError",
     "ReceiverStoppedError",
@@ -108,6 +104,7 @@
     "Sender",
     "SenderError",
     "UnhandledSelectedError",
+    "merge",
     "select",
     "selected_from",
 ]
diff --git a/src/frequenz/channels/_anycast.py b/src/frequenz/channels/_anycast.py
index 952aa12f..2a5ac56f 100644
--- a/src/frequenz/channels/_anycast.py
+++ b/src/frequenz/channels/_anycast.py
@@ -38,9 +38,8 @@ class Anycast(Generic[_T]):
     thread-safe.
 
     When there are multiple channel receivers, they can be awaited
-    simultaneously using [select][frequenz.channels.select],
-    [Merge][frequenz.channels.Merge] or
-    [MergeNamed][frequenz.channels.MergeNamed].
+    simultaneously using [select][frequenz.channels.select] or
+    [merge][frequenz.channels.merge].
 
     Example:
         ``` python
diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 327d3a0c..8da322b1 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -32,9 +32,8 @@ class Broadcast(Generic[_T]):
     are thread-safe.  Because of this, `Broadcast` channels are thread-safe.
 
     When there are multiple channel receivers, they can be awaited
-    simultaneously using [select][frequenz.channels.select],
-    [Merge][frequenz.channels.Merge] or
-    [MergeNamed][frequenz.channels.MergeNamed].
+    simultaneously using [select][frequenz.channels.select] or
+    [merge][frequenz.channels.merge].
 
     Example:
         ``` python
diff --git a/src/frequenz/channels/_merge.py b/src/frequenz/channels/_merge.py
index 00461ed9..c78396d6 100644
--- a/src/frequenz/channels/_merge.py
+++ b/src/frequenz/channels/_merge.py
@@ -13,13 +13,13 @@
 _T = TypeVar("_T")
 
 
-class Merge(Receiver[_T]):
-    """Merge messages coming from multiple channels into a single stream.
+def merge(*receivers: Receiver[_T]) -> Receiver[_T]:
+    """Merge messages coming from multiple receivers into a single stream.
 
     Example:
         For example, if there are two channel receivers with the same type,
         they can be awaited together, and their results merged into a single
-        stream, by using `Merge` like this:
+        stream like this:
 
         ```python
         from frequenz.channels import Broadcast
@@ -29,25 +29,46 @@ class Merge(Receiver[_T]):
         receiver1 = channel1.new_receiver()
         receiver2 = channel2.new_receiver()
 
-        merge = Merge(receiver1, receiver2)
-        while msg := await merge.receive():
-            # do something with msg
-            pass
+        async for msg in merge(receiver1, receiver2):
+            print(f"received {msg}")
         ```
 
-        When `merge` is no longer needed, then it should be stopped using
-        `self.stop()` method. This will cleanup any internal pending async tasks.
+    Args:
+        *receivers: The receivers to merge.
+
+    Returns:
+        A receiver that merges the messages coming from multiple receivers into a
+            single stream.
+
+    Raises:
+        ValueError: if no receivers are provided.
     """
+    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 _Merge(*receivers, name="merge")
+
+
+class _Merge(Receiver[_T]):
+    """A receiver that merges messages coming from multiple receivers into a single stream."""
 
-    def __init__(self, *args: Receiver[_T]) -> None:
-        """Create a `Merge` instance.
+    def __init__(self, *receivers: Receiver[_T], name: str | None) -> None:
+        """Create a `_Merge` instance.
 
         Args:
-            *args: sequence of channel receivers.
+            *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]] = {
-            str(id): recv for id, recv in enumerate(args)
+            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()
@@ -61,7 +82,7 @@ def __del__(self) -> None:
                 task.cancel()
 
     async def stop(self) -> None:
-        """Stop the `Merge` instance and cleanup any pending tasks."""
+        """Stop the `_Merge` instance and cleanup any pending tasks."""
         for task in self._pending:
             task.cancel()
         await asyncio.gather(*self._pending, return_exceptions=True)
@@ -127,11 +148,11 @@ def __str__(self) -> str:
             receivers.append("…")
         else:
             receivers = [str(p) for p in self._receivers.values()]
-        return f"{type(self).__name__}:{','.join(receivers)}"
+        return f"{self._name}:{','.join(receivers)}"
 
     def __repr__(self) -> str:
         """Return a string representation of this receiver."""
         return (
-            f"{type(self).__name__}("
+            f"{self._name}("
             f"{', '.join(f'{k}={v!r}' for k, v in self._receivers.items())})"
         )
diff --git a/src/frequenz/channels/_merge_named.py b/src/frequenz/channels/_merge_named.py
deleted file mode 100644
index fb08c867..00000000
--- a/src/frequenz/channels/_merge_named.py
+++ /dev/null
@@ -1,121 +0,0 @@
-# License: MIT
-# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
-
-"""Merge messages coming from channels into a single stream containing name of message."""
-import asyncio
-import itertools
-from collections import deque
-from typing import Any, TypeVar
-
-from ._receiver import Receiver, ReceiverStoppedError
-
-_T = TypeVar("_T")
-
-
-class MergeNamed(Receiver[tuple[str, _T]]):
-    """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.
-
-        Args:
-            **kwargs: sequence of channel receivers.
-        """
-        self._receivers: dict[str, Receiver[_T]] = kwargs
-        """The sequence of channel receivers to get the messages to merge."""
-
-        self._pending: set[asyncio.Task[Any]] = {
-            asyncio.create_task(anext(recv), name=name)
-            for name, recv in self._receivers.items()
-        }
-        """The set of pending tasks to merge messages."""
-
-        self._results: deque[tuple[str, _T]] = deque(maxlen=len(self._receivers))
-        """The internal buffer of merged messages."""
-
-    def __del__(self) -> None:
-        """Cleanup any pending tasks."""
-        for task in self._pending:
-            if not task.done() and task.get_loop().is_running():
-                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) -> bool:
-        """Wait until the receiver is ready with a value or an error.
-
-        Once a call to `ready()` has finished, the value should be read with
-        a call to `consume()` (`receive()` or iterated over). The receiver will
-        remain ready (this method will return immediately) until it is
-        consumed.
-
-        Returns:
-            Whether the receiver is still active.
-        """
-        # we use a while loop to continue to wait for new data, in case the
-        # previous `wait` completed because a channel was closed.
-        while True:
-            # if there are messages waiting to be consumed, return immediately.
-            if len(self._results) > 0:
-                return True
-
-            # if there are no more pending receivers, we return immediately.
-            if len(self._pending) == 0:
-                return False
-
-            done, self._pending = await asyncio.wait(
-                self._pending, return_when=asyncio.FIRST_COMPLETED
-            )
-            for item in done:
-                name = item.get_name()
-                # if channel is closed, don't add a task for it again.
-                if isinstance(item.exception(), StopAsyncIteration):
-                    continue
-                result = item.result()
-                self._results.append((name, result))
-                self._pending.add(
-                    # pylint: disable=unnecessary-dunder-call
-                    asyncio.create_task(anext(self._receivers[name]), name=name)
-                )
-
-    def consume(self) -> tuple[str, _T]:
-        """Return the latest value once `ready` is complete.
-
-        Returns:
-            The next key, value that was received.
-
-        Raises:
-            ReceiverStoppedError: if the receiver stopped producing messages.
-            ReceiverError: if there is some problem with the receiver.
-        """
-        if not self._results and not self._pending:
-            raise ReceiverStoppedError(self)
-
-        assert self._results, "`consume()` must be preceded by a call to `ready()`"
-
-        return self._results.popleft()
-
-    def __str__(self) -> str:
-        """Return a string representation of this receiver."""
-        if len(self._receivers) > 3:
-            receivers = [str(p) for p in itertools.islice(self._receivers, 3)]
-            receivers.append("…")
-        else:
-            receivers = [str(p) for p in self._receivers]
-        return f"{type(self).__name__}:{','.join(receivers)}"
-
-    def __repr__(self) -> str:
-        """Return a string representation of this receiver."""
-        return (
-            f"{type(self).__name__}("
-            f"{', '.join(f'{k}={v!r}' for k, v in self._receivers.items())})"
-        )
diff --git a/src/frequenz/channels/_select.py b/src/frequenz/channels/_select.py
index a7ec4c90..2d30bd81 100644
--- a/src/frequenz/channels/_select.py
+++ b/src/frequenz/channels/_select.py
@@ -258,10 +258,8 @@ async def select(*receivers: Receiver[Any]) -> AsyncIterator[Selected[Any]]:
         receivers from a select loop, there are a few alternatives.  Depending on your
         use case, one or the other could work better for you:
 
-        * Use [`Merge`][frequenz.channels.Merge] or
-          [`MergeNamed`][frequenz.channels.MergeNamed]: this is useful when you
-          have and unknown number of receivers of the same type that can be handled as
-          a group.
+        * Use [`merge()`][frequenz.channels.merge]: this is useful when you have an
+          unknown number of receivers of the same type that can be handled as a group.
         * Use tasks to manage each receiver individually: this is better if there are no
           relationships between the receivers.
         * Break the `select()` loop and start a new one with the new set of receivers


Test patch:
diff --git a/tests/test_merge.py b/tests/test_merge.py
index 5f6b14a5..a96dbcec 100644
--- a/tests/test_merge.py
+++ b/tests/test_merge.py
@@ -1,39 +1,24 @@
 # License: MIT
 # Copyright © 2022 Frequenz Energy-as-a-Service GmbH
 
-"""Tests for the Merge implementation."""
-
-import asyncio
-
-from frequenz.channels import Anycast, Merge, Sender
-
-
-async def test_merge() -> None:
-    """Ensure Merge receives messages in order."""
-    chan1 = Anycast[int](name="chan1")
-    chan2 = Anycast[int](name="chan2")
-
-    async def send(ch1: Sender[int], ch2: Sender[int]) -> None:
-        for ctr in range(5):
-            await ch1.send(ctr + 1)
-            await ch2.send(ctr + 101)
-        await chan1.close()
-        await ch2.send(1000)
-        await chan2.close()
-
-    senders = asyncio.create_task(send(chan1.new_sender(), chan2.new_sender()))
-
-    merge = Merge(chan1.new_receiver(), chan2.new_receiver())
-    results: list[int] = []
-    async for item in merge:
-        results.append(item)
-    await senders
-    for ctr in range(5):
-        idx = ctr * 2
-        # It is hard to get messages from multiple channels in the same order,
-        # so we use a `set` to check the next N messages are the same, in any
-        # order, where N is the number of channels.  This only works in this
-        # example because the `send` method sends values in immediate
-        # succession.
-        assert set(results[idx : idx + 2]) == {ctr + 1, ctr + 101}
-    assert results[-1] == 1000
+"""Tests for the merge implementation."""
+
+from unittest import mock
+
+import pytest
+
+from frequenz.channels import Receiver, 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
diff --git a/tests/test_mergenamed.py b/tests/test_merge_integration.py
similarity index 63%
rename from tests/test_mergenamed.py
rename to tests/test_merge_integration.py
index 03d272ea..a6313fd5 100644
--- a/tests/test_mergenamed.py
+++ b/tests/test_merge_integration.py
@@ -1,15 +1,18 @@
 # License: MIT
 # Copyright © 2022 Frequenz Energy-as-a-Service GmbH
 
-"""Tests for the MergeNamed implementation."""
+"""Integration tests for the merge implementation."""
 
 import asyncio
 
-from frequenz.channels import Anycast, MergeNamed, Sender
+import pytest
 
+from frequenz.channels import Anycast, Sender, merge
 
-async def test_mergenamed() -> None:
-    """Ensure MergeNamed receives messages in order."""
+
+@pytest.mark.integration
+async def test_merge() -> None:
+    """Ensure merge() receives messages in order."""
     chan1 = Anycast[int](name="chan1")
     chan2 = Anycast[int](name="chan2")
 
@@ -22,11 +25,9 @@ async def send(ch1: Sender[int], ch2: Sender[int]) -> None:
         await chan2.close()
 
     senders = asyncio.create_task(send(chan1.new_sender(), chan2.new_sender()))
-    recvs = {"chan1": chan1.new_receiver(), "chan2": chan2.new_receiver()}
 
-    merge = MergeNamed(**recvs)
-    results: list[tuple[str, int]] = []
-    async for item in merge:
+    results: list[int] = []
+    async for item in merge(chan1.new_receiver(), chan2.new_receiver()):
         results.append(item)
     await senders
     for ctr in range(5):
@@ -36,8 +37,5 @@ async def send(ch1: Sender[int], ch2: Sender[int]) -> None:
         # order, where N is the number of channels.  This only works in this
         # example because the `send` method sends values in immediate
         # succession.
-        assert set(results[idx : idx + 2]) == {
-            ("chan1", ctr + 1),
-            ("chan2", ctr + 101),
-        }
-    assert results[-1] == ("chan2", 1000)
+        assert set(results[idx : idx + 2]) == {ctr + 1, ctr + 101}
+    assert results[-1] == 1000