Clean up public API (#55)

This PR hides all internal modules by prepending a `_` to the module
name.

It also expose anything that is not a channel (or core channel classes)
in
a `util` package. So this is how classes are exposed:

* `frequenz.channels.Anycast`
* `frequenz.channels.Broadcast`
* `frequenz.channels.Anycast`
* `frequenz.channels.Bidirectional`
* `frequenz.channels.Broadcast`
* `frequenz.channels.Peekable`
* `frequenz.channels.Receiver`
* `frequenz.channels.Sender`
* `frequenz.channels.util.Merge`
* `frequenz.channels.util.MergeNamed`
* `frequenz.channels.util.FileWatcher`
* `frequenz.channels.util.Select`
* `frequenz.channels.util.Timer`

This also completely removes `BufferedReceiver` and moves
`BidirectionalHandle`
to `Bidirectional.Handle` and `EventType` to `FileWatcher.EventType`.

While at it, add also a short summary of each module contents.

Author: llucax

.github/labeler.yml

@@ -30,11 +30,12 @@
 "part:channels":
   - any:
       - "src/frequenz/channels/**"
-      - "!src/frequenz/channels/select.py"
       - "!src/frequenz/channels/util/**"
 
 "part:receivers":
-  - "src/frequenz/channels/util/**"
+  - any:
+      - "src/frequenz/channels/util/**"
+      - "!src/frequenz/channels/util/_select.py"
 
 "part:select":
-  - "src/frequenz/channels/select.py"
+  - "src/frequenz/channels/util/_select.py"

RELEASE_NOTES.md

@@ -18,6 +18,33 @@ time.
   `new_receiver()` and `new_sender()` respectively. This is to make it more
   clear that new objects are being created.
 
+* The public API surface has been reduced considerably to make it more clear
+  where to import symbols.  You should update your imports.  The new symbol
+  locations are:
+
+  * `frequenz.channels.Anycast`
+  * `frequenz.channels.Broadcast`
+  * `frequenz.channels.Anycast`
+  * `frequenz.channels.Bidirectional`
+  * `frequenz.channels.Broadcast`
+  * `frequenz.channels.Peekable`
+  * `frequenz.channels.Receiver`
+  * `frequenz.channels.Sender`
+  * `frequenz.channels.util.Merge`
+  * `frequenz.channels.util.MergeNamed`
+  * `frequenz.channels.util.FileWatcher`
+  * `frequenz.channels.util.Select`
+  * `frequenz.channels.util.Timer`
+
+* The class `BufferedReceiver` was removed because the interface was really
+  intended for channel implementations. Users are not supposed to enqueue
+  messages to receiver but just receive from them. If you used it you can
+  implement it yourself.
+
+* The class `BidirectionalHandle` was moved to `Bidirectional.Handle`.
+
+* The class `EventType` was moved to `FileWatcher.EventType`.
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->

docs/mkdocstrings_autoapi.py

@@ -8,12 +8,30 @@
 """
 
 from pathlib import Path
+from typing import Tuple
 
 import mkdocs_gen_files
 
 SRC_PATH = "src"
 DST_PATH = "reference"
 
+
+def is_internal(path_parts: Tuple[str, ...]) -> bool:
+    """Tell if the path is internal judging by the parts.
+
+    Args:
+        path_parts: Path.parts of the path to check.
+
+    Returns:
+        True if the path is internal.
+    """
+
+    def with_underscore_not_init(part: str) -> bool:
+        return part.startswith("_") and part != "__init__"
+
+    return any(p for p in path_parts if with_underscore_not_init(p))
+
+
 # type ignore because mkdocs_gen_files uses a very weird module-level
 # __getattr__() which messes up the type system
 nav = mkdocs_gen_files.Nav()  # type: ignore
@@ -24,15 +42,17 @@
     doc_path = path.relative_to(SRC_PATH).with_suffix(".md")
     full_doc_path = Path(DST_PATH, doc_path)
     parts = tuple(module_path.parts)
+    if is_internal(parts):
+        continue
     if parts[-1] == "__init__":
         doc_path = doc_path.with_name("index.md")
         full_doc_path = full_doc_path.with_name("index.md")
         parts = parts[:-1]
 
     nav[parts] = doc_path.as_posix()
 
-    with mkdocs_gen_files.open(full_doc_path, "w") as fd:
-        fd.write(f"::: {'.'.join(parts)}\n")
+    with mkdocs_gen_files.open(full_doc_path, "w") as output_file:
+        output_file.write(f"::: {'.'.join(parts)}\n")
 
     mkdocs_gen_files.set_edit_path(full_doc_path, Path("..") / path)
 

src/frequenz/channels/__init__.py

@@ -1,30 +1,64 @@
 # License: MIT
 # Copyright © 2022 Frequenz Energy-as-a-Service GmbH
 
-"""Channel implementations."""
-
-from frequenz.channels.anycast import Anycast
-from frequenz.channels.base_classes import BufferedReceiver, Peekable, Receiver, Sender
-from frequenz.channels.bidirectional import Bidirectional, BidirectionalHandle
-from frequenz.channels.broadcast import Broadcast
-from frequenz.channels.merge import Merge
-from frequenz.channels.merge_named import MergeNamed
-from frequenz.channels.select import Select
-from frequenz.channels.utils.file_watcher import FileWatcher
-from frequenz.channels.utils.timer import Timer
+"""Frequenz Channels.
+
+This package contains
+[channel](https://en.wikipedia.org/wiki/Channel_(programming)) implementations.
+
+Channels:
+
+* [Anycast][frequenz.channels.Anycast]: A channel that supports multiple
+  senders and multiple receivers.  A message sent through a sender will be
+  received by exactly one receiver.
+
+* [Bidirectional][frequenz.channels.Bidirectional]: A channel providing
+  a `client` and a `service` handle to send and receive bidirectionally.
+
+* [Broadcast][frequenz.channels.Broadcast]: A channel to broadcast messages
+  from multiple senders to multiple receivers. Each message sent through any of
+  the senders is received by all of the receivers.
+
+Other base classes:
+
+* [Peekable][frequenz.channels.Peekable]: An object to allow users to get
+  a peek at the latest value in the channel, without consuming anything.
+
+* [Receiver][frequenz.channels.Receiver]: An object that can wait for and
+  consume messages from a channel.
+
+* [Sender][frequenz.channels.Sender]: An object that can send messages to
+  a channel.
+
+Utilities:
+
+* [util][frequenz.channels.util]: A module with utilities, like special
+  receivers that implement timers, file watchers, merge receivers, or wait for
+  messages in multiple channels.
+
+Exception classes:
+
+* [ChannelError][frequenz.channels.ChannelError]: Base class for all errors
+  related to channels.
+
+* [ChannelClosedError][frequenz.channels.ChannelClosedError]: Error raised when
+  trying to operate (send, receive, etc.) through a closed channel.
+"""
+
+from . import util
+from ._anycast import Anycast
+from ._base_classes import ChannelClosedError, ChannelError, Peekable, Receiver, Sender
+from ._bidirectional import Bidirectional
+from ._broadcast import Broadcast
 
 __all__ = [
     "Anycast",
     "Bidirectional",
-    "BidirectionalHandle",
     "Broadcast",
-    "BufferedReceiver",
-    "FileWatcher",
-    "Merge",
-    "MergeNamed",
+    "ChannelClosedError",
+    "ChannelError",
     "Peekable",
     "Receiver",
-    "Select",
     "Sender",
-    "Timer",
+    "util",
 ]

src/frequenz/channels/anycast.py src/frequenz/channels/_anycast.pysrc/frequenz/channels/anycast.py renamed to src/frequenz/channels/_anycast.py

@@ -9,10 +9,10 @@
 from collections import deque
 from typing import Deque, Generic, Optional
 
-from frequenz.channels.base_classes import ChannelClosedError
-from frequenz.channels.base_classes import Receiver as BaseReceiver
-from frequenz.channels.base_classes import Sender as BaseSender
-from frequenz.channels.base_classes import T
+from ._base_classes import ChannelClosedError
+from ._base_classes import Receiver as BaseReceiver
+from ._base_classes import Sender as BaseSender
+from ._base_classes import T
 
 
 class Anycast(Generic[T]):
@@ -28,9 +28,9 @@ 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.util.Select],
+    [Merge][frequenz.channels.util.Merge] or
+    [MergeNamed][frequenz.channels.util.MergeNamed].
 
     Example:
         ``` python

src/frequenz/channels/base_classes.py src/frequenz/channels/_base_classes.pysrc/frequenz/channels/base_classes.py renamed to src/frequenz/channels/_base_classes.py

@@ -164,18 +164,6 @@ def peek(self) -> Optional[T]:
         """
 
 
-class BufferedReceiver(Receiver[T]):
-    """A channel receiver with a buffer."""
-
-    @abstractmethod
-    def enqueue(self, msg: T) -> None:
-        """Put a message into this buffered receiver's queue.
-
-        Args:
-            msg: The message to be added to the queue.
-        """
-
-
 class _Map(Receiver[U], Generic[T, U]):
     """Apply a transform function on a channel receiver.
 

src/frequenz/channels/_bidirectional.py

@@ -0,0 +1,97 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""An abstraction to provide bi-directional communication between actors."""
+
+from __future__ import annotations
+
+from typing import Generic, TypeVar
+
+from ._base_classes import Receiver, Sender, T, U
+from ._broadcast import Broadcast
+
+V = TypeVar("V")
+W = TypeVar("W")
+
+
+class Bidirectional(Generic[T, U]):
+    """A wrapper class for simulating bidirectional channels."""
+
+    class Handle(Sender[V], Receiver[W]):
+        """A handle to a [Bidirectional][frequenz.channels.Bidirectional] instance.
+
+        It can be used to send/receive values between the client and service.
+        """
+
+        def __init__(self, sender: Sender[V], receiver: Receiver[W]) -> None:
+            """Create a `Bidirectional.Handle` instance.
+
+            Args:
+                sender: A sender to send values with.
+                receiver: A receiver to receive values from.
+            """
+            self._sender = sender
+            self._receiver = receiver
+
+        async def send(self, msg: V) -> bool:
+            """Send a value to the other side.
+
+            Args:
+                msg: The value to send.
+
+            Returns:
+                Whether the send was successful or not.
+            """
+            return await self._sender.send(msg)
+
+        async def ready(self) -> None:
+            """Wait until the receiver is ready with a value."""
+            await self._receiver.ready()  # pylint: disable=protected-access
+
+        def consume(self) -> W:
+            """Return the latest value once `_ready` is complete.
+
+            Returns:
+                The next value that was received.
+            """
+            return self._receiver.consume()  # pylint: disable=protected-access
+
+    def __init__(self, client_id: str, service_id: str) -> None:
+        """Create a `Bidirectional` instance.
+
+        Args:
+            client_id: A name for the client, used to name the channels.
+            service_id: A name for the service end of the channels.
+        """
+        self._client_id = client_id
+        self._request_channel: Broadcast[T] = Broadcast(f"req_{service_id}_{client_id}")
+        self._response_channel: Broadcast[U] = Broadcast(
+            f"resp_{service_id}_{client_id}"
+        )
+
+        self._client_handle = Bidirectional.Handle(
+            self._request_channel.new_sender(),
+            self._response_channel.new_receiver(),
+        )
+        self._service_handle = Bidirectional.Handle(
+            self._response_channel.new_sender(),
+            self._request_channel.new_receiver(),
+        )
+
+    @property
+    def client_handle(self) -> Bidirectional.Handle[T, U]:
+        """Get a `Handle` for the client side to use.
+
+        Returns:
+            Object to send/receive messages with.
+        """
+        return self._client_handle
+
+    @property
+    def service_handle(self) -> Bidirectional.Handle[U, T]:
+        """Get a `Handle` for the service side to use.
+
+        Returns:
+            Object to send/receive messages with.
+        """
+        return self._service_handle

src/frequenz/channels/broadcast.py src/frequenz/channels/_broadcast.pysrc/frequenz/channels/broadcast.py renamed to src/frequenz/channels/_broadcast.py

@@ -12,10 +12,11 @@
 from typing import Deque, Dict, Generic, Optional
 from uuid import UUID, uuid4
 
-from frequenz.channels.base_classes import BufferedReceiver, ChannelClosedError
-from frequenz.channels.base_classes import Peekable as BasePeekable
-from frequenz.channels.base_classes import Sender as BaseSender
-from frequenz.channels.base_classes import T
+from ._base_classes import ChannelClosedError
+from ._base_classes import Peekable as BasePeekable
+from ._base_classes import Receiver as BaseReceiver
+from ._base_classes import Sender as BaseSender
+from ._base_classes import T
 
 logger = logging.Logger(__name__)
 
@@ -32,9 +33,9 @@ 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.util.Select],
+    [Merge][frequenz.channels.util.Merge] or
+    [MergeNamed][frequenz.channels.util.MergeNamed].
 
     Example:
         ``` python
@@ -192,7 +193,7 @@ async def send(self, msg: T) -> bool:
         return True
 
 
-class Receiver(BufferedReceiver[T]):
+class Receiver(BaseReceiver[T]):
     """A receiver to receive messages from the broadcast channel.
 
     Should not be created directly, but through the