Make broadcast sender clonable and subscribable

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

Author: shsms

src/frequenz/channels/_broadcast.py

@@ -16,7 +16,7 @@
 from ._exceptions import ChannelClosedError
 from ._generic import ChannelMessageT
 from ._receiver import Receiver, ReceiverStoppedError
-from ._sender import Sender, SenderError
+from ._sender import ClonableSubscribableSender, SenderError
 
 _logger = logging.getLogger(__name__)
 
@@ -269,7 +269,7 @@ async def close(self) -> None:  # noqa: D402
         """Close the channel, deprecated alias for `aclose()`."""  # noqa: D402
         return await self.aclose()
 
-    def new_sender(self) -> Sender[ChannelMessageT]:
+    def new_sender(self) -> ClonableSubscribableSender[ChannelMessageT]:
         """Return a new sender attached to this channel."""
         return _Sender(self)
 
@@ -317,7 +317,7 @@ def __repr__(self) -> str:
 _T = TypeVar("_T")
 
 
-class _Sender(Sender[_T]):
+class _Sender(ClonableSubscribableSender[_T]):
     """A sender to send messages to the broadcast channel.
 
     Should not be created directly, but through the
@@ -365,6 +365,16 @@ async def send(self, message: _T, /) -> None:
             self._channel._recv_cv.notify_all()
         # pylint: enable=protected-access
 
+    @override
+    def clone(self) -> _Sender[_T]:
+        """Return a clone of this sender."""
+        return _Sender(self._channel)
+
+    @override
+    def subscribe(self) -> Receiver[_T]:
+        """Return a new receiver attached to this sender's channel."""
+        return self._channel.new_receiver()
+
     def __str__(self) -> str:
         """Return a string representation of this sender."""
         return f"{self._channel}:{type(self).__name__}"

src/frequenz/channels/_sender.py

@@ -49,11 +49,14 @@
 ```
 """
 
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
 from typing import Generic
 
 from ._exceptions import Error
 from ._generic import SenderMessageT_co, SenderMessageT_contra
+from ._receiver import Receiver
 
 
 class Sender(ABC, Generic[SenderMessageT_contra]):
@@ -88,3 +91,35 @@ def __init__(self, message: str, sender: Sender[SenderMessageT_co]):
         super().__init__(message)
         self.sender: Sender[SenderMessageT_co] = sender
         """The sender where the error happened."""
+
+
+class SubscribableSender(Sender[SenderMessageT_contra], ABC):
+    """A [Sender][frequenz.channels.Sender] that can be subscribed to."""
+
+    @abstractmethod
+    def subscribe(self) -> Receiver[SenderMessageT_contra]:
+        """Subscribe to this sender.
+
+        Returns:
+            A new sender that sends messages to the same channel as this sender.
+        """
+
+
+class ClonableSender(Sender[SenderMessageT_contra], ABC):
+    """A [Sender][frequenz.channels.Sender] that can be cloned."""
+
+    @abstractmethod
+    def clone(self) -> ClonableSender[SenderMessageT_contra]:
+        """Clone this sender.
+
+        Returns:
+            A new sender that sends messages to the same channel as this sender.
+        """
+
+
+class ClonableSubscribableSender(
+    SubscribableSender[SenderMessageT_contra],
+    ClonableSender[SenderMessageT_contra],
+    ABC,
+):
+    """A [Sender][frequenz.channels.Sender] that can be both cloned and subscribed to."""