Replace Merge and MergeNamed with merge() (#238)

- Remove `MergeNamed`
- Add a `merge()` wrapper function that replaces `Merge`
- Rename `Merge` to `_Merge` to make it private more explicitly
- Rename `_Merge` variable `args` to `receivers`
- Improve string representation of merge receiver
- Raise an exception is `merge()` is called with less than two receivers

Fixes #236, fixes #237.

Author: llucax

.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

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.

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",
 ]

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

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

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())})"
         )

src/frequenz/channels/_merge_named.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