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, SenderClosedError, SenderError
+from ._sender import ClonableSubscribableSender, SenderClosedError, 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
@@ -381,6 +381,23 @@ async def aclose(self) -> None:
         """
         self._closed = True
 
+    @override
+    def clone(self) -> _Sender[_T]:
+        """Return a clone of this sender."""
+        return _Sender(self._channel)
+
+    @override
+    def subscribe(
+        self,
+        name: str | None = None,
+        limit: int = 50,
+        warn_on_overflow: bool = True,
+    ) -> Receiver[_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
+        )
+
     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]):
@@ -109,3 +112,35 @@ def __init__(self, sender: Sender[SenderMessageT_co]):
             sender: The [Sender][frequenz.channels.Sender] that was closed.
         """
         super().__init__("Sender is closed", sender)
+
+
+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."""