Improve and add exceptions (#61)

- Properly await for service task
- Make output more verbose for darglint
- Add base exception
- Make `send()` raise a `SenderError`
- Make receivers raise `ReceiverError`
- Move exceptions into its own module
- Add a `ReceiverInvalidatedError`
- Don't raise exceptions in Receiver.ready()

Fixes #52.

Author: llucax

RELEASE_NOTES.md

@@ -6,11 +6,33 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including if there are any depractions and what they should be replaced with -->
+* 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 `ReceiverInvalidatedError` 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).
+
+* Now exceptions are not raised in Receiver.ready() but in Receiver.consume() (receive() or the async iterator `anext`).
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* New exceptions were added:
+
+  * `Error`: A base exception from which all exceptions from this library inherit.
+
+  * `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.
+
+  * `ReceiverInvalidatedError`: Raised when a receiver was invalidated (for example it was converted into a `Peekable`).
 
 ## Bug Fixes
 

noxfile.py

@@ -66,7 +66,7 @@ def docstrings(session: nox.Session) -> None:
     # Darglint checks that function argument and return values are documented.
     # This is needed only for the `src` dir, so we exclude the other top level
     # dirs that contain code.
-    session.run("darglint", "src")
+    session.run("darglint", "-v2", "src")
 
 
 @nox.session

src/frequenz/channels/__init__.py

@@ -38,27 +38,57 @@
 
 Exception classes:
 
+* [Error][frequenz.channels.Error]: Base class for all errors in this
+  library.
+
 * [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.
+
+* [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.
+
+* [ReceiverInvalidatedError][frequenz.channels.ReceiverInvalidatedError]:
+  A receiver is not longer valid (for example if it was converted into
+  a peekable.
 """
 
 from . import util
 from ._anycast import Anycast
-from ._base_classes import ChannelClosedError, ChannelError, Peekable, Receiver, Sender
+from ._base_classes import Peekable, Receiver, Sender
 from ._bidirectional import Bidirectional
 from ._broadcast import Broadcast
+from ._exceptions import (
+    ChannelClosedError,
+    ChannelError,
+    Error,
+    ReceiverError,
+    ReceiverInvalidatedError,
+    ReceiverStoppedError,
+    SenderError,
+)
 
 __all__ = [
     "Anycast",
     "Bidirectional",
     "Broadcast",
     "ChannelClosedError",
     "ChannelError",
+    "Error",
     "Peekable",
     "Receiver",
+    "ReceiverError",
+    "ReceiverInvalidatedError",
+    "ReceiverStoppedError",
     "Sender",
+    "SenderError",
     "util",
 ]

src/frequenz/channels/_anycast.py

@@ -9,10 +9,10 @@
 from collections import deque
 from typing import Deque, Generic, Optional
 
-from ._base_classes import ChannelClosedError
 from ._base_classes import Receiver as BaseReceiver
 from ._base_classes import Sender as BaseSender
 from ._base_classes import T
+from ._exceptions import ChannelClosedError, ReceiverStoppedError, SenderError
 
 
 class Anycast(Generic[T]):
@@ -123,7 +123,7 @@ def __init__(self, chan: Anycast[T]) -> None:
         """
         self._chan = chan
 
-    async def send(self, msg: T) -> bool:
+    async def send(self, msg: T) -> None:
         """Send a message across the channel.
 
         To send, this method inserts the message into the Anycast channel's
@@ -134,19 +134,21 @@ async def send(self, msg: T) -> bool:
         Args:
             msg: The message to be sent.
 
-        Returns:
-            Whether the message was sent, based on whether the channel is open
-                or not.
+        Raises:
+            SenderError: if the underlying channel was closed.
+                A [ChannelClosedError][frequenz.channels.ChannelClosedError] is
+                set as the cause.
         """
         if self._chan.closed:
-            return False
+            raise SenderError("The channel was closed", self) from ChannelClosedError(
+                self._chan
+            )
         while len(self._chan.deque) == self._chan.deque.maxlen:
             async with self._chan.send_cv:
                 await self._chan.send_cv.wait()
         self._chan.deque.append(msg)
         async with self._chan.recv_cv:
             self._chan.recv_cv.notify(1)
-        return True
 
 
 class Receiver(BaseReceiver[T]):
@@ -165,31 +167,44 @@ def __init__(self, chan: Anycast[T]) -> None:
         self._chan = chan
         self._next: Optional[T] = None
 
-    async def ready(self) -> None:
-        """Wait until the receiver is ready with a value.
+    async def ready(self) -> bool:
+        """Wait until the receiver is ready with a value or an error.
 
-        Raises:
-            ChannelClosedError: if the underlying channel is closed.
+        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.
         """
         # if a message is already ready, then return immediately.
         if self._next is not None:
-            return
+            return True
 
         while len(self._chan.deque) == 0:
             if self._chan.closed:
-                raise ChannelClosedError()
+                return False
             async with self._chan.recv_cv:
                 await self._chan.recv_cv.wait()
         self._next = self._chan.deque.popleft()
         async with self._chan.send_cv:
             self._chan.send_cv.notify(1)
+        return True
 
     def consume(self) -> T:
         """Return the latest value once `ready()` is complete.
 
         Returns:
             The next value that was received.
+
+        Raises:
+            ReceiverStoppedError: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
+        if self._next is None and self._chan.closed:
+            raise ReceiverStoppedError(self) from ChannelClosedError(self._chan)
+
         assert (
             self._next is not None
         ), "calls to `consume()` must be follow a call to `ready()`"

src/frequenz/channels/_base_classes.py

@@ -6,54 +6,26 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import Any, Callable, Generic, Optional, TypeVar
+from typing import Callable, Generic, Optional, TypeVar
+
+from ._exceptions import ReceiverStoppedError
 
 T = TypeVar("T")
 U = TypeVar("U")
 
 
-class ChannelError(RuntimeError):
-    """Base channel error.
-
-    All exceptions generated by channels inherit from this exception.
-    """
-
-    def __init__(self, message: Any, channel: Any = None):
-        """Create a ChannelError instance.
-
-        Args:
-            message: An error message.
-            channel: A reference to the channel that encountered the error.
-        """
-        super().__init__(message)
-        self.channel: Any = channel
-
-
-class ChannelClosedError(ChannelError):
-    """Error raised when trying to operate on a closed channel."""
-
-    def __init__(self, channel: Any = None):
-        """Create a `ChannelClosedError` instance.
-
-        Args:
-            channel: A reference to the channel that was closed.
-        """
-        super().__init__(f"Channel {channel} was closed", channel)
-
-
 class Sender(ABC, Generic[T]):
     """A channel Sender."""
 
     @abstractmethod
-    async def send(self, msg: T) -> bool:
+    async def send(self, msg: T) -> None:
         """Send a message to the channel.
 
         Args:
             msg: The message to be sent.
 
-        Returns:
-            Whether the message was sent, based on whether the channel is open
-                or not.
+        Raises:
+            SenderError: if there was an error sending the message.
         """
 
 
@@ -67,23 +39,26 @@ 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
-    async def ready(self) -> None:
-        """Wait until the receiver is ready with a value.
+    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()`.
+        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.
 
