Make receivers raise ReceiverError

The `Receiver.ready()` method (and related `receive()` and `__anext__`
when used as an async iterator) now `raise`s a `ReceiverError` and in
particular a `ReceiverStoppedError` when the receiver has no more
messages to receive.

`Receiver.consume()` doesn't raise any exceptions now (except for
`AssertionError` if there are programming errors).

Receivers raising `EOFError` now raise `ReceiverError` instead.

For channels which receivers stop receiving when the channel is closed,
the `ReceiverStoppedError` is chained with a `__cause__` that is a
`ChannelClosedError` with the channel that was closed.

Since now there is no need to use `ChannelClosedError` without
specifying a channel, now it requires the argument `channel` (before it
was optional).

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

Author: llucax

RELEASE_NOTES.md

@@ -8,6 +8,16 @@
 
 * The `Sender.send()` method now `raise`s a `SenderError` instead of returning `False`. The `SenderError` will typically have a `ChannelClosedError` and the underlying reason as a chained exception.
 
+* The `Receiver.ready()` method (and related `receive()` and `__anext__` when used as an async iterator) now `raise`s a `ReceiverError` and in particular a `ReceiverStoppedError` when the receiver has no more messages to receive.
+
+  `Receiver.consume()` doesn't raise any exceptions.
+
+  Receivers raising `EOFError` now raise `ReceiverError` instead.
+
+* For channels which senders raise an error when the channel is closed or which receivers stop receiving when the channel is closed, the `SenderError` and `ReceiverStoppedError` are chained with a `__cause__` that is a `ChannelClosedError` with the channel that was closed.
+
+* `ChannelClosedError` now requires the argument `channel` (before it was optional).
+
 ## New Features
 
 * New exceptions were added:
@@ -16,6 +26,10 @@
 
   * `SendError`: Raised for errors when sending messages.
 
+  * `ReceiverError`: Raised for errors when receiving messages.
+
+  * `ReceiverClosedError`: Raised when a receiver don't have more messages to receive.
+
 ## Bug Fixes
 
 <!-- Here goes notable bug fixes that are worth a special mention or explanation -->

src/frequenz/channels/__init__.py

@@ -49,6 +49,12 @@
 
 * [SenderError][frequenz.channels.SenderError]: Base class for all errors
   related to senders.
+
+* [ReceiverError][frequenz.channels.ReceiverError]: Base class for all errors
+  related to receivers.
+
+* [ReceiverStoppedError][frequenz.channels.ReceiverStoppedError]: A receiver
+  stopped producing messages.
 """
 
 from . import util
@@ -59,6 +65,8 @@
     Error,
     Peekable,
     Receiver,
+    ReceiverError,
+    ReceiverStoppedError,
     Sender,
     SenderError,
 )
@@ -74,6 +82,8 @@
     "Error",
     "Peekable",
     "Receiver",
+    "ReceiverError",
+    "ReceiverStoppedError",
     "Sender",
     "SenderError",
     "util",

src/frequenz/channels/_anycast.py

@@ -11,6 +11,7 @@
 
 from ._base_classes import ChannelClosedError
 from ._base_classes import Receiver as BaseReceiver
+from ._base_classes import ReceiverStoppedError
 from ._base_classes import Sender as BaseSender
 from ._base_classes import SenderError, T
 
@@ -171,15 +172,16 @@ async def ready(self) -> None:
         """Wait until the receiver is ready with a value.
 
         Raises:
-            ChannelClosedError: if the underlying channel is closed.
+            ReceiverStoppedError: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
         # if a message is already ready, then return immediately.
         if self._next is not None:
             return
 
         while len(self._chan.deque) == 0:
             if self._chan.closed:
-                raise ChannelClosedError()
+                raise ReceiverStoppedError(self) from ChannelClosedError(self._chan)
             async with self._chan.recv_cv:
                 await self._chan.recv_cv.wait()
         self._next = self._chan.deque.popleft()

src/frequenz/channels/_base_classes.py

@@ -33,7 +33,7 @@ class ChannelError(Error):
     All exceptions generated by channels inherit from this exception.
     """
 
-    def __init__(self, message: Any, channel: Any = None):
+    def __init__(self, message: Any, channel: Any):
         """Create a ChannelError instance.
 
         Args:
@@ -47,7 +47,7 @@ def __init__(self, message: Any, channel: Any = None):
 class ChannelClosedError(ChannelError):
     """Error raised when trying to operate on a closed channel."""
 
