Don't raise exceptions in Receiver.ready()

Instead, exceptions are raised in Receiver.consume(). This improves the
API for several reasons:

* There is no need to try: a whole select.ready() loop. Exceptions will
  be raised when something is actually consumed.
* It removes exception handling from tasks monitoring pending receivers.

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

Author: llucax

RELEASE_NOTES.md

@@ -18,6 +18,8 @@
 
 * `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
 
 * New exceptions were added:

src/frequenz/channels/_anycast.py

@@ -167,32 +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:
-            ReceiverStoppedError: if the receiver stopped producing messages.
-            ReceiverError: if there is some problem with the receiver.
+        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 ReceiverStoppedError(self) from ChannelClosedError(self._chan)
+                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

@@ -49,15 +49,16 @@ async def __anext__(self) -> T:
             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:
-            ReceiverStoppedError: if the receiver stopped producing messages.
-            ReceiverError: if there is some problem with the receiver.
+        Returns:
+            Whether the receiver is still active.
         """
 
     @abstractmethod
@@ -68,6 +69,10 @@ def consume(self) -> T:
 
         Returns:
             The next value received.
+
+        Raises:
+            ReceiverStoppedError: if the receiver stopped producing messages.
+            ReceiverError: if there is some problem with the receiver.
         """
 
     def __aiter__(self) -> Receiver[T]:
@@ -81,13 +86,13 @@ def __aiter__(self) -> Receiver[T]:
     async def receive(self) -> T:
         """Receive a message from the channel.
 
+        Returns:
+            The received message.
+
         Raises:
             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:
@@ -169,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

src/frequenz/channels/_bidirectional.py

@@ -68,15 +68,33 @@ async def send(self, msg: V) -> None:
                     err.__cause__ = this_chan_error
                 raise err
 
-        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()` (`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._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.
 
             Raises:
-                ReceiverStoppedError: if the receiver stopped producing messages.
+                ReceiverStoppedError: if there is some problem with the receiver.
                 ReceiverError: if there is some problem with the receiver.
+
+            # noqa: DAR401 err (https://github.com/terrencepreilly/darglint/issues/181)
             """
             try:
-                await self._receiver.ready()  # pylint: disable=protected-access
+                return self._receiver.consume()  # 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
@@ -91,14 +109,6 @@ async def ready(self) -> None:
                     err.__cause__ = this_chan_error
                 raise err
 
-        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.
 

src/frequenz/channels/_broadcast.py

@@ -257,29 +257,35 @@ def __len__(self) -> int:
         """
         return len(self._q)
 
-    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:
-            ReceiverStoppedError: if there is some problem with the receiver.
-            ReceiverInvalidatedError: if the receiver was converted into
-                a peekable.
+        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 there are still messages to consume from the queue, return immediately
+        if self._q:
+            return True
+
+        # if it is not longer active, return immediately
         if not self._active:
-            raise ReceiverInvalidatedError(
-                "This receiver was converted into a Peekable so it is not longer valid.",
-                self,
-            )
+            return False
 
         # 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 ReceiverStoppedError(self) from ChannelClosedError(self._chan)
+                return False
             async with self._chan.recv_cv:
                 await self._chan.recv_cv.wait()
+        return True
 
     def _deactivate(self) -> None:
         """Set the receiver as inactive and remove it from the channel."""
@@ -292,10 +298,23 @@ def consume(self) -> T:
 
         Returns:
             The next value that was received.
+
+        Raises:
+            ReceiverStoppedError: if there is some problem with the receiver.
+            ReceiverInvalidatedError: if the receiver was converted into
+                a peekable.
         """
+        if not self._q and not self._active:
+            raise ReceiverInvalidatedError(
+                "This receiver was converted into a Peekable so it is not longer valid.",
+                self,
+            )
+
+        if not self._q and self._chan.closed:
+            raise ReceiverStoppedError(self) from ChannelClosedError(self._chan)
+
         assert self._q, "calls to `consume()` must be follow a call to `ready()`"
-        ret = self._q.popleft()
-        return ret
+        return self._q.popleft()
 
     def into_peekable(self) -> Peekable[T]:
         """Convert the `Receiver` implementation into a `Peekable`.

src/frequenz/channels/util/_file_watcher.py

@@ -53,6 +53,7 @@ def __init__(
                 and pathlib.Path(path_str).is_file()
             ),
         )
+        self._awatch_stopped_exc: Optional[Exception] = None
         self._changes: Set[FileChange] = set()
 
     def __del__(self) -> None:
@@ -64,31 +65,44 @@ def __del__(self) -> None:
         """
         self._stop_event.set()
 
-    async def ready(self) -> None:
-        """Wait for the next file event and return its path.
+    async def ready(self) -> bool:
+        """Wait until the receiver is ready with a value or an error.
 
-        Returns:
-            Path of next file.
+        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:
-            ReceiverStoppedError: if the receiver stopped producing messages.
-            ReceiverError: if there is some problem with the receiver.
+        Returns:
+            Whether the receiver is still active.
         """
         # if there are messages waiting to be consumed, return immediately.
         if self._changes:
-            return
+            return True
+
+        # if it was already stopped, return immediately.
+        if self._awatch_stopped_exc is not None:
+            return False
 
         try:
             self._changes = await self._awatch.__anext__()
         except StopAsyncIteration as err:
-            raise ReceiverStoppedError(self) from err
+            self._awatch_stopped_exc = err
+
+        return True
 
     def consume(self) -> pathlib.Path:
         """Return the latest change once `ready` is complete.
 
         Returns:
             The next change that was received.
+
+        Raises:
+            ReceiverStoppedError: if there is some problem with the receiver.
         """
+        if not self._changes and self._awatch_stopped_exc is not None:
+            raise ReceiverStoppedError(self) from self._awatch_stopped_exc
+
         assert self._changes, "calls to `consume()` must be follow a call to `ready()`"
         change = self._changes.pop()
         # Tuple of (Change, path) returned by watchfiles

src/frequenz/channels/util/_merge.py

@@ -55,22 +55,28 @@ async def stop(self) -> None:
         await asyncio.gather(*self._pending, return_exceptions=True)
         self._pending = set()
 
-    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:
-            ReceiverStoppedError: if the receiver stopped producing messages.
-            ReceiverError: if there is some problem with the receiver.
+        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
+                return True
 
+            # if there are no more pending receivers, we return immediately.
             if len(self._pending) == 0:
-                raise ReceiverStoppedError(self)
+                return False
+
             done, self._pending = await asyncio.wait(
                 self._pending, return_when=asyncio.FIRST_COMPLETED
             )
@@ -91,7 +97,14 @@ def consume(self) -> T:
 
         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 not self._results and not self._pending:
+            raise ReceiverStoppedError(self)
+
         assert self._results, "calls to `consume()` must be follow a call to `ready()`"
 
         return self._results.popleft()