Make send() raise a SenderError

Instead of returning a `bool`, which can be easily overlooked, it now
raises a new `SenerError`.

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

Author: llucax

RELEASE_NOTES.md

@@ -6,12 +6,15 @@
 
 ## 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.
 
 ## New Features
 
-* A new base exception `frequenz.channels.Error` was added. All exceptions from
-  this library inherit from this exception.
+* New exceptions were added:
+
+  * `Error`: A base exception from which all exceptions from this library inherit.
+
+  * `SendError`: Raised for errors when sending messages.
 
 ## Bug Fixes
 

src/frequenz/channels/__init__.py

@@ -46,6 +46,9 @@
 
 * [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.
 """
 
 from . import util
@@ -57,6 +60,7 @@
     Peekable,
     Receiver,
     Sender,
+    SenderError,
 )
 from ._bidirectional import Bidirectional
 from ._broadcast import Broadcast
@@ -71,5 +75,6 @@
     "Peekable",
     "Receiver",
     "Sender",
+    "SenderError",
     "util",
 ]

src/frequenz/channels/_anycast.py

@@ -12,7 +12,7 @@
 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 ._base_classes import SenderError, T
 
 
 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]):

src/frequenz/channels/_base_classes.py

@@ -28,7 +28,7 @@ def __init__(self, message: Any):
 
 
 class ChannelError(Error):
-    """Base channel error.
+    """An error produced in a channel.
 
     All exceptions generated by channels inherit from this exception.
     """
@@ -56,19 +56,36 @@ def __init__(self, channel: Any = None):
         super().__init__(f"Channel {channel} was closed", channel)
 
 
+class SenderError(Error, Generic[T]):
+    """An error produced in a [Sender][frequenz.channels.Sender].
+
+    All exceptions generated by senders inherit from this exception.
+    """
+
+    def __init__(self, message: Any, sender: Sender[T]):
+        """Create an instance.
+
+        Args:
+            message: An error message.
+            sender: The [Sender][frequenz.channels.Sender] where the error
+                happened.
+        """
+        super().__init__(message)
+        self.sender: Sender[T] = sender
+
+
 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.
         """
 
 

src/frequenz/channels/_bidirectional.py

@@ -7,7 +7,7 @@
 
 from typing import Generic, TypeVar
 
-from ._base_classes import Receiver, Sender, T, U
+from ._base_classes import ChannelError, Receiver, Sender, SenderError, T, U
 from ._broadcast import Broadcast
 
 V = TypeVar("V")
@@ -23,26 +23,49 @@ class Handle(Sender[V], Receiver[W]):
         It can be used to send/receive values between the client and service.
         """
 
-        def __init__(self, sender: Sender[V], receiver: Receiver[W]) -> None:
+        def __init__(
+            self,
+            channel: Bidirectional[V, W] | Bidirectional[W, V],
+            sender: Sender[V],
+            receiver: Receiver[W],
+        ) -> None:
             """Create a `Bidirectional.Handle` instance.
 
             Args:
+                channel: The underlying channel.
                 sender: A sender to send values with.
                 receiver: A receiver to receive values from.
             """
+            self._chan = channel
             self._sender = sender
             self._receiver = receiver
 
-        async def send(self, msg: V) -> bool:
+        async def send(self, msg: V) -> None:
             """Send a value to the other side.
 
             Args:
                 msg: The value to send.
 
-            Returns:
-                Whether the send was successful or not.
+            Raises:
+                SenderError: if the underlying channel was closed.
+                    A [ChannelClosedError][frequenz.channels.ChannelClosedError]
+                    is set as the cause.
             """
-            return await self._sender.send(msg)
+            try:
+                await self._sender.send(msg)
+            except SenderError 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
 
         async def ready(self) -> None:
             """Wait until the receiver is ready with a value."""
@@ -70,10 +93,12 @@ def __init__(self, client_id: str, service_id: str) -> None:
         )
 
         self._client_handle = Bidirectional.Handle(
+            self,
             self._request_channel.new_sender(),
             self._response_channel.new_receiver(),
         )
         self._service_handle = Bidirectional.Handle(
+            self,
             self._response_channel.new_sender(),
             self._request_channel.new_receiver(),
         )