-        Raises:
-            ChannelClosedError: if the underlying channel is closed.
+        Returns:
+            Whether the receiver is still active.
         """
 
     @abstractmethod
@@ -96,7 +71,8 @@ def consume(self) -> T:
             The next value received.
 
         Raises:
-            ChannelClosedError: if the underlying channel is closed.
+            ReceiverStoppedError: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
 
     def __aiter__(self) -> Receiver[T]:
@@ -110,16 +86,30 @@ def __aiter__(self) -> Receiver[T]:
     async def receive(self) -> T:
         """Receive a message from the channel.
 
-        Raises:
-            ChannelClosedError: if the underlying channel is closed.
-
         Returns:
             The received message.
+
+        Raises:
+            ReceiverStoppedError: if there is some problem with the receiver.
+            ReceiverError: if there is some problem with the receiver.
+
+        # 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]:
@@ -184,14 +174,26 @@ def __init__(self, recv: Receiver[T], transform: Callable[[T], U]) -> None:
         self._recv = recv
         self._transform = transform
 
-    async def ready(self) -> None:
-        """Wait until the receiver is ready with a value."""
-        await self._recv.ready()  # pylint: disable=protected-access
+    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.
+        """
+        return await self._recv.ready()  # pylint: disable=protected-access
 
     def consume(self) -> U:
         """Return a transformed value once `ready()` is complete.
 
         Returns:
             The next value that was received.
+
+        Raises:
+            ChannelClosedError: if the underlying channel is closed.
         """
         return self._transform(self._recv.consume())  # pylint: disable=protected-access