Expose strongly-typed Senders and Receivers from Broadcast

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>

Author: shsms

src/frequenz/channels/_broadcast.py

@@ -211,7 +211,7 @@ def __init__(self, *, name: str, resend_latest: bool = False) -> None:
         """The number of senders attached to this channel."""
 
         self._receivers: dict[
-            int, weakref.ReferenceType[_Receiver[ChannelMessageT]]
+            int, weakref.ReferenceType[BroadcastReceiver[ChannelMessageT]]
         ] = {}
         """The receivers attached to the channel, indexed by their hash()."""
 
@@ -272,13 +272,13 @@ async def close(self) -> None:  # noqa: D402
         """Close the channel, deprecated alias for `aclose()`."""  # noqa: D402
         return await self.aclose()
 
-    def new_sender(self) -> ClonableSubscribableSender[ChannelMessageT]:
+    def new_sender(self) -> BroadcastSender[ChannelMessageT]:
         """Return a new sender attached to this channel."""
-        return _Sender(self)
+        return BroadcastSender(self)
 
     def new_receiver(
         self, *, name: str | None = None, limit: int = 50, warn_on_overflow: bool = True
-    ) -> Receiver[ChannelMessageT]:
+    ) -> BroadcastReceiver[ChannelMessageT]:
         """Return a new receiver attached to this channel.
 
         Broadcast receivers have their own buffer, and when messages are not
@@ -294,7 +294,7 @@ def new_receiver(
         Returns:
             A new receiver attached to this channel.
         """
-        recv: _Receiver[ChannelMessageT] = _Receiver(
+        recv: BroadcastReceiver[ChannelMessageT] = BroadcastReceiver(
             self, name=name, limit=limit, warn_on_overflow=warn_on_overflow
         )
         self._receivers[hash(recv)] = weakref.ref(recv)
@@ -320,7 +320,7 @@ def __repr__(self) -> str:
 _T = TypeVar("_T")
 
 
-class _Sender(ClonableSubscribableSender[_T]):
+class BroadcastSender(ClonableSubscribableSender[_T]):
     """A sender to send messages to the broadcast channel.
 
     Should not be created directly, but through the
@@ -400,17 +400,17 @@ def __del__(self) -> None:
             self._channel._sender_count -= 1
 
     @override
-    def clone(self) -> _Sender[_T]:
+    def clone(self) -> BroadcastSender[_T]:
         """Return a clone of this sender."""
-        return _Sender(self._channel)
+        return BroadcastSender(self._channel)
 
     @override
     def subscribe(
         self,
         name: str | None = None,
         limit: int = 50,
         warn_on_overflow: bool = True,
-    ) -> Receiver[_T]:
+    ) -> BroadcastReceiver[_T]:
         """Return a new receiver attached to this sender's channel."""
         return self._channel.new_receiver(
             name=name, limit=limit, warn_on_overflow=warn_on_overflow
@@ -425,7 +425,7 @@ def __repr__(self) -> str:
         return f"{type(self).__name__}({self._channel!r})"
 
 
-class _Receiver(Receiver[_T]):
+class BroadcastReceiver(Receiver[_T]):
     """A receiver to receive messages from the broadcast channel.
 
     Should not be created directly, but through the

tests/test_broadcast.py

@@ -107,7 +107,7 @@ async def test_broadcast_after_close() -> None:
 async def test_broadcast_overflow() -> None:
     """Ensure messages sent to full broadcast receivers get dropped."""
     from frequenz.channels._broadcast import (  # pylint: disable=import-outside-toplevel
-        _Receiver,
+        BroadcastReceiver,
     )
 
     bcast: Broadcast[int] = Broadcast(name="meter_5")
@@ -117,9 +117,9 @@ async def test_broadcast_overflow() -> None:
     sender = bcast.new_sender()
 
     big_receiver = bcast.new_receiver(name="named-recv", limit=big_recv_size)
-    assert isinstance(big_receiver, _Receiver)
+    assert isinstance(big_receiver, BroadcastReceiver)
     small_receiver = bcast.new_receiver(limit=small_recv_size)
-    assert isinstance(small_receiver, _Receiver)
+    assert isinstance(small_receiver, BroadcastReceiver)
 
     async def drain_receivers() -> tuple[int, int]:
         big_sum = 0