src/frequenz/channels/_broadcast.py

@@ -16,7 +16,7 @@
 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
+from ._base_classes import SenderError, T
 
 logger = logging.Logger(__name__)
 
@@ -165,18 +165,21 @@ def __init__(self, chan: Broadcast[T]) -> None:
         """
         self._chan = chan
 
-    async def send(self, msg: T) -> bool:
+    async def send(self, msg: T) -> None:
         """Send a message to all broadcast receivers.
 
         Args:
             msg: The message to be broadcast.
 
-        Returns:
-            Whether the message was sent, based on whether the broadcast
-                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
+            )
         # pylint: disable=protected-access
         self._chan._latest = msg
         stale_refs = []
@@ -190,7 +193,6 @@ async def send(self, msg: T) -> bool:
             del self._chan.receivers[name]
         async with self._chan.recv_cv:
             self._chan.recv_cv.notify_all()
-        return True
 
 
 class Receiver(BaseReceiver[T]):

tests/test_anycast.py

@@ -7,7 +7,7 @@
 
 import pytest
 
-from frequenz.channels import Anycast, ChannelClosedError, Receiver, Sender
+from frequenz.channels import Anycast, ChannelClosedError, Receiver, Sender, SenderError
 
 
 async def test_anycast() -> None:
@@ -58,7 +58,8 @@ async def update_tracker_on_receive(receiver_id: int, chan: Receiver[int]) -> No
     await acast.close()
     await receivers_runs
 
-    assert await after_close_sender.send(5) is False
+    with pytest.raises(SenderError):
+        await after_close_sender.send(5)
     with pytest.raises(ChannelClosedError):
         await after_close_receiver.receive()
 
@@ -77,11 +78,13 @@ async def test_anycast_after_close() -> None:
     receiver = acast.new_receiver()
     sender = acast.new_sender()
 
-    assert await sender.send(2) is True
+    await sender.send(2)
 
     await acast.close()
 
-    assert await sender.send(5) is False
+    with pytest.raises(SenderError):
+        await sender.send(5)
+
     assert await receiver.receive() == 2
     with pytest.raises(ChannelClosedError):
         await receiver.receive()

tests/test_bidirectional.py

@@ -5,7 +5,14 @@
 
 import asyncio
 
-from frequenz.channels import Bidirectional
+import pytest
+
+from frequenz.channels import (
+    Bidirectional,
+    ChannelClosedError,
+    ChannelError,
+    SenderError,
+)
 
 
 async def test_request_response() -> None:
@@ -41,3 +48,20 @@ async def service(handle: Bidirectional.Handle[str, int]) -> None:
 
     await client_handle.send(42)  # Stop the service task
     await service_task
+
+
+async def test_sender_error_chaining() -> None:
+    """Ensure bi-directional communication is possible."""
+
+    req_resp: Bidirectional[int, str] = Bidirectional("test_client", "test_service")
+
+    await req_resp._response_channel.close()  # pylint: disable=protected-access
+
+    with pytest.raises(SenderError, match="The channel was closed") as exc_info:
+        await req_resp.service_handle.send("I'm closed!")
+
+    err = exc_info.value
+    cause = err.__cause__
+    assert isinstance(cause, ChannelError)
+    assert cause.args[0].startswith("Error in the underlying channel")
+    assert isinstance(cause.__cause__, ChannelClosedError)

tests/test_broadcast.py

@@ -8,7 +8,13 @@
 
 import pytest
 
-from frequenz.channels import Broadcast, ChannelClosedError, Receiver, Sender
+from frequenz.channels import (
+    Broadcast,
+    ChannelClosedError,
+    Receiver,
+    Sender,
+    SenderError,
+)
 
 
 async def test_broadcast() -> None:
@@ -68,7 +74,8 @@ async def test_broadcast_after_close() -> None:
 
     await bcast.close()
 
-    assert await sender.send(5) is False
+    with pytest.raises(SenderError):
+        await sender.send(5)
     with pytest.raises(ChannelClosedError):
         await receiver.receive()