-    def __init__(self, channel: Any = None):
+    def __init__(self, channel: Any):
         """Create a `ChannelClosedError` instance.
 
         Args:
@@ -74,6 +74,37 @@ def __init__(self, message: Any, sender: Sender[T]):
         self.sender: Sender[T] = sender
 
 
+class ReceiverError(Error, Generic[T]):
+    """An error produced in a [Receiver][frequenz.channels.Receiver].
+
+    All exceptions generated by receivers inherit from this exception.
+    """
+
+    def __init__(self, message: Any, receiver: Receiver[T]):
+        """Create an instance.
+
+        Args:
+            message: An error message.
+            receiver: The [Receiver][frequenz.channels.Receiver] where the
+                error happened.
+        """
+        super().__init__(message)
+        self.receiver: Receiver[T] = receiver
+
+
+class ReceiverStoppedError(ReceiverError[T]):
+    """The [Receiver][frequenz.channels.Receiver] stopped producing messages."""
+
+    def __init__(self, receiver: Receiver[T]):
+        """Create an instance.
+
+        Args:
+            receiver: The [Receiver][frequenz.channels.Receiver] where the
+                error happened.
+        """
+        super().__init__(f"Receiver {receiver} was stopped", receiver)
+
+
 class Sender(ABC, Generic[T]):
     """A channel Sender."""
 
@@ -99,12 +130,13 @@ async def __anext__(self) -> T:
             The next value received.
 
         Raises:
-            StopAsyncIteration: if the underlying channel is closed.
+            StopAsyncIteration: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
         try:
             await self.ready()
             return self.consume()
-        except ChannelClosedError as exc:
+        except ReceiverStoppedError as exc:
             raise StopAsyncIteration() from exc
 
     @abstractmethod
@@ -115,7 +147,8 @@ async def ready(self) -> None:
         `consume()`.
 
         Raises:
-            ChannelClosedError: if the underlying channel is closed.
+            ReceiverStoppedError: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
 
     @abstractmethod
@@ -126,9 +159,6 @@ def consume(self) -> T:
 
         Returns:
             The next value received.
-
-        Raises:
-            ChannelClosedError: if the underlying channel is closed.
         """
 
     def __aiter__(self) -> Receiver[T]:
@@ -143,15 +173,29 @@ async def receive(self) -> T:
         """Receive a message from the channel.
 
         Raises:
-            ChannelClosedError: if the underlying channel is closed.
+            ReceiverStoppedError: if there is some problem with the receiver.
+            ReceiverError: if there is some problem with the receiver.
 
         Returns:
             The received message.
+
+        # noqa: DAR401 __cause__ (https://github.com/terrencepreilly/darglint/issues/181)
         """
         try:
             received = await self.__anext__()  # pylint: disable=unnecessary-dunder-call
         except StopAsyncIteration as exc:
-            raise ChannelClosedError() from exc
+            # If we already had a cause and it was the receiver was stopped,
+            # then reuse that error, as StopAsyncIteration is just an artifact
+            # introduced by __anext__.
+            if (
+                isinstance(exc.__cause__, ReceiverStoppedError)
+                # pylint is not smart enough to figure out we checked above
+                # this is a ReceiverStoppedError and thus it does have
+                # a receiver member
+                and exc.__cause__.receiver is self  # pylint: disable=no-member
+            ):
+                raise exc.__cause__
+            raise ReceiverStoppedError(self) from exc
         return received
 
     def map(self, call: Callable[[T], U]) -> Receiver[U]:

src/frequenz/channels/_bidirectional.py

@@ -7,7 +7,15 @@
 
 from typing import Generic, TypeVar
 
-from ._base_classes import ChannelError, Receiver, Sender, SenderError, T, U
+from ._base_classes import (
+    ChannelError,
+    Receiver,
+    ReceiverError,
+    Sender,
+    SenderError,
+    T,
+    U,
+)
 from ._broadcast import Broadcast
 
 V = TypeVar("V")
@@ -68,8 +76,27 @@ async def send(self, msg: V) -> None:
                 raise err
 
         async def ready(self) -> None:
-            """Wait until the receiver is ready with a value."""
-            await self._receiver.ready()  # pylint: disable=protected-access
+            """Wait until the receiver is ready with a value.
+
+            Raises:
+                ReceiverStoppedError: if the receiver stopped producing messages.
+                ReceiverError: if there is some problem with the receiver.
+            """
+            try:
+                await self._receiver.ready()  # pylint: disable=protected-access
+            except ReceiverError as err:
+                # If this comes from a channel error, then we inject another
+                # ChannelError having the information about the Bidirectional
+                # channel to hide (at least partially) the underlaying
+                # Broadcast channels we use.
+                if isinstance(err.__cause__, ChannelError):
+                    this_chan_error = ChannelError(
+                        f"Error in the underlying channel {err.__cause__.channel}: {err.__cause__}",
+                        self._chan,  # pylint: disable=protected-access
+                    )
+                    this_chan_error.__cause__ = err.__cause__
+                    err.__cause__ = this_chan_error
+                raise err
 
         def consume(self) -> W:
             """Return the latest value once `_ready` is complete.

src/frequenz/channels/_broadcast.py

@@ -15,6 +15,7 @@
 from ._base_classes import ChannelClosedError
 from ._base_classes import Peekable as BasePeekable
 from ._base_classes import Receiver as BaseReceiver
+from ._base_classes import ReceiverError, ReceiverStoppedError
 from ._base_classes import Sender as BaseSender
 from ._base_classes import SenderError, T
 
@@ -256,19 +257,19 @@ async def ready(self) -> None:
         """Wait until the receiver is ready with a value.
 
         Raises:
-            EOFError: if this receiver is no longer active.
-            ChannelClosedError: if the underlying channel is closed.
+            ReceiverStoppedError: if there is some problem with the receiver.
+            ReceiverError: if the receiver is not longer active.
         """
         if not self._active:
-            raise EOFError("This receiver is no longer active.")
+            raise ReceiverError("This receiver is no longer active.", self)
 
         # Use a while loop here, to handle spurious wakeups of condition variables.
         #
         # The condition also makes sure that if there are already messages ready to be
         # consumed, then we return immediately.
         while len(self._q) == 0:
             if self._chan.closed:
-                raise ChannelClosedError()
+                raise ReceiverStoppedError(self) from ChannelClosedError(self._chan)
             async with self._chan.recv_cv:
                 await self._chan.recv_cv.wait()
 

src/frequenz/channels/util/_file_watcher.py

@@ -10,7 +10,7 @@
 from watchfiles import Change, awatch
 from watchfiles.main import FileChange
 
-from .._base_classes import Receiver
+from .._base_classes import Receiver, ReceiverStoppedError
 
 
 class FileWatcher(Receiver[pathlib.Path]):
@@ -66,24 +66,25 @@ def __del__(self) -> None:
     async def ready(self) -> None:
         """Wait for the next file event and return its path.
 
-        Raises:
-            StopAsyncIteration: When the channel is closed.
-
         Returns:
             Path of next file.
+
+        Raises:
+            ReceiverStoppedError: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
         # if there are messages waiting to be consumed, return immediately.
         if self._changes:
             return
 
-        self._changes = await self._awatch.__anext__()
+        try:
+            self._changes = await self._awatch.__anext__()
+        except StopAsyncIteration as err:
+            raise ReceiverStoppedError(self) from err
 
     def consume(self) -> pathlib.Path:
         """Return the latest change once `ready` is complete.
 
-        Raises:
-            ChannelClosedError: When the channel is closed.
-
         Returns:
             The next change that was received.
         """

src/frequenz/channels/util/_merge.py

@@ -7,7 +7,7 @@
 from collections import deque
 from typing import Any, Deque, Set
 
-from .._base_classes import ChannelClosedError, Receiver, T
+from .._base_classes import Receiver, ReceiverStoppedError, T
 
 
 class Merge(Receiver[T]):
@@ -58,7 +58,8 @@ async def ready(self) -> None:
         """Wait until the receiver is ready with a value.
 
         Raises:
-            ChannelClosedError: if the underlying channel is closed.
+            ReceiverStoppedError: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
         # we use a while loop to continue to wait for new data, in case the
         # previous `wait` completed because a channel was closed.
@@ -68,7 +69,7 @@ async def ready(self) -> None:
                 return
 
             if len(self._pending) == 0:
-                raise ChannelClosedError()
+                raise ReceiverStoppedError(self)
             done, self._pending = await asyncio.wait(
                 self._pending, return_when=asyncio.FIRST_COMPLETED
             )
@@ -87,9 +88,6 @@ async def ready(self) -> None:
     def consume(self) -> T:
         """Return the latest value once `ready` is complete.
 
-        Raises:
-            EOFError: When called before a call to `ready()` finishes.
-
         Returns:
             The next value that was received.
         """