From 96d9fd0e8680bbef1d823a9bfb56b735b4d850c8 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:04:20 +0000
Subject: [PATCH 01/15] Wake waiting broadcast receivers when they are closed

`BroadcastReceiver.close()` is synchronous and couldn't notify the
channel's condition variable, so a task waiting in `ready()` hung until
the next message arrived or the channel was closed.  Receivers now wait
on an event that can be set from synchronous code.

Waiters left on a closed event loop are skipped, so they don't make
`close()` fail.  For the same reason, `send()` and `aclose()` no longer
raise `RuntimeError: Event loop is closed` because of them.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 48 ++++++++++++-------
 tests/test_broadcast.py             | 72 +++++++++++++++++++++++++++++
 2 files changed, 104 insertions(+), 16 deletions(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 0f010ea0..80e10bf0 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -7,7 +7,7 @@
 
 import logging
 import weakref
-from asyncio import Condition
+from asyncio import Event
 from collections import deque
 from typing import Generic, TypeVar
 
@@ -216,8 +216,14 @@ def __init__(
         Only used for debugging purposes.
         """
 
-        self._recv_cv: Condition = Condition()
-        """The condition to wait for data in the channel's buffer."""
+        self._wakeup: Event = Event()
+        """The event receivers wait on for new messages or state changes.
+
+        Setting it wakes up all receivers waiting on it, and it is then replaced
+        with a fresh event for the next wait.  It can be set from synchronous code
+        running on the receivers' event loop, like
+        [`BroadcastReceiver.close()`][frequenz.channels.BroadcastReceiver.close].
+        """
 
         self._sender_count: int = 0
         """The number of senders attached to this channel."""
@@ -280,8 +286,21 @@ async def aclose(self) -> None:
         """
         self._latest = None
         self._closed = True
-        async with self._recv_cv:
-            self._recv_cv.notify_all()
+        self._wake_receivers()
+
+    def _wake_receivers(self) -> None:
+        """Wake up all receivers waiting for new messages."""
+        wakeup, self._wakeup = self._wakeup, Event()
+        try:
+            wakeup.set()
+        except RuntimeError:
+            # `set()` raises if the waiters' loop is closed, and they can't be woken
+            # anyway.  All waiters of an event share one loop, so any other
+            # `RuntimeError`, like a call from another thread in debug mode, is
+            # re-raised.
+            loop = getattr(wakeup, "_loop", None)
+            if loop is None or not loop.is_closed():
+                raise
 
     @deprecated("The close() method is deprecated, use aclose() instead")
     async def close(self) -> None:  # noqa: D402
@@ -398,8 +417,7 @@ async def send(self, message: _T, /) -> None:
                 self._channel
             )
         self._channel._latest = message
-        async with self._channel._recv_cv:
-            self._channel._recv_cv.notify_all()
+        self._channel._wake_receivers()
         # pylint: enable=protected-access
 
     @override
@@ -562,10 +580,8 @@ async def ready(self) -> bool:
         if self._q:
             return True
 
-        # 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.
+        # Use a while loop here, because receivers are woken up on every channel
+        # state change, not only when a message for this receiver arrives.
         # pylint: disable=protected-access
         while len(self._q) == 0:
             if self._channel._closed or self._closed:
@@ -573,8 +589,7 @@ async def ready(self) -> bool:
             if self._channel._auto_close_enabled and self._channel._sender_count == 0:
                 await self._channel.aclose()
                 return False
-            async with self._channel._recv_cv:
-                await self._channel._recv_cv.wait()
+            await self._channel._wakeup.wait()
         return True
         # pylint: enable=protected-access
 
@@ -606,9 +621,10 @@ def close(self) -> None:
         [`ReceiverStoppedError`][...ReceiverStoppedError].
         """
         self._closed = True
-        self._channel._receivers.pop(  # pylint: disable=protected-access
-            hash(self), None
-        )
+        # pylint: disable=protected-access
+        self._channel._receivers.pop(hash(self), None)
+        self._channel._wake_receivers()
+        # pylint: enable=protected-access
 
     def __str__(self) -> str:
         """Return a string representation of this receiver."""
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index 7035250b..b0c53c42 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -427,6 +427,78 @@ async def test_broadcast_close_receiver() -> None:
         _ = await receiver_2.receive()
 
 
+async def test_broadcast_close_receiver_while_waiting() -> None:
+    """Ensure closing a receiver wakes up a task waiting on it."""
+    chan = Broadcast[int](name="input-chan")
+    receiver = chan.new_receiver()
+
+    waiter = asyncio.create_task(receiver.ready())
+    await asyncio.sleep(0)
+    assert not waiter.done()
+
+    receiver.close()
+
+    assert await asyncio.wait_for(waiter, timeout=1.0) is False
+
+
+def _wait_on_closed_loop(receiver: Receiver[int]) -> None:
+    """Leave a task waiting on the receiver in an event loop that is then closed."""
+
+    async def wait() -> None:
+        asyncio.create_task(receiver.ready())
+        await asyncio.sleep(0)
+
+    # The loop is closed without cancelling the waiting task.
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(wait())
+    finally:
+        loop.close()
+
+
+def test_broadcast_wake_after_loop_closed() -> None:
+    """Ensure waking receivers works after a waiting receiver's loop closed."""
+    chan = Broadcast[int](name="input-chan")
+    sender = chan.new_sender()
+    receiver = chan.new_receiver()
+    receiver_2 = chan.new_receiver()
+
+    _wait_on_closed_loop(receiver)
+    receiver_2.close()
+    with pytest.raises(ReceiverStoppedError):
+        receiver_2.consume()
+
+    _wait_on_closed_loop(receiver)
+    asyncio.run(sender.send(1))
+    assert receiver.consume() == 1
+
+    _wait_on_closed_loop(receiver)
+    asyncio.run(chan.aclose())
+    assert chan.is_closed
+
+
+def test_broadcast_wake_from_other_thread_in_debug_mode() -> None:
+    """Ensure a wake-up from another thread still fails in debug mode."""
+
+    async def close_in_thread() -> None:
+        chan = Broadcast[int](name="input-chan")
+        receiver = chan.new_receiver()
+        receiver_2 = chan.new_receiver()
+        asyncio.create_task(receiver.ready())
+        await asyncio.sleep(0)
+        with pytest.raises(RuntimeError, match="Non-thread-safe operation"):
+            await asyncio.to_thread(receiver_2.close)
+
+    # The failed wake-up leaves the waiting task stuck, so this loop is closed
+    # without cancelling it.
+    loop = asyncio.new_event_loop()
+    loop.set_debug(True)
+    try:
+        loop.run_until_complete(close_in_thread())
+    finally:
+        loop.close()
+
+
 async def test_broadcast_auto_close_1() -> None:
     """Ensure broadcast auto close works when all receivers are closed."""
     sender, receiver = BroadcastChannel[int](name="auto-close-test")

From a42f4b25dad6c704dc0cf34d8b6a5ee1d18ffe33 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Wed, 21 Oct 2026 15:39:05 +0000
Subject: [PATCH 02/15] Stop reporting closed broadcast receivers as ready

A closed receiver with messages left in its buffer kept returning True
from `ready()`, while `consume()` raised `ReceiverStoppedError`.  So
`select()` yielded stopped selections for it forever, and a `filter()`
on it made `select()` fail.  A closed receiver is now never ready.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py |  4 ++-
 tests/test_broadcast.py             | 39 +++++++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 80e10bf0..b1854e74 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -576,6 +576,8 @@ async def ready(self) -> bool:
         Returns:
             Whether the receiver is still active.
         """
+        if self._closed:
+            return False
         # if there are still messages to consume from the queue, return immediately
         if self._q:
             return True
@@ -590,7 +592,7 @@ async def ready(self) -> bool:
                 await self._channel.aclose()
                 return False
             await self._channel._wakeup.wait()
-        return True
+        return not self._closed
         # pylint: enable=protected-access
 
     @override
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index b0c53c42..db846abe 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -17,6 +17,8 @@
     ReceiverStoppedError,
     Sender,
     SenderError,
+    select,
+    selected_from,
 )
 
 
@@ -441,6 +443,43 @@ async def test_broadcast_close_receiver_while_waiting() -> None:
     assert await asyncio.wait_for(waiter, timeout=1.0) is False
 
 
+async def test_broadcast_closed_receiver_with_buffered_messages() -> None:
+    """Ensure a closed receiver with buffered messages stops `select()`."""
+    chan = Broadcast[int](name="input-chan")
+    sender = chan.new_sender()
+    receiver = chan.new_receiver()
+    filtered = chan.new_receiver().filter(lambda _: True)
+    await sender.send(1)
+    receiver.close()
+    filtered.close()
+
+    assert not await receiver.ready()
+
+    async def stopped(recv: Receiver[int]) -> list[bool]:
+        return [
+            selected.was_stopped
+            async for selected in select(recv)
+            if selected_from(selected, recv)
+        ]
+
+    assert await asyncio.wait_for(stopped(receiver), timeout=1.0) == [True]
+    assert await asyncio.wait_for(stopped(filtered), timeout=1.0) == [True]
+
+
+async def test_broadcast_receiver_closed_while_waking() -> None:
+    """Ensure a receiver closed while it wakes up isn't reported as ready."""
+    chan = Broadcast[int](name="input-chan")
+    sender = chan.new_sender()
+    receiver = chan.new_receiver()
+
+    waiter = asyncio.create_task(receiver.ready())
+    await asyncio.sleep(0)
+    await sender.send(1)
+    receiver.close()
+
+    assert await asyncio.wait_for(waiter, timeout=1.0) is False
+
+
 def _wait_on_closed_loop(receiver: Receiver[int]) -> None:
     """Leave a task waiting on the receiver in an event loop that is then closed."""
 

From cec54abcae4d84e3d734c1ce608f2327f989545d Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:40:13 +0000
Subject: [PATCH 03/15] Auto-close broadcast channels when the last sender is
 dropped

Senders that were garbage-collected without `aclose()` stayed in the
sender count, so their receivers waited forever.  Senders now detach
from the channel when they are closed or garbage-collected, and an
auto-closing channel, like `BroadcastChannel`, closes when the last one
is gone.

The garbage collector can detach a sender in any thread, or in the
middle of `Event.wait()`.  So the detach only updates the channel's
state, and wakes up the receivers through `call_soon_threadsafe()` on
the event loop they wait on.  An explicit `aclose()` still wakes them up
at once.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py |  65 +++++++++++---
 tests/test_broadcast.py             | 128 ++++++++++++++++++++++++++++
 2 files changed, 181 insertions(+), 12 deletions(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index b1854e74..c1cd6ce7 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -5,9 +5,10 @@
 
 from __future__ import annotations
 
+import contextlib
 import logging
 import weakref
-from asyncio import Event
+from asyncio import AbstractEventLoop, Event, get_running_loop
 from collections import deque
 from typing import Generic, TypeVar
 
@@ -225,6 +226,9 @@ def __init__(
         [`BroadcastReceiver.close()`][frequenz.channels.BroadcastReceiver.close].
         """
 
+        self._loop: AbstractEventLoop | None = None
+        """The event loop the receivers last waited on."""
+
         self._sender_count: int = 0
         """The number of senders attached to this channel."""
 
@@ -284,9 +288,17 @@ async def aclose(self) -> None:
         [`receive()`][...Receiver.receive] calls will return `None`
         immediately.
         """
+        self._close()
+
+    def _close(self) -> None:
+        """Close this channel synchronously, see [`.aclose()`][..aclose]."""
+        self._mark_closed()
+        self._wake_receivers()
+
+    def _mark_closed(self) -> None:
+        """Mark this channel as closed, without waking up its receivers."""
         self._latest = None
         self._closed = True
-        self._wake_receivers()
 
     def _wake_receivers(self) -> None:
         """Wake up all receivers waiting for new messages."""
@@ -302,6 +314,32 @@ def _wake_receivers(self) -> None:
             if loop is None or not loop.is_closed():
                 raise
 
+    def _remove_sender(self, *, from_gc: bool = True) -> None:
+        """Detach a closed or garbage-collected sender from this channel.
+
+        The garbage collector can run this in any thread, and in the middle of
+        other code.  So when it does, this only updates the channel's state, and
+        the receivers are woken up later, on the event loop they wait on.
+
+        Args:
+            from_gc: Whether the garbage collector runs this, through the sender's
+                finalizer.
+        """
+        self._sender_count -= 1
+        if self._auto_close_enabled and self._sender_count == 0:
+            if from_gc:
+                self._mark_closed()
+                self._wake_receivers_soon()
+            else:
+                self._close()
+
+    def _wake_receivers_soon(self) -> None:
+        """Schedule waking up the receivers on the event loop they wait on."""
+        if self._loop is not None:
+            # A closed loop raises `RuntimeError`, and has no receivers left to wake.
+            with contextlib.suppress(RuntimeError):
+                self._loop.call_soon_threadsafe(self._wake_receivers)
+
     @deprecated("The close() method is deprecated, use aclose() instead")
     async def close(self) -> None:  # noqa: D402
         """Close the channel, deprecated alias for [`.aclose()`][..aclose]."""  # noqa: D402
@@ -375,7 +413,12 @@ def __init__(self, channel: Broadcast[_T], /) -> None:
         self._closed: bool = False
         """Whether this sender is closed."""
 
-        self._channel._sender_count += 1
+        # pylint: disable=protected-access
+        self._detach = weakref.finalize(self, channel._remove_sender)
+        """Detaches this sender from the channel when it is garbage-collected."""
+
+        channel._sender_count += 1
+        # pylint: enable=protected-access
 
     @property
     def sender_count(self) -> int:
@@ -412,7 +455,7 @@ async def send(self, message: _T, /) -> None:
         for _hash in stale_refs:
             del self._channel._receivers[_hash]
         if self._channel._auto_close_enabled and len(self._channel._receivers) == 0:
-            await self._channel.aclose()
+            self._channel._close()
             raise SenderError("The channel was closed", self) from ChannelClosedError(
                 self._channel
             )
@@ -431,13 +474,9 @@ async def aclose(self) -> None:
         if self._closed:
             return
         self._closed = True
-        self._channel._sender_count -= 1
-
-        if (
-            self._channel._sender_count == 0  # pylint: disable=protected-access
-            and self._channel._auto_close_enabled  # pylint: disable=protected-access
-        ):
-            await self._channel.aclose()
+        if self._detach.detach() is not None:
+            # pylint: disable-next=protected-access
+            self._channel._remove_sender(from_gc=False)
 
     @override
     def clone(self) -> BroadcastSender[_T]:
@@ -582,6 +621,8 @@ async def ready(self) -> bool:
         if self._q:
             return True
 
+        self._channel._loop = get_running_loop()  # pylint: disable=protected-access
+
         # Use a while loop here, because receivers are woken up on every channel
         # state change, not only when a message for this receiver arrives.
         # pylint: disable=protected-access
@@ -589,7 +630,7 @@ async def ready(self) -> bool:
             if self._channel._closed or self._closed:
                 return False
             if self._channel._auto_close_enabled and self._channel._sender_count == 0:
-                await self._channel.aclose()
+                self._channel._close()
                 return False
             await self._channel._wakeup.wait()
         return not self._closed
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index db846abe..c02aa9f7 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -4,6 +4,7 @@
 """Tests for the Broadcast implementation."""
 
 import asyncio
+import gc
 from dataclasses import dataclass
 from typing import TypeGuard, assert_never
 
@@ -585,6 +586,133 @@ async def test_broadcast_auto_close_2() -> None:
     assert isinstance(excinfo.value.__cause__, ChannelClosedError)
 
 
+async def test_broadcast_auto_close_on_last_sender_dropped() -> None:
+    """Ensure dropping the last sender stops a receiver waiting for messages."""
+    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+    sender_2 = sender.clone()
+    await sender.aclose()
+
+    waiter = asyncio.create_task(receiver.receive())
+    await asyncio.sleep(0)
+    assert not waiter.done()
+
+    del sender_2
+    gc.collect()
+
+    with pytest.raises(ReceiverStoppedError) as excinfo:
+        await asyncio.wait_for(waiter, timeout=1.0)
+    assert isinstance(excinfo.value.__cause__, ChannelClosedError)
+
+
+def test_broadcast_auto_close_on_last_sender_dropped_in_thread() -> None:
+    """Ensure a sender garbage-collected in another thread wakes up the receivers."""
+
+    async def drop_sender_in_thread() -> None:
+        sender, receiver = BroadcastChannel[int](name="auto-close-test")
+        senders = [sender]
+        del sender
+
+        waiter = asyncio.create_task(receiver.receive())
+        await asyncio.sleep(0)
+        assert not waiter.done()
+
+        await asyncio.to_thread(senders.clear)
+
+        done, _ = await asyncio.wait({waiter}, timeout=1.0)
+        assert waiter in done
+        with pytest.raises(ReceiverStoppedError) as excinfo:
+            waiter.result()
+        assert isinstance(excinfo.value.__cause__, ChannelClosedError)
+
+    # Debug mode makes the loop raise if the worker thread wakes the receivers
+    # directly, without `call_soon_threadsafe()`, which leaves the waiter stuck.
+    # This loop is closed without cancelling its tasks, so a stuck waiter fails
+    # the test instead of hanging it.
+    loop = asyncio.new_event_loop()
+    loop.set_debug(True)
+    try:
+        loop.run_until_complete(drop_sender_in_thread())
+    finally:
+        loop.close()
+
+
+async def test_broadcast_last_sender_aclose_wakes_receivers() -> None:
+    """Ensure closing the last sender wakes up a waiting receiver at once."""
+    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+
+    waiter = asyncio.create_task(receiver.ready())
+    await asyncio.sleep(0)
+    assert not waiter.done()
+
+    await sender.aclose()
+    await asyncio.sleep(0)
+
+    assert waiter.done()
+    assert waiter.result() is False
+
+
+async def test_broadcast_closed_sender_dropped() -> None:
+    """Ensure a closed sender isn't detached again when it is garbage-collected."""
+    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+    sender_2 = sender.clone()
+
+    await sender.aclose()
+    del sender
+    gc.collect()
+
+    assert sender_2.sender_count == 1
+    await sender_2.send(1)
+    assert await receiver.receive() == 1
+
+
+async def test_broadcast_finalized_sender_closed() -> None:
+    """Ensure closing a sender whose finalizer already ran doesn't detach it again."""
+    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+    sender_2 = sender.clone()
+
+    # Interpreter exit runs the finalizers of senders that are still alive.
+    sender._detach()  # pylint: disable=protected-access
+    await sender.aclose()
+
+    assert sender_2.sender_count == 1
+    await sender_2.send(1)
+    assert await receiver.receive() == 1
+
+
+def test_broadcast_last_sender_dropped_after_loop_closed() -> None:
+    """Ensure dropping the last sender after the receivers' loop closed is fine."""
+    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+
+    async def wait() -> None:
+        # Let the receiver wait, so the channel knows the receivers' event loop.
+        _ = asyncio.create_task(receiver.ready())
+        await asyncio.sleep(0)
+
+    asyncio.run(wait())
+
+    del sender
+    gc.collect()
+
+    with pytest.raises(ReceiverStoppedError):
+        asyncio.run(receiver.receive())
+
+
+async def test_broadcast_no_auto_close_on_sender_dropped() -> None:
+    """Ensure dropping all senders doesn't close a non-auto-closing channel."""
+    chan = Broadcast[int](name="input-chan")
+    receiver = chan.new_receiver()
+
+    sender = chan.new_sender()
+    del sender
+    gc.collect()
+
+    assert not chan.is_closed
+    sender = chan.new_sender()
+    assert sender.sender_count == 1
+    await sender.send(1)
+    assert await receiver.receive() == 1
+
+
 async def test_broadcast_closed_channels_remain_closed() -> None:
     """Ensure that a closed channel can't be resurrected."""
     sender, receiver = BroadcastChannel[int](name="auto-close-test")

From 8278c9760bb3f695d1df62c085ebe763ef9d3533 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Wed, 21 Oct 2026 15:08:04 +0000
Subject: [PATCH 04/15] Keep broadcast channels open when their receivers are
 gone

Like tokio's broadcast channel, a send to an auto-closing broadcast
channel without receivers still fails, but now leaves the channel open,
so new receivers can still be subscribed.  `send()` raises a
`NoReceiversError` and drops the message.  With `resend_latest=True`,
new receivers get the last message that was sent successfully instead.
Such a channel now closes itself only when its senders are gone.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/__init__.py   |  5 ++
 src/frequenz/channels/_broadcast.py | 66 +++++++++++++++------------
 src/frequenz/channels/_sender.py    | 18 ++++++++
 tests/test_broadcast.py             | 71 ++++++++++++++++++++++++++---
 4 files changed, 123 insertions(+), 37 deletions(-)

diff --git a/src/frequenz/channels/__init__.py b/src/frequenz/channels/__init__.py
index 627a890f..b12ad24e 100644
--- a/src/frequenz/channels/__init__.py
+++ b/src/frequenz/channels/__init__.py
@@ -54,6 +54,9 @@
 * [`SenderError`][.SenderError]: Base class for all errors
   related to senders.
 
+* [`NoReceiversError`][.NoReceiversError]: Error raised when sending
+  a message to a [`BroadcastChannel`][.BroadcastChannel] that has no receivers.
+
 * [`ReceiverError`][.ReceiverError]: Base class for all errors
   related to receivers.
 
@@ -104,6 +107,7 @@
 from ._sender import (
     CloneableSender,
     CloneableSubscribableSender,
+    NoReceiversError,
     Sender,
     SenderClosedError,
     SenderError,
@@ -126,6 +130,7 @@
     "LatestValueCache",
     "MappedMessageT_co",
     "Merger",
+    "NoReceiversError",
     "OneshotChannel",
     "OneshotReceiver",
     "OneshotSender",
diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index c1cd6ce7..0d080254 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -17,7 +17,12 @@
 from ._exceptions import ChannelClosedError
 from ._generic import ChannelMessageT
 from ._receiver import Receiver, ReceiverStoppedError
-from ._sender import CloneableSubscribableSender, SenderClosedError, SenderError
+from ._sender import (
+    CloneableSubscribableSender,
+    NoReceiversError,
+    SenderClosedError,
+    SenderError,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -201,15 +206,18 @@ def __init__(
             name: The name of the channel. This is for logging purposes, and it will be
                 shown in the string representation of the channel.
             resend_latest: When True, every time a new receiver is created with
-                [`.new_receiver()`][..new_receiver], the last message seen by the channel
-                will be sent to the
-                new receiver automatically. This allows new receivers on slow streams to
-                get the latest message as soon as they are created, without having to
-                wait for the next message on the channel to arrive.  It is safe to be
-                set in data/reporting channels, but is not recommended for use in
-                channels that stream control instructions.
-            auto_close: If True, the channel will be closed when all senders or all
-                receivers are closed.
+                [`.new_receiver()`][..new_receiver], the last message that was sent
+                successfully will be sent to the new receiver automatically, unless
+                it was `None` or the channel was closed.  This allows new receivers on
+                slow streams to get the latest message as soon as they are created,
+                without having to wait for the next message on the channel to arrive.
+                It is safe to be set in data/reporting channels, but is not
+                recommended for use in channels that stream control instructions.
+            auto_close: If True, the channel is closed when its last sender is
+                closed or garbage-collected, or when a receiver waits for messages
+                while the channel has no senders.  A `send()` while it has no
+                receivers raises a
+                [`NoReceiversError`][frequenz.channels.NoReceiversError].
         """
         self._name: str = name
         """The name of the broadcast channel.
@@ -244,19 +252,12 @@ def __init__(
         """The latest message sent to the channel."""
 
         self._auto_close_enabled: bool = auto_close
-        """Whether to close the channel when all senders or all receivers are closed."""
+        """Whether the channel closes automatically, see the `auto_close` argument."""
 
         self.resend_latest: bool = resend_latest
         """Whether to resend the latest message to new receivers.
 
-        When `True`, every time a new receiver is created with [`.new_receiver()`][..new_receiver],
-        it will
-        automatically get sent the latest message on the channel.  This allows new
-        receivers on slow streams to get the latest message as soon as they are created,
-        without having to wait for the next message on the channel to arrive.
-
-        It is safe to be set in data/reporting channels, but is not recommended for use
-        in channels that stream control instructions.
+        See the `resend_latest` argument of the constructor.
         """
 
     @property
@@ -437,6 +438,10 @@ async def send(self, message: _T, /) -> None:
                 A [`ChannelClosedError`][...ChannelClosedError] is
                 set as the cause.
             SenderClosedError: If this sender was closed.
+            NoReceiversError: If this sender and its channel are open, but the
+                channel has no receivers, and it is a
+                [`BroadcastChannel`][frequenz.channels.BroadcastChannel] or a
+                `Broadcast` created with `auto_close=True`.
         """
         if self._closed:
             raise SenderClosedError(self)
@@ -455,10 +460,7 @@ async def send(self, message: _T, /) -> None:
         for _hash in stale_refs:
             del self._channel._receivers[_hash]
         if self._channel._auto_close_enabled and len(self._channel._receivers) == 0:
-            self._channel._close()
-            raise SenderError("The channel was closed", self) from ChannelClosedError(
-                self._channel
-            )
+            raise NoReceiversError(self)
         self._channel._latest = message
         self._channel._wake_receivers()
         # pylint: enable=protected-access
@@ -739,7 +741,10 @@ class BroadcastChannel(
     [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError].
 
     When all senders of a channel are closed, all its receivers will be
-    automatically closed, and vice versa.
+    automatically closed.  While all its receivers are closed,
+    [`send()`][frequenz.channels.BroadcastSender.send] drops the message and
+    raises a [`NoReceiversError`][frequenz.channels.NoReceiversError], but the
+    channel stays open.
 
     This channel is useful, for example, to implement a pub/sub pattern, where
     multiple consumers can subscribe to a channel to receive all messages.
@@ -865,12 +870,13 @@ def __new__(
                 shown in the string representation of the channel.
             resend_latest: When True, every time a new receiver is created with
                 [`Broadcast.new_receiver()`][...Broadcast.new_receiver],
-                the last message seen by the channel will be sent to the
-                new receiver automatically. This allows new receivers on slow streams to
-                get the latest message as soon as they are created, without having to
-                wait for the next message on the channel to arrive.  It is safe to be
-                set in data/reporting channels, but is not recommended for use in
-                channels that stream control instructions.
+                the last message that was sent successfully will be sent to the new
+                receiver automatically, unless it was `None`.  This allows new
+                receivers on slow streams to get the latest message as soon as they
+                are created, without having to wait for the next message on the
+                channel to arrive.  It is safe to be set in data/reporting channels,
+                but is not recommended for use in channels that stream control
+                instructions.
             limit: Number of messages the receivers can hold in their buffers.
             warn_on_overflow: Whether to log a warning when a receiver's buffer is full
                 and a message is dropped.
diff --git a/src/frequenz/channels/_sender.py b/src/frequenz/channels/_sender.py
index 7149880c..88cded0c 100644
--- a/src/frequenz/channels/_sender.py
+++ b/src/frequenz/channels/_sender.py
@@ -114,6 +114,24 @@ def __init__(self, sender: Sender[SenderMessageT_co]):
         super().__init__("Sender is closed", sender)
 
 
+class NoReceiversError(SenderError[SenderMessageT_co]):
+    """An error indicating that a message was sent to a channel without receivers.
+
+    It is raised by the senders of a
+    [`BroadcastChannel`][frequenz.channels.BroadcastChannel], and of a `Broadcast`
+    created with `auto_close=True`.  The message is dropped, but the channel stays
+    open, so sending works again once a new receiver is attached.
+    """
+
+    def __init__(self, sender: Sender[SenderMessageT_co]):
+        """Initialize this error.
+
+        Args:
+            sender: The [`Sender`][...Sender] that tried to send the message.
+        """
+        super().__init__(f"Sender {sender} has no receivers", sender)
+
+
 class SubscribableSender(Sender[SenderMessageT_contra], ABC):
     """A [`Sender`][..Sender] that can be subscribed to."""
 
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index c02aa9f7..5c5a814c 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -5,6 +5,7 @@
 
 import asyncio
 import gc
+import re
 from dataclasses import dataclass
 from typing import TypeGuard, assert_never
 
@@ -14,6 +15,7 @@
     Broadcast,
     BroadcastChannel,
     ChannelClosedError,
+    NoReceiversError,
     Receiver,
     ReceiverStoppedError,
     Sender,
@@ -539,8 +541,8 @@ async def close_in_thread() -> None:
         loop.close()
 
 
-async def test_broadcast_auto_close_1() -> None:
-    """Ensure broadcast auto close works when all receivers are closed."""
+async def test_broadcast_no_receivers_after_all_receivers_closed() -> None:
+    """Ensure sending fails once all receivers are closed."""
     sender, receiver = BroadcastChannel[int](name="auto-close-test")
 
     receiver_2 = sender.subscribe()
@@ -558,12 +560,16 @@ async def test_broadcast_auto_close_1() -> None:
 
     receiver_2.close()
 
-    with pytest.raises(SenderError) as excinfo:
+    with pytest.raises(
+        SenderError,
+        match=f"^Sender {re.escape(str(sender))} has no receivers$",
+    ) as excinfo:
         await sender.send(3)
-    assert isinstance(excinfo.value.__cause__, ChannelClosedError)
+    assert isinstance(excinfo.value, NoReceiversError)
+    assert excinfo.value.__cause__ is None
 
 
-async def test_broadcast_auto_close_2() -> None:
+async def test_broadcast_auto_close_on_all_senders_closed() -> None:
     """Ensure broadcast auto close works when all senders are closed."""
     sender, receiver = BroadcastChannel[int](name="auto-close-test")
 
@@ -586,6 +592,56 @@ async def test_broadcast_auto_close_2() -> None:
     assert isinstance(excinfo.value.__cause__, ChannelClosedError)
 
 
+async def test_broadcast_subscribe_after_last_receiver_closed() -> None:
+    """Ensure a channel without receivers can get new ones after a failed send."""
+    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+
+    receiver.close()
+    with pytest.raises(NoReceiversError):
+        await sender.send(0)
+
+    receiver_2 = sender.subscribe()
+    await sender.send(1)
+    assert await receiver_2.receive() == 1
+
+
+async def test_broadcast_no_auto_close_send_without_receivers() -> None:
+    """Ensure a non-auto-closing `Broadcast` keeps a message sent without receivers.
+
+    With `resend_latest=True`, a receiver created later gets it.
+    """
+    chan = Broadcast[int](name="no-auto-close-test", resend_latest=True)
+    sender = chan.new_sender()
+
+    await sender.send(1)
+
+    receiver = chan.new_receiver()
+    assert await asyncio.wait_for(receiver.receive(), timeout=1.0) == 1
+
+
+async def test_broadcast_failed_send_not_resent() -> None:
+    """Ensure a message sent without receivers isn't resent to new receivers."""
+    sender, receiver = BroadcastChannel[int](name="resend-test", resend_latest=True)
+    await sender.send(1)
+    receiver.close()
+
+    with pytest.raises(NoReceiversError):
+        await sender.send(2)
+
+    receiver_2 = sender.subscribe()
+    assert await asyncio.wait_for(receiver_2.receive(), timeout=1.0) == 1
+
+
+async def test_broadcast_no_receivers_deprecated_auto_close() -> None:
+    """Ensure an auto-closing `Broadcast` stays open when sending without receivers."""
+    chan = Broadcast[int](name="auto-close-test", auto_close=True)
+    sender = chan.new_sender()
+
+    with pytest.raises(NoReceiversError):
+        await sender.send(1)
+    assert not chan.is_closed
+
+
 async def test_broadcast_auto_close_on_last_sender_dropped() -> None:
     """Ensure dropping the last sender stops a receiver waiting for messages."""
     sender, receiver = BroadcastChannel[int](name="auto-close-test")
@@ -715,9 +771,10 @@ async def test_broadcast_no_auto_close_on_sender_dropped() -> None:
 
 async def test_broadcast_closed_channels_remain_closed() -> None:
     """Ensure that a closed channel can't be resurrected."""
-    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+    chan = Broadcast[int](name="closed-test")
+    sender = chan.new_sender()
 
-    receiver.close()
+    await chan.aclose()
 
     with pytest.raises(SenderError) as excinfo:
         await sender.send(1)

From fb0e42c6167200aee9693efb73f099964b9d6778 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Wed, 21 Oct 2026 15:08:39 +0000
Subject: [PATCH 05/15] Detach garbage-collected broadcast receivers at once

Garbage-collected receivers stayed in the channel until the next
`send()` swept them out.  They now detach as soon as they are collected,
so the channel doesn't keep dead receivers between sends.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 34 +++++++++++++++++------------
 tests/test_broadcast.py             | 11 ++++++++++
 2 files changed, 31 insertions(+), 14 deletions(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 0d080254..b31bff7a 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -341,6 +341,14 @@ def _wake_receivers_soon(self) -> None:
             with contextlib.suppress(RuntimeError):
                 self._loop.call_soon_threadsafe(self._wake_receivers)
 
+    def _remove_receiver(self, key: int) -> None:
+        """Detach a closed or garbage-collected receiver from this channel.
+
+        Args:
+            key: The key of the receiver in `self._receivers`.
+        """
+        self._receivers.pop(key, None)
+
     @deprecated("The close() method is deprecated, use aclose() instead")
     async def close(self) -> None:  # noqa: D402
         """Close the channel, deprecated alias for [`.aclose()`][..aclose]."""  # noqa: D402
@@ -450,17 +458,13 @@ async def send(self, message: _T, /) -> None:
             raise SenderError("The channel was closed", self) from ChannelClosedError(
                 self._channel
             )
-        stale_refs = []
-        for _hash, recv_ref in self._channel._receivers.items():
-            recv = recv_ref()
-            if recv is None:
-                stale_refs.append(_hash)
-                continue
-            recv._enqueue(message)
-        for _hash in stale_refs:
-            del self._channel._receivers[_hash]
-        if self._channel._auto_close_enabled and len(self._channel._receivers) == 0:
+        if self._channel._auto_close_enabled and not self._channel._receivers:
             raise NoReceiversError(self)
+        # Iterate over a copy, as receivers garbage-collected during the loop remove
+        # themselves from the dict.
+        for recv_ref in list(self._channel._receivers.values()):
+            if (recv := recv_ref()) is not None:
+                recv._enqueue(message)
         self._channel._latest = message
         self._channel._wake_receivers()
         # pylint: enable=protected-access
@@ -577,6 +581,10 @@ def __init__(
         self._closed: bool = False
         """Whether the receiver is closed."""
 
+        # pylint: disable-next=protected-access
+        self._detach = weakref.finalize(self, channel._remove_receiver, hash(self))
+        """Detaches this receiver from the channel when closed or garbage-collected."""
+
     def _enqueue(self, message: _T, /) -> None:
         """Put a message into this receiver's queue.
 
@@ -666,10 +674,8 @@ def close(self) -> None:
         [`ReceiverStoppedError`][...ReceiverStoppedError].
         """
         self._closed = True
-        # pylint: disable=protected-access
-        self._channel._receivers.pop(hash(self), None)
-        self._channel._wake_receivers()
-        # pylint: enable=protected-access
+        self._detach()
+        self._channel._wake_receivers()  # pylint: disable=protected-access
 
     def __str__(self) -> str:
         """Return a string representation of this receiver."""
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index 5c5a814c..c5b1f8d1 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -605,6 +605,17 @@ async def test_broadcast_subscribe_after_last_receiver_closed() -> None:
     assert await receiver_2.receive() == 1
 
 
+async def test_broadcast_no_receivers_after_last_receiver_dropped() -> None:
+    """Ensure sending fails once the last receiver is garbage-collected."""
+    sender, receiver = BroadcastChannel[int](name="auto-close-test")
+
+    del receiver
+    gc.collect()
+
+    with pytest.raises(NoReceiversError):
+        await sender.send(1)
+
+
 async def test_broadcast_no_auto_close_send_without_receivers() -> None:
     """Ensure a non-auto-closing `Broadcast` keeps a message sent without receivers.
 

From 1da3a65de9e5f328c5bc1632fcf6620126889658 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:07:32 +0000
Subject: [PATCH 06/15] Refuse to clone closed broadcast senders

Unlike `send()` and `subscribe()`, `clone()` worked on a closed sender
or channel.  On a closed channel, it handed out senders that could never
send.  It now raises like `send()` and `subscribe()`.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 16 +++++++++++++++-
 tests/test_broadcast.py             | 20 ++++++++++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index b31bff7a..ca744e8b 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -486,7 +486,21 @@ async def aclose(self) -> None:
 
     @override
     def clone(self) -> BroadcastSender[_T]:
-        """Return a clone of this sender."""
+        """Return a clone of this sender.
+
+        Returns:
+            A new sender attached to this sender's channel.
+
+        Raises:
+            SenderError: If the underlying channel is closed.
+            SenderClosedError: If this sender is closed.
+        """
+        if self._closed:
+            raise SenderClosedError(self)
+        if self._channel._closed:  # pylint: disable=protected-access
+            raise SenderError("The channel was closed", self) from ChannelClosedError(
+                self._channel
+            )
         return BroadcastSender(self._channel)
 
     @override
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index c5b1f8d1..515f386c 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -19,6 +19,7 @@
     Receiver,
     ReceiverStoppedError,
     Sender,
+    SenderClosedError,
     SenderError,
     select,
     selected_from,
@@ -794,3 +795,22 @@ async def test_broadcast_closed_channels_remain_closed() -> None:
     with pytest.raises(SenderError) as excinfo:
         _ = sender.subscribe()
     assert isinstance(excinfo.value.__cause__, ChannelClosedError)
+
+    with pytest.raises(SenderError) as excinfo:
+        _ = sender.clone()
+    assert isinstance(excinfo.value.__cause__, ChannelClosedError)
+
+
+async def test_broadcast_closed_sender() -> None:
+    """Ensure a closed sender can't be used to send, clone or subscribe."""
+    sender, _receiver = BroadcastChannel[int](name="closed-sender-test")
+    sender_2 = sender.clone()
+    await sender.aclose()
+    assert sender_2.sender_count == 1
+
+    with pytest.raises(SenderClosedError):
+        await sender.send(1)
+    with pytest.raises(SenderClosedError):
+        _ = sender.clone()
+    with pytest.raises(SenderClosedError):
+        _ = sender.subscribe()

From 36cdadd5f19c23631fe2274cc133087b543147e4 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:08:01 +0000
Subject: [PATCH 07/15] Apply BroadcastChannel receiver options to subscribed
 receivers

`limit` and `warn_on_overflow` were documented as applying to the
channel's receivers, but only the initial receiver got them;
`subscribe()` always fell back to 50 and `True`.  They are now the
defaults for `subscribe()`.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 55 +++++++++++++++++++----------
 tests/test_broadcast.py             | 18 ++++++++++
 2 files changed, 54 insertions(+), 19 deletions(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index ca744e8b..98c2ca43 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -254,6 +254,12 @@ def __init__(
         self._auto_close_enabled: bool = auto_close
         """Whether the channel closes automatically, see the `auto_close` argument."""
 
+        self._subscriber_limit: int = 50
+        """The default buffer size of receivers created with `BroadcastSender.subscribe()`."""
+
+        self._subscriber_warn_on_overflow: bool = True
+        """The default overflow warning setting of subscribed receivers."""
+
         self.resend_latest: bool = resend_latest
         """Whether to resend the latest message to new receivers.
 
@@ -507,16 +513,20 @@ def clone(self) -> BroadcastSender[_T]:
     def subscribe(
         self,
         name: str | None = None,
-        limit: int = 50,
-        warn_on_overflow: bool = True,
+        limit: int | None = None,
+        warn_on_overflow: bool | None = None,
     ) -> BroadcastReceiver[_T]:
         """Return a new receiver attached to this sender's channel.
 
         Args:
             name: A name to identify the receiver in the logs.
-            limit: Number of messages the receiver can hold in its buffer.
+            limit: Number of messages the receiver can hold in its buffer.  If
+                `None`, the `limit` the `BroadcastChannel` was created with is used,
+                or `50` for a `Broadcast`.
             warn_on_overflow: Whether to log a warning when the receiver's buffer is
-                full and a message is dropped.
+                full and a message is dropped.  If `None`, the `warn_on_overflow`
+                setting the `BroadcastChannel` was created with is used, or `True`
+                for a `Broadcast`.
 
         Returns:
             A new receiver attached to this sender's channel.
@@ -527,13 +537,21 @@ def subscribe(
         """
         if self._closed:
             raise SenderClosedError(self)
-        if self._channel._closed:  # pylint: disable=protected-access
+        # pylint: disable=protected-access
+        if self._channel._closed:
             raise SenderError("The channel was closed", self) from ChannelClosedError(
                 self._channel
             )
         return self._channel.new_receiver(
-            name=name, limit=limit, warn_on_overflow=warn_on_overflow
+            name=name,
+            limit=self._channel._subscriber_limit if limit is None else limit,
+            warn_on_overflow=(
+                self._channel._subscriber_warn_on_overflow
+                if warn_on_overflow is None
+                else warn_on_overflow
+            ),
         )
+        # pylint: enable=protected-access
 
     def __str__(self) -> str:
         """Return a string representation of this sender."""
@@ -897,9 +915,13 @@ def __new__(
                 channel to arrive.  It is safe to be set in data/reporting channels,
                 but is not recommended for use in channels that stream control
                 instructions.
-            limit: Number of messages the receivers can hold in their buffers.
+            limit: Number of messages each receiver can hold in its buffer, for the
+                initial receiver and, by default, for receivers created with
+                [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe].
             warn_on_overflow: Whether to log a warning when a receiver's buffer is full
-                and a message is dropped.
+                and a message is dropped.  Applies to the initial receiver and, by
+                default, to receivers created with
+                [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe].
 
         Returns:
             A new BroadcastChannel instance that can be destructured into an initial
@@ -908,14 +930,9 @@ def __new__(
         channel = Broadcast[ChannelMessageT](
             name=name, resend_latest=resend_latest, auto_close=True
         )
-        return tuple.__new__(
-            cls,
-            (
-                channel.new_sender(),
-                channel.new_receiver(
-                    name=f"{name}_receiver",
-                    limit=limit,
-                    warn_on_overflow=warn_on_overflow,
-                ),
-            ),
-        )
+        # pylint: disable=protected-access
+        channel._subscriber_limit = limit
+        channel._subscriber_warn_on_overflow = warn_on_overflow
+        # pylint: enable=protected-access
+        sender = channel.new_sender()
+        return tuple.__new__(cls, (sender, sender.subscribe(name=f"{name}_receiver")))
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index 515f386c..5d085828 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -814,3 +814,21 @@ async def test_broadcast_closed_sender() -> None:
         _ = sender.clone()
     with pytest.raises(SenderClosedError):
         _ = sender.subscribe()
+
+
+async def test_broadcast_subscribe_inherits_receiver_options() -> None:
+    """Ensure subscribed receivers default to the channel's receiver options."""
+    sender, receiver = BroadcastChannel[int](
+        name="options-test", limit=2, warn_on_overflow=False
+    )
+    subscribed = sender.subscribe()
+    overridden = sender.subscribe(limit=3, warn_on_overflow=True)
+
+    for i in range(4):
+        await sender.send(i)
+
+    assert [len(receiver), len(subscribed), len(overridden)] == [2, 2, 3]
+    # pylint: disable=protected-access
+    assert not subscribed._warn_on_overflow
+    assert overridden._warn_on_overflow
+    # pylint: enable=protected-access

From a1cff07260d531019eb01e66dd08f1d987642a4c Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:08:29 +0000
Subject: [PATCH 08/15] Add has_receivers and has_senders to broadcast
 endpoints

Lets producers stop when nobody is listening anymore, and consumers tell
whether more messages can still arrive, without having to send or wait
on the channel.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 30 ++++++++++++++++++
 tests/test_broadcast.py             | 49 +++++++++++++++++++++++++++++
 2 files changed, 79 insertions(+)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 98c2ca43..3b6d41b4 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -440,6 +440,22 @@ def sender_count(self) -> int:
         """The number of open senders attached to this sender's channel."""
         return self._channel._sender_count  # pylint: disable=protected-access
 
+    @property
+    def has_receivers(self) -> bool:
+        """Whether this sender's channel is open and has any receivers.
+
+        Producers can check this to stop producing messages that nobody would
+        receive.  When this is `False` on an open
+        [`BroadcastChannel`][frequenz.channels.BroadcastChannel],
+        [`send()`][..send] on an open sender raises a
+        [`NoReceiversError`][frequenz.channels.NoReceiversError] until a new
+        receiver is created with [`subscribe()`][..subscribe].
+
+        This doesn't check whether this sender itself is closed.
+        """
+        # pylint: disable=protected-access
+        return not self._channel._closed and bool(self._channel._receivers)
+
     @override
     async def send(self, message: _T, /) -> None:
         """Send a message to all broadcast receivers.
@@ -636,6 +652,20 @@ def _enqueue(self, message: _T, /) -> None:
                 )
         self._q.append(message)
 
+    @property
+    def has_senders(self) -> bool:
+        """Whether this receiver's channel is open and has any senders.
+
+        A [`BroadcastChannel`][frequenz.channels.BroadcastChannel] is closed as soon
+        as its last sender is closed or garbage-collected.  So once this is `False`
+        on a `BroadcastChannel`, it stays `False`, and no new messages will arrive.
+        The messages already in this receiver's buffer can still be received.
+
+        This doesn't check whether this receiver itself is closed.
+        """
+        # pylint: disable=protected-access
+        return not self._channel._closed and self._channel._sender_count > 0
+
     def __len__(self) -> int:
         """Return the number of unconsumed messages in the broadcast receiver.
 
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index 5d085828..c35dce86 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -832,3 +832,52 @@ async def test_broadcast_subscribe_inherits_receiver_options() -> None:
     assert not subscribed._warn_on_overflow
     assert overridden._warn_on_overflow
     # pylint: enable=protected-access
+
+
+async def test_broadcast_has_receivers() -> None:
+    """Ensure senders report whether their channel has receivers."""
+    sender, receiver = BroadcastChannel[int](name="has-receivers-test")
+    receiver_2 = sender.subscribe()
+    assert sender.has_receivers
+
+    receiver.close()
+    assert sender.has_receivers
+
+    receiver_2.close()
+    assert not sender.has_receivers
+
+
+async def test_broadcast_has_senders() -> None:
+    """Ensure receivers report whether their channel has senders."""
+    sender, receiver = BroadcastChannel[int](name="has-senders-test")
+    sender_2 = sender.clone()
+    assert receiver.has_senders
+
+    await sender.aclose()
+    assert receiver.has_senders
+
+    await sender_2.send(1)
+    await sender_2.aclose()
+    assert not receiver.has_senders
+    assert await receiver.receive() == 1
+
+
+async def test_broadcast_has_receivers_and_senders_no_auto_close() -> None:
+    """Ensure non-auto-closing channels report their senders and receivers."""
+    chan = Broadcast[int](name="input-chan")
+    sender = chan.new_sender()
+    assert not sender.has_receivers
+
+    receiver = chan.new_receiver()
+    assert sender.has_receivers
+    assert receiver.has_senders
+
+    await sender.aclose()
+    assert not receiver.has_senders
+
+    sender_2 = chan.new_sender()
+    assert receiver.has_senders
+
+    await chan.aclose()
+    assert not receiver.has_senders
+    assert not sender_2.has_receivers

From babcbda85bb1d3ed57b3c5e17a979738ef8ae20b Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:11:08 +0000
Subject: [PATCH 09/15] Stop BroadcastChannel from emitting a deprecation
 warning

`BroadcastChannel` was built on the deprecated `Broadcast` class, so
creating one warned users to "use BroadcastChannel instead".  The
implementation now lives in a private class that both use, and only
`Broadcast` is deprecated.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 348 ++++++++++++++--------------
 tests/test_broadcast.py             |  23 ++
 2 files changed, 201 insertions(+), 170 deletions(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 3b6d41b4..89a8ab42 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -27,170 +27,14 @@
 _logger = logging.getLogger(__name__)
 
 
-@deprecated("Please use BroadcastChannel instead.")
-class Broadcast(  # pylint: disable=too-many-instance-attributes
+class _Broadcast(  # pylint: disable=too-many-instance-attributes
     Generic[ChannelMessageT]
 ):
-    """A channel that delivers all messages to all receivers.
-
-    # Description
-
-    [`Broadcast`][frequenz.channels.Broadcast] channels can have multiple
-    [senders][frequenz.channels.Sender] and multiple
-    [receivers][frequenz.channels.Receiver]. Each message sent through any of the
-    senders will be received by all receivers.
-
-    <center>
-    ```bob
-    .---------. msg1                           msg1,msg2  .-----------.
-    | Sender  +------.                        .---------->| Receiver  |
-    '---------'      |      .----------.     |            '-----------'
-                     +----->| Channel  +-----+
-    .---------.      |      '----------'     |            .-----------.
-    | Sender  +------'                       '----------->| Receiver  |
-    '---------' msg2                           msg1,msg2  '-----------'
-    ```
-    </center>
-
-    !!! Note inline end "Characteristics"
-
-        * **Buffered:** Yes, with one buffer per receiver
-        * **Buffer full policy:** Drop oldest message
-        * **Multiple receivers:** Yes
-        * **Multiple senders:** Yes
-        * **Thread-safe:** No
-
-    This channel is buffered, and when messages are not being consumed fast
-    enough and the buffer fills up, old messages will get dropped.
-
-    Each receiver has its own buffer, so messages will only be dropped for
-    receivers that can't keep up with the senders, and not for the whole
-    channel.
-
-    To create a new [senders][frequenz.channels.Sender] and
-    [receivers][frequenz.channels.Receiver] you can use the
-    [`.new_sender()`][.new_sender] and
-    [`.new_receiver()`][.new_receiver] methods
-    respectively.
-
-    When a channel is not needed anymore, it should be closed with
-    [`.aclose()`][.aclose]. This will prevent further
-    attempts to [`send()`][frequenz.channels.Sender.send] data, and will allow
-    receivers to drain the pending items on their queues, but after that,
-    subsequent [`receive()`][frequenz.channels.Receiver.receive] calls will
-    raise a [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError].
-
-    This channel is useful, for example, to implement a pub/sub pattern, where
-    multiple receivers can subscribe to a channel to receive all messages.
-
-    In cases where each message needs to be delivered only to one receiver, an
-    [anycast][frequenz.channels.Anycast] channel may be used.
-
-    # Examples
-
-    Example: Send a few numbers to a receiver
-        This is a very simple example that sends a few numbers from a single sender to
-        a single receiver.
-
-        ```python
-        import asyncio
-
-        from frequenz.channels import Broadcast, Sender
-
-
-        async def send(sender: Sender[int]) -> None:
-            for message in range(3):
-                print(f"sending {message}")
-                await sender.send(message)
-
-
-        async def main() -> None:
-            channel = Broadcast[int](name="numbers")
-
-            sender = channel.new_sender()
-            receiver = channel.new_receiver()
-
-            async with asyncio.TaskGroup() as task_group:
-                task_group.create_task(send(sender))
-                for _ in range(3):
-                    message = await receiver.receive()
-                    print(f"received {message}")
-                    await asyncio.sleep(0.1)  # sleep (or work) with the data
-
-
-        asyncio.run(main())
-        ```
-
-        The output should look something like (although the sending and received might
-        appear more interleaved):
-
-        ```
-        sending 0
-        sending 1
-        sending 2
-        received 0
-        received 1
-        received 2
-        ```
+    """The implementation of broadcast channels.
 
-    Example: Send a few number from multiple senders to multiple receivers
-        This is a more complex example that sends a few numbers from multiple senders to
-        multiple receivers, using a small buffer to force the senders to block.
-
-        ```python
-        import asyncio
-
-        from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError, Sender
-
-
-        async def send(name: str, sender: Sender[int], start: int, stop: int) -> None:
-            for message in range(start, stop):
-                print(f"{name} sending {message}")
-                await sender.send(message)
-
-
-        async def recv(name: str, receiver: Receiver[int]) -> None:
-            try:
-                async for message in receiver:
-                    print(f"{name} received {message}")
-                await asyncio.sleep(0.1)  # sleep (or work) with the data
-            except ReceiverStoppedError:
-                pass
-
-
-        async def main() -> None:
-            acast = Broadcast[int](name="numbers")
-
-            async with asyncio.TaskGroup() as task_group:
-                task_group.create_task(send("sender_1", acast.new_sender(), 10, 13))
-                task_group.create_task(send("sender_2", acast.new_sender(), 20, 22))
-                task_group.create_task(recv("receiver_1", acast.new_receiver()))
-                task_group.create_task(recv("receiver_2", acast.new_receiver()))
-
-
-        asyncio.run(main())
-        ```
-
-        The output should look something like this(although the sending and received
-        might appear interleaved in a different way):
-
-        ```
-        sender_1 sending 10
-        sender_1 sending 11
-        sender_1 sending 12
-        sender_2 sending 20
-        sender_2 sending 21
-        receiver_1 received 10
-        receiver_1 received 11
-        receiver_1 received 12
-        receiver_1 received 20
-        receiver_1 received 21
-        receiver_2 received 10
-        receiver_2 received 11
-        receiver_2 received 12
-        receiver_2 received 20
-        receiver_2 received 21
-        ```
+    It is shared by the deprecated [`Broadcast`][frequenz.channels.Broadcast] class
+    and by [`BroadcastChannel`][frequenz.channels.BroadcastChannel], and is not
+    deprecated itself.
     """
 
     def __init__(
@@ -392,12 +236,12 @@ def new_receiver(
 
     def __str__(self) -> str:
         """Return a string representation of this channel."""
-        return f"{type(self).__name__}:{self._name}"
+        return f"{type(self).__name__.removeprefix('_')}:{self._name}"
 
     def __repr__(self) -> str:
         """Return a string representation of this channel."""
         return (
-            f"{type(self).__name__}(name={self._name!r}, "
+            f"{type(self).__name__.removeprefix('_')}(name={self._name!r}, "
             f"resend_latest={self.resend_latest!r}):<"
             f"latest={self._latest!r}, "
             f"receivers={len(self._receivers)!r}, "
@@ -405,6 +249,171 @@ def __repr__(self) -> str:
         )
 
 
+@deprecated("Please use BroadcastChannel instead.")
+class Broadcast(_Broadcast[ChannelMessageT]):
+    """A channel that delivers all messages to all receivers.
+
+    # Description
+
+    [`Broadcast`][frequenz.channels.Broadcast] channels can have multiple
+    [senders][frequenz.channels.Sender] and multiple
+    [receivers][frequenz.channels.Receiver]. Each message sent through any of the
+    senders will be received by all receivers.
+
+    <center>
+    ```bob
+    .---------. msg1                           msg1,msg2  .-----------.
+    | Sender  +------.                        .---------->| Receiver  |
+    '---------'      |      .----------.     |            '-----------'
+                     +----->| Channel  +-----+
+    .---------.      |      '----------'     |            .-----------.
+    | Sender  +------'                       '----------->| Receiver  |
+    '---------' msg2                           msg1,msg2  '-----------'
+    ```
+    </center>
+
+    !!! Note inline end "Characteristics"
+
+        * **Buffered:** Yes, with one buffer per receiver
+        * **Buffer full policy:** Drop oldest message
+        * **Multiple receivers:** Yes
+        * **Multiple senders:** Yes
+        * **Thread-safe:** No
+
+    This channel is buffered, and when messages are not being consumed fast
+    enough and the buffer fills up, old messages will get dropped.
+
+    Each receiver has its own buffer, so messages will only be dropped for
+    receivers that can't keep up with the senders, and not for the whole
+    channel.
+
+    To create a new [senders][frequenz.channels.Sender] and
+    [receivers][frequenz.channels.Receiver] you can use the
+    [`.new_sender()`][.new_sender] and
+    [`.new_receiver()`][.new_receiver] methods
+    respectively.
+
+    When a channel is not needed anymore, it should be closed with
+    [`.aclose()`][.aclose]. This will prevent further
+    attempts to [`send()`][frequenz.channels.Sender.send] data, and will allow
+    receivers to drain the pending items on their queues, but after that,
+    subsequent [`receive()`][frequenz.channels.Receiver.receive] calls will
+    raise a [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError].
+
+    This channel is useful, for example, to implement a pub/sub pattern, where
+    multiple receivers can subscribe to a channel to receive all messages.
+
+    In cases where each message needs to be delivered only to one receiver, an
+    [anycast][frequenz.channels.Anycast] channel may be used.
+
+    # Examples
+
+    Example: Send a few numbers to a receiver
+        This is a very simple example that sends a few numbers from a single sender to
+        a single receiver.
+
+        ```python
+        import asyncio
+
+        from frequenz.channels import Broadcast, Sender
+
+
+        async def send(sender: Sender[int]) -> None:
+            for message in range(3):
+                print(f"sending {message}")
+                await sender.send(message)
+
+
+        async def main() -> None:
+            channel = Broadcast[int](name="numbers")
+
+            sender = channel.new_sender()
+            receiver = channel.new_receiver()
+
+            async with asyncio.TaskGroup() as task_group:
+                task_group.create_task(send(sender))
+                for _ in range(3):
+                    message = await receiver.receive()
+                    print(f"received {message}")
+                    await asyncio.sleep(0.1)  # sleep (or work) with the data
+
+
+        asyncio.run(main())
+        ```
+
+        The output should look something like (although the sending and received might
+        appear more interleaved):
+
+        ```
+        sending 0
+        sending 1
+        sending 2
+        received 0
+        received 1
+        received 2
+        ```
+
+    Example: Send a few number from multiple senders to multiple receivers
+        This is a more complex example that sends a few numbers from multiple senders to
+        multiple receivers, using a small buffer to force the senders to block.
+
+        ```python
+        import asyncio
+
+        from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError, Sender
+
+
+        async def send(name: str, sender: Sender[int], start: int, stop: int) -> None:
+            for message in range(start, stop):
+                print(f"{name} sending {message}")
+                await sender.send(message)
+
+
+        async def recv(name: str, receiver: Receiver[int]) -> None:
+            try:
+                async for message in receiver:
+                    print(f"{name} received {message}")
+                await asyncio.sleep(0.1)  # sleep (or work) with the data
+            except ReceiverStoppedError:
+                pass
+
+
+        async def main() -> None:
+            acast = Broadcast[int](name="numbers")
+
+            async with asyncio.TaskGroup() as task_group:
+                task_group.create_task(send("sender_1", acast.new_sender(), 10, 13))
+                task_group.create_task(send("sender_2", acast.new_sender(), 20, 22))
+                task_group.create_task(recv("receiver_1", acast.new_receiver()))
+                task_group.create_task(recv("receiver_2", acast.new_receiver()))
+
+
+        asyncio.run(main())
+        ```
+
+        The output should look something like this(although the sending and received
+        might appear interleaved in a different way):
+
+        ```
+        sender_1 sending 10
+        sender_1 sending 11
+        sender_1 sending 12
+        sender_2 sending 20
+        sender_2 sending 21
+        receiver_1 received 10
+        receiver_1 received 11
+        receiver_1 received 12
+        receiver_1 received 20
+        receiver_1 received 21
+        receiver_2 received 10
+        receiver_2 received 11
+        receiver_2 received 12
+        receiver_2 received 20
+        receiver_2 received 21
+        ```
+    """
+
+
 _T = TypeVar("_T")
 
 
@@ -416,13 +425,13 @@ class BroadcastSender(CloneableSubscribableSender[_T]):
     method of an existing sender.
     """
 
-    def __init__(self, channel: Broadcast[_T], /) -> None:
+    def __init__(self, channel: _Broadcast[_T], /) -> None:
         """Initialize this sender.
 
         Args:
             channel: A reference to the broadcast channel this sender belongs to.
         """
-        self._channel: Broadcast[_T] = channel
+        self._channel: _Broadcast[_T] = channel
         """The broadcast channel this sender belongs to."""
 
         self._closed: bool = False
@@ -588,7 +597,7 @@ class BroadcastReceiver(Receiver[_T]):
 
     def __init__(
         self,
-        channel: Broadcast[_T],
+        channel: _Broadcast[_T],
         /,
         *,
         name: str | None,
@@ -602,8 +611,7 @@ def __init__(
         get dropped just in this receiver.
 
         Args:
-            channel: A reference to the [`Broadcast`][...Broadcast]
-                channel that this receiver belongs to.
+            channel: A reference to the broadcast channel this receiver belongs to.
             name: A name to identify the receiver in the logs. If `None` an
                 `id(self)`-based name will be used.  This is only for debugging
                 purposes, it will be shown in the string representation of the
@@ -620,7 +628,7 @@ def __init__(
         Only used for debugging purposes.
         """
 
-        self._channel: Broadcast[_T] = channel
+        self._channel: _Broadcast[_T] = channel
         """The broadcast channel that this receiver belongs to."""
 
         self._q: deque[_T] = deque(maxlen=limit)
@@ -957,7 +965,7 @@ def __new__(
             A new BroadcastChannel instance that can be destructured into an initial
                 sender and receiver.
         """
-        channel = Broadcast[ChannelMessageT](
+        channel = _Broadcast[ChannelMessageT](
             name=name, resend_latest=resend_latest, auto_close=True
         )
         # pylint: disable=protected-access
diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py
index c35dce86..d537e638 100644
--- a/tests/test_broadcast.py
+++ b/tests/test_broadcast.py
@@ -6,6 +6,7 @@
 import asyncio
 import gc
 import re
+import warnings
 from dataclasses import dataclass
 from typing import TypeGuard, assert_never
 
@@ -881,3 +882,25 @@ async def test_broadcast_has_receivers_and_senders_no_auto_close() -> None:
     await chan.aclose()
     assert not receiver.has_senders
     assert not sender_2.has_receivers
+
+
+async def test_broadcast_channel_str() -> None:
+    """Ensure BroadcastChannel endpoints are named like Broadcast ones."""
+    sender, receiver = BroadcastChannel[int](name="str-test")
+
+    assert str(sender) == "Broadcast:str-test:BroadcastSender"
+    assert str(receiver) == "Broadcast:str-test:BroadcastReceiver:str-test_receiver"
+    assert repr(sender).startswith("BroadcastSender(Broadcast(name='str-test'")
+
+
+def test_broadcast_deprecated() -> None:
+    """Ensure creating a Broadcast emits a deprecation warning."""
+    with pytest.deprecated_call(match="Please use BroadcastChannel instead."):
+        _ = Broadcast[int](name="deprecated-test")
+
+
+async def test_broadcast_channel_not_deprecated() -> None:
+    """Ensure creating a BroadcastChannel doesn't emit deprecation warnings."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        _ = BroadcastChannel[int](name="no-warnings-test")

From 1fdbbe570b1ad677e80f03936e90c8c73efefcfb Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:40:23 +0000
Subject: [PATCH 10/15] Document how BroadcastChannel closes itself

About auto-closing, the docs only said that closing all senders closes
all receivers, and what a send without receivers does.  They now also
cover garbage-collected senders and receivers, what each side sees, and
`has_receivers`/`has_senders`.

The multi-sender example is fixed too: it claimed to use a small buffer
to make the senders block, it caught a `ReceiverStoppedError` that
`async for` never raises, and its receivers slept once after the loop
instead of after each message.

The first example now stops when its sender is closed.  The
`BroadcastSender` and `BroadcastReceiver` docs now point to
`BroadcastChannel`, the `resend_latest` docs point to `subscribe()`, and
`Broadcast` is marked as deprecated.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 158 ++++++++++++++++++++--------
 1 file changed, 116 insertions(+), 42 deletions(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 89a8ab42..34f0580d 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -3,6 +3,9 @@
 
 """A channel to broadcast messages to all receivers."""
 
+# The docstrings carry long examples.
+# pylint: disable=too-many-lines
+
 from __future__ import annotations
 
 import contextlib
@@ -253,6 +256,10 @@ def __repr__(self) -> str:
 class Broadcast(_Broadcast[ChannelMessageT]):
     """A channel that delivers all messages to all receivers.
 
+    !!! Warning "Deprecated"
+
+        Please use [`BroadcastChannel`][frequenz.channels.BroadcastChannel] instead.
+
     # Description
 
     [`Broadcast`][frequenz.channels.Broadcast] channels can have multiple
@@ -420,9 +427,9 @@ async def main() -> None:
 class BroadcastSender(CloneableSubscribableSender[_T]):
     """A sender to send messages to the broadcast channel.
 
-    Should not be created directly, but through the
-    [`.clone()`][.clone]
-    method of an existing sender.
+    Should not be created directly, but through
+    [`BroadcastChannel`][frequenz.channels.BroadcastChannel] or the
+    [`.clone()`][.clone] method of an existing sender.
     """
 
     def __init__(self, channel: _Broadcast[_T], /) -> None:
@@ -590,7 +597,8 @@ def __repr__(self) -> str:
 class BroadcastReceiver(Receiver[_T]):
     """A receiver to receive messages from the broadcast channel.
 
-    Should not be created directly, but through the
+    Should not be created directly, but through
+    [`BroadcastChannel`][frequenz.channels.BroadcastChannel] or the
     [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe]
     method of an existing sender.
     """
@@ -800,27 +808,46 @@ class BroadcastChannel(
     receivers that can't keep up with the senders, and not for the whole
     channel.
 
-    Instantiating this class will create a new broadcast channel, and return an
-    initial sender and a receiver.  Further senders and receivers can be created
-    with the [`BroadcastSender.clone()`][frequenz.channels.BroadcastSender.clone],
+    Instantiating this class creates a new broadcast channel, and returns an
+    initial sender and receiver.  Further senders and receivers can be created
+    with the [`BroadcastSender.clone()`][frequenz.channels.BroadcastSender.clone]
     and
     [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe]
     methods respectively.
 
     When a sender or a receiver is not needed anymore, it should be closed with
     [`BroadcastSender.aclose()`][frequenz.channels.BroadcastSender.aclose] or
-    [`BroadcastReceiver.close()`][frequenz.channels.BroadcastReceiver.close]. This will prevent
-    further attempts to [`send()`][frequenz.channels.BroadcastSender.send] data,
-    and will allow receivers to drain the pending items on their queues, but
-    after that, subsequent [`receive()`][frequenz.channels.Receiver.receive] calls
-    will raise a
-    [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError].
-
-    When all senders of a channel are closed, all its receivers will be
-    automatically closed.  While all its receivers are closed,
-    [`send()`][frequenz.channels.BroadcastSender.send] drops the message and
-    raises a [`NoReceiversError`][frequenz.channels.NoReceiversError], but the
-    channel stays open.
+    [`BroadcastReceiver.close()`][frequenz.channels.BroadcastReceiver.close].
+
+    The channel closes itself when all its senders are gone.  Senders and receivers
+    are gone when they are closed or garbage-collected.  This is what happens when
+    either side is gone:
+
+    * As soon as all senders are gone, the channel is closed.  Receivers can still
+      drain the messages pending in their buffers, but after that,
+      [`receive()`][frequenz.channels.Receiver.receive] raises a
+      [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError], and `async
+      for` loops over them end.
+
+    * While all receivers are gone,
+      [`send()`][frequenz.channels.BroadcastSender.send] drops the message and
+      raises a [`NoReceiversError`][frequenz.channels.NoReceiversError].  The
+      channel stays open, so once
+      [`subscribe()`][frequenz.channels.BroadcastSender.subscribe] creates a new
+      receiver, sending works again.
+
+    Senders can check
+    [`has_receivers`][frequenz.channels.BroadcastSender.has_receivers] to stop
+    producing messages that nobody would receive, and receivers can check
+    [`has_senders`][frequenz.channels.BroadcastReceiver.has_senders] to find out
+    whether new messages can still arrive.
+
+    Tip:
+        Because garbage-collected senders count as closed, keep a reference to every
+        sender that is still needed.  For example, after
+        `_, receiver = BroadcastChannel[int](name="numbers")`, the channel closes as
+        soon as `_` is reassigned or goes out of scope.  The receiver then stops once
+        it has received the messages already in its buffer.
 
     This channel is useful, for example, to implement a pub/sub pattern, where
     multiple consumers can subscribe to a channel to receive all messages.
@@ -829,12 +856,12 @@ class BroadcastChannel(
 
     Example: Send a few numbers to a receiver
         This is a very simple example that sends a few numbers from a single sender to
-        a single receiver.
+        a single receiver.  The receiver stops when the sender is closed.
 
         ```python
         import asyncio
 
-        from frequenz.channels import BroadcastChannel, Sender
+        from frequenz.channels import BroadcastChannel, Receiver, Sender
 
 
         async def send(sender: Sender[int]) -> None:
@@ -844,15 +871,18 @@ async def send(sender: Sender[int]) -> None:
             await sender.aclose()
 
 
+        async def receive(receiver: Receiver[int]) -> None:
+            async for message in receiver:
+                print(f"received {message}")
+            print("all senders are closed")
+
+
         async def main() -> None:
             sender, receiver = BroadcastChannel[int](name="numbers")
 
             async with asyncio.TaskGroup() as task_group:
                 task_group.create_task(send(sender))
-                for _ in range(3):
-                    message = await receiver.receive()
-                    print(f"received {message}")
-                    await asyncio.sleep(0.1)  # sleep (or work) with the data
+                task_group.create_task(receive(receiver))
 
 
         asyncio.run(main())
@@ -868,16 +898,17 @@ async def main() -> None:
         received 0
         received 1
         received 2
+        all senders are closed
         ```
 
-    Example: Send a few number from multiple senders to multiple receivers
+    Example: Send a few numbers from multiple senders to multiple receivers
         This is a more complex example that sends a few numbers from multiple senders to
-        multiple receivers, using a small buffer to force the senders to block.
+        multiple receivers.  The receivers stop when all senders are closed.
 
         ```python
         import asyncio
 
-        from frequenz.channels import BroadcastChannel, Receiver, ReceiverStoppedError, Sender
+        from frequenz.channels import BroadcastChannel, Receiver, Sender
 
 
         async def send(name: str, sender: Sender[int], start: int, stop: int) -> None:
@@ -887,30 +918,28 @@ async def send(name: str, sender: Sender[int], start: int, stop: int) -> None:
             await sender.aclose()
 
 
-        async def recv(name: str, receiver: Receiver[int]) -> None:
-            try:
-                async for message in receiver:
-                    print(f"{name} received {message}")
+        async def receive(name: str, receiver: Receiver[int]) -> None:
+            async for message in receiver:
+                print(f"{name} received {message}")
                 await asyncio.sleep(0.1)  # sleep (or work) with the data
-            except ReceiverStoppedError:
-                pass
 
 
         async def main() -> None:
             sender_1, receiver_1 = BroadcastChannel[int](name="numbers")
             sender_2 = sender_1.clone()
             receiver_2 = sender_1.subscribe()
+
             async with asyncio.TaskGroup() as task_group:
                 task_group.create_task(send("sender_1", sender_1, 10, 13))
                 task_group.create_task(send("sender_2", sender_2, 20, 22))
-                task_group.create_task(recv("receiver_1", receiver_1))
-                task_group.create_task(recv("receiver_2", receiver_2))
+                task_group.create_task(receive("receiver_1", receiver_1))
+                task_group.create_task(receive("receiver_2", receiver_2))
 
 
         asyncio.run(main())
         ```
 
-        The output should look something like this(although the sending and received
+        The output should look something like this (although the sending and received
         might appear interleaved in a different way):
 
         ```
@@ -920,16 +949,61 @@ async def main() -> None:
         sender_2 sending 20
         sender_2 sending 21
         receiver_1 received 10
-        receiver_1 received 11
-        receiver_1 received 12
-        receiver_1 received 20
-        receiver_1 received 21
         receiver_2 received 10
+        receiver_1 received 11
         receiver_2 received 11
+        receiver_1 received 12
         receiver_2 received 12
+        receiver_1 received 20
         receiver_2 received 20
+        receiver_1 received 21
         receiver_2 received 21
         ```
+
+    Example: Stop producing messages when nobody is listening
+        This example produces messages only for as long as there are receivers to
+        receive them.
+
+        ```python
+        import asyncio
+
+        from frequenz.channels import BroadcastChannel, BroadcastSender, Receiver
+
+
+        async def produce(sender: BroadcastSender[int]) -> None:
+            counter = 0
+            while sender.has_receivers:
+                await sender.send(counter)
+                counter += 1
+                await asyncio.sleep(0.1)  # wait for (or compute) the next message
+            print(f"no receivers left, stopping after {counter} messages")
+
+
+        async def consume(receiver: Receiver[int]) -> None:
+            for _ in range(3):
+                print(f"received {await receiver.receive()}")
+            receiver.close()
+
+
+        async def main() -> None:
+            sender, receiver = BroadcastChannel[int](name="numbers")
+
+            async with asyncio.TaskGroup() as task_group:
+                task_group.create_task(produce(sender))
+                task_group.create_task(consume(receiver))
+
+
+        asyncio.run(main())
+        ```
+
+        The output should be:
+
+        ```
+        received 0
+        received 1
+        received 2
+        no receivers left, stopping after 3 messages
+        ```
     """
 
     def __new__(
@@ -945,7 +1019,7 @@ def __new__(
             name: The name of the channel. This is for logging purposes, and it will be
                 shown in the string representation of the channel.
             resend_latest: When True, every time a new receiver is created with
-                [`Broadcast.new_receiver()`][...Broadcast.new_receiver],
+                [`BroadcastSender.subscribe()`][...BroadcastSender.subscribe],
                 the last message that was sent successfully will be sent to the new
                 receiver automatically, unless it was `None`.  This allows new
                 receivers on slow streams to get the latest message as soon as they

From 9e5fd97f363f2df91a7d1a6b014cd7e07beb1bed Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:40:23 +0000
Subject: [PATCH 11/15] Correct the broadcast docs on closing

The `aclose()` docs said that receiving after the buffer is drained
returns `None`, and the `BroadcastReceiver.close()` docs said that the
buffer can still be drained.  Both now say when a `ReceiverStoppedError`
is raised.  The receiver guide and `Receiver.close()` said every closed
receiver drains its buffer first; they now say this depends on the
receiver.

The multi-sender example of the deprecated `Broadcast` had the same
mistakes as the `BroadcastChannel` one, and it never closed its channel,
so it never ended.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_broadcast.py | 44 ++++++++++++++---------------
 src/frequenz/channels/_receiver.py  | 15 +++++-----
 2 files changed, 30 insertions(+), 29 deletions(-)

diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py
index 34f0580d..9f589c79 100644
--- a/src/frequenz/channels/_broadcast.py
+++ b/src/frequenz/channels/_broadcast.py
@@ -139,8 +139,8 @@ async def aclose(self) -> None:
 
         Receivers will still be able to drain the pending items on their queues,
         but after that, subsequent
-        [`receive()`][...Receiver.receive] calls will return `None`
-        immediately.
+        [`receive()`][...Receiver.receive] calls will raise a
+        [`ReceiverStoppedError`][...ReceiverStoppedError].
         """
         self._close()
 
@@ -360,14 +360,14 @@ async def main() -> None:
         received 2
         ```
 
-    Example: Send a few number from multiple senders to multiple receivers
+    Example: Send a few numbers from multiple senders to multiple receivers
         This is a more complex example that sends a few numbers from multiple senders to
-        multiple receivers, using a small buffer to force the senders to block.
+        multiple receivers.  The receivers stop when the channel is closed.
 
         ```python
         import asyncio
 
-        from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError, Sender
+        from frequenz.channels import Broadcast, Receiver, Sender
 
 
         async def send(name: str, sender: Sender[int], start: int, stop: int) -> None:
@@ -377,28 +377,28 @@ async def send(name: str, sender: Sender[int], start: int, stop: int) -> None:
 
 
         async def recv(name: str, receiver: Receiver[int]) -> None:
-            try:
-                async for message in receiver:
-                    print(f"{name} received {message}")
+            async for message in receiver:
+                print(f"{name} received {message}")
                 await asyncio.sleep(0.1)  # sleep (or work) with the data
-            except ReceiverStoppedError:
-                pass
 
 
         async def main() -> None:
             acast = Broadcast[int](name="numbers")
+            receiver_1 = acast.new_receiver()
+            receiver_2 = acast.new_receiver()
 
             async with asyncio.TaskGroup() as task_group:
-                task_group.create_task(send("sender_1", acast.new_sender(), 10, 13))
-                task_group.create_task(send("sender_2", acast.new_sender(), 20, 22))
-                task_group.create_task(recv("receiver_1", acast.new_receiver()))
-                task_group.create_task(recv("receiver_2", acast.new_receiver()))
+                task_group.create_task(recv("receiver_1", receiver_1))
+                task_group.create_task(recv("receiver_2", receiver_2))
+                await send("sender_1", acast.new_sender(), 10, 13)
+                await send("sender_2", acast.new_sender(), 20, 22)
+                await acast.aclose()
 
 
         asyncio.run(main())
         ```
 
-        The output should look something like this(although the sending and received
+        The output should look something like this (although the sending and received
         might appear interleaved in a different way):
 
         ```
@@ -408,14 +408,14 @@ async def main() -> None:
         sender_2 sending 20
         sender_2 sending 21
         receiver_1 received 10
-        receiver_1 received 11
-        receiver_1 received 12
-        receiver_1 received 20
-        receiver_1 received 21
         receiver_2 received 10
+        receiver_1 received 11
         receiver_2 received 11
+        receiver_1 received 12
         receiver_2 received 12
+        receiver_1 received 20
         receiver_2 received 20
+        receiver_1 received 21
         receiver_2 received 21
         ```
     """
@@ -747,9 +747,9 @@ def consume(self) -> _T:
     def close(self) -> None:
         """Close the receiver.
 
-        After calling this method, new messages will not be received.  Once the
-        receiver's buffer is drained, trying to receive a message will raise a
-        [`ReceiverStoppedError`][...ReceiverStoppedError].
+        After calling this method, trying to receive a message raises a
+        [`ReceiverStoppedError`][...ReceiverStoppedError], even if there are still
+        messages in the receiver's buffer.
         """
         self._closed = True
         self._detach()
diff --git a/src/frequenz/channels/_receiver.py b/src/frequenz/channels/_receiver.py
index ef1ea594..28f94df5 100644
--- a/src/frequenz/channels/_receiver.py
+++ b/src/frequenz/channels/_receiver.py
@@ -131,11 +131,10 @@
 [`Receiver.close()`][..Receiver.close] method can be used for this
 purpose.
 
-After [`Receiver.close()`][..Receiver.close] is called, we can still receive
-messages that are in the receiver's
-buffer, but as soon as the receiver's buffer has been drained, trying to receive further
-messages from a *closed* receiver will raise a
-[`ReceiverStoppedError`][..ReceiverStoppedError].
+After [`Receiver.close()`][..Receiver.close] is called, some receivers still
+deliver the messages in their buffer, and raise a
+[`ReceiverStoppedError`][..ReceiverStoppedError] once it is drained.  Others,
+like [`BroadcastReceiver`][..BroadcastReceiver], raise it at once.
 
 # Advanced Usage
 
@@ -235,8 +234,10 @@ def close(self) -> None:
         """Close the receiver.
 
         After calling this method, new messages will not be available from the receiver.
-        Once the receiver's buffer is drained, trying to receive a message will raise a
-        [`ReceiverStoppedError`][...ReceiverStoppedError].
+        Trying to receive a message raises a
+        [`ReceiverStoppedError`][...ReceiverStoppedError], either at once or once the
+        messages already in the receiver's buffer are received, depending on the
+        receiver.
         """
         raise NotImplementedError("close() must be implemented by subclasses")
 

From 490acca458775c26626dd8ad1677d3f660d64793 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:40:23 +0000
Subject: [PATCH 12/15] Show BroadcastChannel in the user guide and README

The user guide, package overview and README still presented the
deprecated `Broadcast` and `Anycast` classes.  They now show
`BroadcastChannel` and mark `Anycast` as deprecated, and a new page
covers `OneshotChannel`.  The README showcase now closes its data
senders, and its receiver stops when the data channels close.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 README.md                                     | 53 ++++++++-----------
 docs/user-guide/channels/anycast.md           |  6 +++
 docs/user-guide/channels/broadcast.md         |  2 +-
 docs/user-guide/channels/index.md             |  3 +-
 docs/user-guide/channels/oneshot.md           | 10 ++++
 .../receiving/synchronization/index.md        |  4 +-
 src/frequenz/channels/__init__.py             | 22 ++++++--
 src/frequenz/channels/_oneshot.py             |  4 +-
 8 files changed, 63 insertions(+), 41 deletions(-)
 create mode 100644 docs/user-guide/channels/oneshot.md

diff --git a/README.md b/README.md
index a08ffd64..10ee1cc7 100644
--- a/README.md
+++ b/README.md
@@ -67,13 +67,11 @@ python3 -m pip install frequenz-channels
 ```python
 import asyncio
 
-from frequenz.channels import Anycast
+from frequenz.channels import BroadcastChannel
 
 
 async def main() -> None:
-    hello_channel = Anycast[str](name="hello-world-channel")
-    sender = hello_channel.new_sender()
-    receiver = hello_channel.new_receiver()
+    sender, receiver = BroadcastChannel[str](name="hello-world-channel")
 
     await sender.send("Hello World!")
     message = await receiver.receive()
@@ -100,8 +98,7 @@ from enum import Enum, auto
 from typing import assert_never
 
 from frequenz.channels import (
-    Anycast,
-    Broadcast,
+    BroadcastChannel,
     Receiver,
     Sender,
     merge,
@@ -151,6 +148,7 @@ async def send(
                     await control_reply.send(Reply(ReplyCommand.PONG, str(sender)))
                 case _ as unknown:
                     assert_never(unknown)
+    await sender.aclose()
     print(f"{sender}: Finished")
 
 
@@ -159,13 +157,16 @@ async def receive(
     control_command: Receiver[Command],
     control_reply: Sender[Reply],
 ) -> None:
-    """Receive data from multiple channels, until no more data is received for 2 seconds."""
+    """Receive data from multiple channels, until they're closed or idle for 2 seconds."""
     print("receive: Starting")
     timer = Timer(timedelta(seconds=2.0), SkipMissedAndDrift())
     print(f"{timer=}")
     merged = merge(*receivers)
     async for selected in select(merged, timer, control_command):
         if selected_from(selected, merged):
+            if selected.was_stopped:
+                print("receive: All data channels were closed")
+                break
             message = selected.message
             print(f"receive: Received {message=}")
             timer.reset()
@@ -191,51 +192,41 @@ async def receive(
 
 
 async def main() -> None:
-    data_channel_1 = Anycast[str](name="data-channel-1")
-    data_channel_2 = Anycast[str](name="data-channel-2")
-    command_channel = Broadcast[Command](name="control-channel")  # (1)!
-    reply_channel = Anycast[Reply](name="reply-channel")
+    data_sender_1, data_receiver_1 = BroadcastChannel[str](name="data-channel-1")
+    data_sender_2, data_receiver_2 = BroadcastChannel[str](name="data-channel-2")
+    command_sender, command_receiver = BroadcastChannel[Command](
+        name="control-channel"
+    )
+    reply_sender, reply_receiver = BroadcastChannel[Reply](name="reply-channel")
 
     async with asyncio.TaskGroup() as tasks:
         tasks.create_task(
-            send(
-                data_channel_1.new_sender(),
-                command_channel.new_receiver(),
-                reply_channel.new_sender(),
-            ),
+            send(data_sender_1, command_receiver, reply_sender.clone()),
             name="send-channel-1",
         )
         tasks.create_task(
-            send(
-                data_channel_2.new_sender(),
-                command_channel.new_receiver(),
-                reply_channel.new_sender(),
-            ),
+            send(data_sender_2, command_sender.subscribe(), reply_sender.clone()),
             name="send-channel-2",
         )
         tasks.create_task(
             receive(
-                [data_channel_1.new_receiver(), data_channel_2.new_receiver()],
-                command_channel.new_receiver(),
-                reply_channel.new_sender(),
+                [data_receiver_1, data_receiver_2],
+                command_sender.subscribe(),
+                reply_sender,
             ),
             name="receive",
         )
 
-        control_sender = command_channel.new_sender()
-        reply_receiver = reply_channel.new_receiver()
-
         # Send a ping command to all tasks and wait for the replies
-        await control_sender.send(Command.PING)
+        await command_sender.send(Command.PING)
         print(f"main: {await reply_receiver.receive()}")
         print(f"main: {await reply_receiver.receive()}")
         print(f"main: {await reply_receiver.receive()}")
 
         await asyncio.sleep(5.0)
 
-        # Stop senders, after 2 seconds not receiving any data,
-        # the receiver will stop too
-        await control_sender.send(Command.STOP_SENDER)
+        # Stop the senders.  This closes the data channels, so the receiver stops too.
+        await command_sender.send(Command.STOP_SENDER)
 
 
 asyncio.run(main())
diff --git a/docs/user-guide/channels/anycast.md b/docs/user-guide/channels/anycast.md
index eca01889..07448cc8 100644
--- a/docs/user-guide/channels/anycast.md
+++ b/docs/user-guide/channels/anycast.md
@@ -1,5 +1,11 @@
 # Anycast
 
+!!! Warning "Deprecated"
+
+    Anycast channels are deprecated, because of the high cost of maintaining
+    something for which we don't have a use case.  The implementation will remain
+    in the codebase in its current form at least until the next major version.
+
 ::: frequenz.channels._anycast.Anycast
     options:
         inherited_members: []
diff --git a/docs/user-guide/channels/broadcast.md b/docs/user-guide/channels/broadcast.md
index f759bce3..ba8ebe2e 100644
--- a/docs/user-guide/channels/broadcast.md
+++ b/docs/user-guide/channels/broadcast.md
@@ -1,6 +1,6 @@
 # Broadcast
 
-::: frequenz.channels._broadcast.Broadcast
+::: frequenz.channels._broadcast.BroadcastChannel
     options:
         inherited_members: []
         members: []
diff --git a/docs/user-guide/channels/index.md b/docs/user-guide/channels/index.md
index c6f2df64..0142ba8d 100644
--- a/docs/user-guide/channels/index.md
+++ b/docs/user-guide/channels/index.md
@@ -35,7 +35,8 @@ various factors:
 Because these questions can have many answers, there are different types of
 channels. Frequenz Channels offers a few of them:
 
-* [Anycast](anycast.md)
 * [Broadcast](broadcast.md)
+* [Oneshot](oneshot.md)
+* [Anycast](anycast.md) (deprecated)
 
 More might be added in the future, and you can also create your own.
diff --git a/docs/user-guide/channels/oneshot.md b/docs/user-guide/channels/oneshot.md
new file mode 100644
index 00000000..9405ca72
--- /dev/null
+++ b/docs/user-guide/channels/oneshot.md
@@ -0,0 +1,10 @@
+# Oneshot
+
+::: frequenz.channels._oneshot.OneshotChannel
+    options:
+        inherited_members: []
+        members: []
+        show_bases: false
+        show_root_heading: false
+        show_root_toc_entry: false
+        show_source: false
diff --git a/docs/user-guide/receiving/synchronization/index.md b/docs/user-guide/receiving/synchronization/index.md
index 60d70d86..2c589daf 100644
--- a/docs/user-guide/receiving/synchronization/index.md
+++ b/docs/user-guide/receiving/synchronization/index.md
@@ -6,8 +6,8 @@ as soon as it is available in one single task. A naive approach like this will
 not work:
 
 ```python show_lines="4:"
-receiver1: Receiver[int] = channel1.new_receiver()
-receiver2: Receiver[int] = channel2.new_receiver()
+sender1, receiver1 = BroadcastChannel[int](name="channel1")
+sender2, receiver2 = BroadcastChannel[int](name="channel2")
 
 message = await receiver1.receive()
 print(f"Received from channel1: {message}")
diff --git a/src/frequenz/channels/__init__.py b/src/frequenz/channels/__init__.py
index b12ad24e..109be6ef 100644
--- a/src/frequenz/channels/__init__.py
+++ b/src/frequenz/channels/__init__.py
@@ -17,15 +17,29 @@
 * [`Sender`][.Sender]: An object that can send messages to
   a channel.
 
+* [`CloneableSender`][.CloneableSender],
+  [`SubscribableSender`][.SubscribableSender] and
+  [`CloneableSubscribableSender`][.CloneableSubscribableSender]: Senders that can
+  create more senders, or receivers, attached to the same channel.
+
 Channels:
 
-* [`Anycast`][.Anycast]: A channel that supports multiple
+* [`BroadcastChannel`][.BroadcastChannel]: A channel to broadcast messages
+  from multiple senders to multiple receivers. Each message sent through any of
+  the senders is received by all of the receivers.  The channel closes itself
+  when all its senders are closed or garbage-collected.  A `send()` while it has
+  no receivers raises a [`NoReceiversError`][.NoReceiversError].
+
+* [`OneshotChannel`][.OneshotChannel]: A channel that can send a single
+  message from one sender to one receiver.
+
+* [`Anycast`][.Anycast] (deprecated): A channel that supports multiple
   senders and multiple receivers.  A message sent through a sender will be
   received by exactly one receiver.
 
-* [`Broadcast`][.Broadcast]: A channel to broadcast messages
-  from multiple senders to multiple receivers. Each message sent through any of
-  the senders is received by all of the receivers.
+* [`Broadcast`][.Broadcast] (deprecated): The predecessor of
+  [`BroadcastChannel`][.BroadcastChannel].  Unless it is created with
+  `auto_close=True`, it must be closed with `aclose()`.
 
 Utilities to work with channels:
 
diff --git a/src/frequenz/channels/_oneshot.py b/src/frequenz/channels/_oneshot.py
index 3147d4ac..4d2bb9c4 100644
--- a/src/frequenz/channels/_oneshot.py
+++ b/src/frequenz/channels/_oneshot.py
@@ -25,7 +25,7 @@ class _Oneshot(typing.Generic[ChannelMessageT]):
 
     A one-shot channel is a channel that can only send one message. After the first
     message is sent, the sender is closed and any further attempts to send a message
-    will raise a [`SenderClosedError`][..SenderClosedError].
+    will raise a [`SenderClosedError`][frequenz.channels.SenderClosedError].
     """
 
     def __init__(self) -> None:
@@ -116,7 +116,7 @@ class OneshotChannel(
 
     A one-shot channel is a channel that can only send one message. After the first
     message is sent, the sender is closed and any further attempts to send a message
-    will raise a [`SenderClosedError`][..SenderClosedError].
+    will raise a [`SenderClosedError`][frequenz.channels.SenderClosedError].
 
     Example: Sending a message from one task to another
         ```python

From 43d1b9cd8bfce681f79023a2474f2dceee677f55 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:15:59 +0000
Subject: [PATCH 13/15] Use BroadcastChannel in docstring examples

The examples used the deprecated `Anycast` and `Broadcast` classes.
Also fix some examples with missing imports, and the latest value cache
examples, which checked the cache before its task could run.  The
`Anycast` docs now point to `BroadcastChannel`.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/_anycast.py             |  2 +-
 src/frequenz/channels/_exceptions.py          | 14 +++---
 src/frequenz/channels/_latest_value_cache.py  | 10 ++--
 src/frequenz/channels/_merge.py               | 24 ++++-----
 src/frequenz/channels/_receiver.py            | 49 +++++++++----------
 src/frequenz/channels/_select.py              | 40 ++++++---------
 src/frequenz/channels/_sender.py              | 22 ++++-----
 src/frequenz/channels/event.py                |  6 +--
 .../_grouping_latest_value_cache.py           | 10 ++--
 src/frequenz/channels/experimental/_pipe.py   | 25 +++++-----
 .../channels/experimental/_relay_sender.py    | 11 ++---
 .../channels/experimental/_with_previous.py   | 14 +++---
 src/frequenz/channels/timer.py                |  9 ++--
 13 files changed, 106 insertions(+), 130 deletions(-)

diff --git a/src/frequenz/channels/_anycast.py b/src/frequenz/channels/_anycast.py
index 7ccb809c..765b6ac1 100644
--- a/src/frequenz/channels/_anycast.py
+++ b/src/frequenz/channels/_anycast.py
@@ -93,7 +93,7 @@ class Anycast(Generic[ChannelMessageT]):
     This channel is useful, for example, to distribute work across multiple workers.
 
     In cases where each message needs to be received by every receiver, a
-    [broadcast][frequenz.channels.Broadcast] channel may be used.
+    [`BroadcastChannel`][frequenz.channels.BroadcastChannel] may be used.
 
     # Examples
 
diff --git a/src/frequenz/channels/_exceptions.py b/src/frequenz/channels/_exceptions.py
index b9e5f95d..ff88b05e 100644
--- a/src/frequenz/channels/_exceptions.py
+++ b/src/frequenz/channels/_exceptions.py
@@ -19,11 +19,10 @@
 was closed while sending or receiving a message, the original exception will be
 available as the cause of the exception:
 
-```python show_lines="6:"
-from frequenz.channels import Anycast, ChannelClosedError, SenderError
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel, ChannelClosedError, SenderError
 
-channel = Anycast[int](name="test-channel")
-sender = channel.new_sender()
+sender, receiver = BroadcastChannel[int](name="test-channel")
 
 try:
     await sender.send(42)
@@ -44,11 +43,10 @@
     explicitly calling [`receive()`][..Receiver.receive] on the
     receiver after the iteration is done:
 
-    ```python show_lines="6:"
-    from frequenz.channels import Anycast, ChannelClosedError, ReceiverStoppedError
+    ```python show_lines="5:"
+    from frequenz.channels import BroadcastChannel, ChannelClosedError, ReceiverStoppedError
 
-    channel = Anycast[int](name="test-channel")
-    receiver = channel.new_receiver()
+    sender, receiver = BroadcastChannel[int](name="test-channel")
 
     async for message in receiver:
         print(message)
diff --git a/src/frequenz/channels/_latest_value_cache.py b/src/frequenz/channels/_latest_value_cache.py
index 2530dbe4..32c4fe16 100644
--- a/src/frequenz/channels/_latest_value_cache.py
+++ b/src/frequenz/channels/_latest_value_cache.py
@@ -17,16 +17,18 @@
 
 Example:
     ```python
-    from frequenz.channels import Broadcast, LatestValueCache
+    import asyncio
 
-    channel = Broadcast[int](name="lvc_test")
+    from frequenz.channels import BroadcastChannel, LatestValueCache
 
-    cache = LatestValueCache(channel.new_receiver())
-    sender = channel.new_sender()
+    sender, receiver = BroadcastChannel[int](name="lvc_test")
+
+    cache = LatestValueCache(receiver)
 
     assert not cache.has_value()
 
     await sender.send(5)
+    await asyncio.sleep(0)  # Let the cache process the message
 
     assert cache.has_value()
     assert cache.get() == 5
diff --git a/src/frequenz/channels/_merge.py b/src/frequenz/channels/_merge.py
index b412d4ca..ce066171 100644
--- a/src/frequenz/channels/_merge.py
+++ b/src/frequenz/channels/_merge.py
@@ -9,21 +9,19 @@
 stream, you can use [`merge()`][.merge] to create a new receiver that
 will receive messages from all the given receivers:
 
-```python show_lines="8:"
-from frequenz.channels import Anycast, Receiver, merge
+```python show_lines="6:"
+from frequenz.channels import BroadcastChannel, Receiver, merge
 
-channel1: Anycast[int] = Anycast(name="channel1")
-channel2: Anycast[int] = Anycast(name="channel2")
-receiver1 = channel1.new_receiver()
-receiver2 = channel2.new_receiver()
+sender1, receiver1 = BroadcastChannel[int](name="channel1")
+sender2, receiver2 = BroadcastChannel[int](name="channel2")
 
 async for message in merge(receiver1, receiver2):
     print(message)
 ```
 
-If the first message comes from `channel2` and the second message from `channel1`, the
-first message will be received immediately, and the second message will be received as
-soon as it is available.
+If the first message comes from `receiver2` and the second message from `receiver1`,
+the first message will be received immediately, and the second message will be
+received as soon as it is available.
 
 This can be helpful when you just need to receive messages and don't care about
 where are they coming from specifically. If you need to know where the message came
@@ -69,12 +67,10 @@ def merge(*receivers: Receiver[ReceiverMessageT_co]) -> Merger[ReceiverMessageT_
         stream like this:
 
         ```python
-        from frequenz.channels import Broadcast
+        from frequenz.channels import BroadcastChannel, merge
 
-        channel1 = Broadcast[int](name="input-channel-1")
-        channel2 = Broadcast[int](name="input-channel-2")
-        receiver1 = channel1.new_receiver()
-        receiver2 = channel2.new_receiver()
+        sender1, receiver1 = BroadcastChannel[int](name="input-channel-1")
+        sender2, receiver2 = BroadcastChannel[int](name="input-channel-2")
 
         async for message in merge(receiver1, receiver2):
             print(f"received {message}")
diff --git a/src/frequenz/channels/_receiver.py b/src/frequenz/channels/_receiver.py
index 28f94df5..6cdc4a9f 100644
--- a/src/frequenz/channels/_receiver.py
+++ b/src/frequenz/channels/_receiver.py
@@ -7,15 +7,15 @@
 
 Messages are received from [channels](/user-guide/channels/index.md) through
 [`Receiver`][..Receiver] objects. [`Receiver`][..Receiver]s
-are usually created by calling `channel.new_receiver()` and are [async
-iterators][typing.AsyncIterator], so the easiest way to receive messages from them as
-a stream is to use `async for`:
+usually come from creating a channel, like a
+[`BroadcastChannel`][..BroadcastChannel], or from calling
+`sender.subscribe()`.  They are [async iterators][typing.AsyncIterator], so the
+easiest way to receive messages from them as a stream is to use `async for`:
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel
 
-channel = Anycast[int](name="test-channel")
-receiver = channel.new_receiver()
+sender, receiver = BroadcastChannel[int](name="test-channel")
 
 async for message in receiver:
     print(message)
@@ -24,11 +24,10 @@
 If you need to receive messages in different places or expecting a particular
 sequence, you can use the [`receive()`][..Receiver.receive] method:
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel
 
-channel = Anycast[int](name="test-channel")
-receiver = channel.new_receiver()
+sender, receiver = BroadcastChannel[int](name="test-channel")
 
 first_message = await receiver.receive()
 print(f"First message: {first_message}")
@@ -42,11 +41,10 @@
 If you need to transform the received messages, receivers provide a
 [`map()`][..Receiver.map] method to easily do so:
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel
 
-channel = Anycast[int](name="test-channel")
-receiver = channel.new_receiver()
+sender, receiver = BroadcastChannel[int](name="test-channel")
 
 async for message in receiver.map(lambda x: x + 1):
     print(message)
@@ -60,11 +58,10 @@
 If you need to filter the received messages, receivers provide a
 [`filter()`][..Receiver.filter] method to easily do so:
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel
 
-channel = Anycast[int](name="test-channel")
-receiver = channel.new_receiver()
+sender, receiver = BroadcastChannel[int](name="test-channel")
 
 async for message in receiver.filter(lambda x: x % 2 == 0):
     print(message)  # Only even numbers will be printed
@@ -90,11 +87,10 @@
 closed), a [`ReceiverStoppedError`][..ReceiverStoppedError] exception
 is raised by [`receive()`][..Receiver.receive] method.
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel, ReceiverError, ReceiverStoppedError
 
-channel = Anycast[int](name="test-channel")
-receiver = channel.new_receiver()
+sender, receiver = BroadcastChannel[int](name="test-channel")
 
 try:
     await receiver.receive()
@@ -107,11 +103,10 @@
 When used as an async iterator, the iteration will just stop without raising an
 exception:
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel, ReceiverError, ReceiverStoppedError
 
-channel = Anycast[int](name="test-channel")
-receiver = channel.new_receiver()
+sender, receiver = BroadcastChannel[int](name="test-channel")
 
 try:
     async for message in receiver:
diff --git a/src/frequenz/channels/_select.py b/src/frequenz/channels/_select.py
index d6130b5e..655c0145 100644
--- a/src/frequenz/channels/_select.py
+++ b/src/frequenz/channels/_select.py
@@ -18,13 +18,11 @@
 function to check the source of the message, and the
 [`message`][..Selected.message] attribute to access the message:
 
-```python show_lines="8:"
-from frequenz.channels import Anycast, ReceiverStoppedError, select, selected_from
+```python show_lines="6:"
+from frequenz.channels import BroadcastChannel, ReceiverStoppedError, select, selected_from
 
-channel1: Anycast[int] = Anycast(name="channel1")
-channel2: Anycast[str] = Anycast(name="channel2")
-receiver1 = channel1.new_receiver()
-receiver2 = channel2.new_receiver()
+sender1, receiver1 = BroadcastChannel[int](name="channel1")
+sender2, receiver2 = BroadcastChannel[str](name="channel2")
 
 async for selected in select(receiver1, receiver2):
     if selected_from(selected, receiver1):
@@ -46,13 +44,11 @@
     If for some reason you want to ignore a received message, just add the receiver to
     the if-chain and do nothing with the message:
 
-    ```python show_lines="8:"
-    from frequenz.channels import Anycast, select, selected_from
+    ```python show_lines="6:"
+    from frequenz.channels import BroadcastChannel, select, selected_from
 
-    channel1: Anycast[int] = Anycast(name="channel1")
-    channel2: Anycast[str] = Anycast(name="channel2")
-    receiver1 = channel1.new_receiver()
-    receiver2 = channel2.new_receiver()
+    sender1, receiver1 = BroadcastChannel[int](name="channel1")
+    sender2, receiver2 = BroadcastChannel[str](name="channel2")
 
     async for selected in select(receiver1, receiver2):
         if selected_from(selected, receiver1):
@@ -71,13 +67,11 @@
 [`was_stopped`][..Selected.was_stopped] method to check if the
 selected [`Receiver`][..Receiver] was stopped:
 
-```python show_lines="8:"
-from frequenz.channels import Anycast, select, selected_from
+```python show_lines="6:"
+from frequenz.channels import BroadcastChannel, select, selected_from
 
-channel1: Anycast[int] = Anycast(name="channel1")
-channel2: Anycast[str] = Anycast(name="channel2")
-receiver1 = channel1.new_receiver()
-receiver2 = channel2.new_receiver()
+sender1, receiver1 = BroadcastChannel[int](name="channel1")
+sender2, receiver2 = BroadcastChannel[str](name="channel2")
 
 async for selected in select(receiver1, receiver2):
     if selected_from(selected, receiver1):
@@ -106,13 +100,11 @@
 
 You can use a try-except block to handle exceptions as usual:
 
-```python show_lines="8:"
-from frequenz.channels import Anycast, ReceiverStoppedError, select, selected_from
+```python show_lines="6:"
+from frequenz.channels import BroadcastChannel, ReceiverStoppedError, select, selected_from
 
-channel1: Anycast[int] = Anycast(name="channel1")
-channel2: Anycast[str] = Anycast(name="channel2")
-receiver1 = channel1.new_receiver()
-receiver2 = channel2.new_receiver()
+sender1, receiver1 = BroadcastChannel[int](name="channel1")
+sender2, receiver2 = BroadcastChannel[str](name="channel2")
 
 async for selected in select(receiver1, receiver2):
     if selected_from(selected, receiver1):
diff --git a/src/frequenz/channels/_sender.py b/src/frequenz/channels/_sender.py
index 88cded0c..86ca72ad 100644
--- a/src/frequenz/channels/_sender.py
+++ b/src/frequenz/channels/_sender.py
@@ -6,15 +6,16 @@
 # Senders
 
 Messages are sent to a [channel](/user-guide/channels) through
-[`Sender`][..Sender] objects. [`Sender`][..Sender]s are
-usually created by calling `channel.new_sender()`, and are a very simple abstraction
-that only provides a single [`send()`][..Sender.send] method:
+[`Sender`][..Sender] objects. [`Sender`][..Sender]s
+usually come from creating a channel, like a
+[`BroadcastChannel`][..BroadcastChannel], or from calling
+`sender.clone()`.  They are a very simple abstraction that mainly provides a
+[`send()`][..Sender.send] method:
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel
 
-channel = Anycast[int](name="test-channel")
-sender = channel.new_sender()
+sender, receiver = BroadcastChannel[str](name="test-channel")
 
 await sender.send("Hello, world!")
 ```
@@ -36,11 +37,10 @@
 If there is any failure sending a message,
 a [`SenderError`][..SenderError] exception is raised.
 
-```python show_lines="6:"
-from frequenz.channels import Anycast
+```python show_lines="5:"
+from frequenz.channels import BroadcastChannel, SenderError
 
-channel = Anycast[int](name="test-channel")
-sender = channel.new_sender()
+sender, receiver = BroadcastChannel[str](name="test-channel")
 
 try:
     await sender.send("Hello, world!")
diff --git a/src/frequenz/channels/event.py b/src/frequenz/channels/event.py
index c40b4f6d..79767c21 100644
--- a/src/frequenz/channels/event.py
+++ b/src/frequenz/channels/event.py
@@ -46,12 +46,10 @@ class Event(Receiver[None]):
         ```python
         import asyncio
 
-        from frequenz.channels import Anycast, select, selected_from
+        from frequenz.channels import BroadcastChannel, select, selected_from
         from frequenz.channels.event import Event
 
-        channel: Anycast[int] = Anycast(name="channel")
-        receiver = channel.new_receiver()
-        sender = channel.new_sender()
+        sender, receiver = BroadcastChannel[int](name="channel")
         stop_event = Event(name="stop")
 
 
diff --git a/src/frequenz/channels/experimental/_grouping_latest_value_cache.py b/src/frequenz/channels/experimental/_grouping_latest_value_cache.py
index 96f59029..e6c2d12d 100644
--- a/src/frequenz/channels/experimental/_grouping_latest_value_cache.py
+++ b/src/frequenz/channels/experimental/_grouping_latest_value_cache.py
@@ -67,18 +67,20 @@ class GroupingLatestValueCache(Mapping[HashableT, ValueT_co]):
 
     Example:
         ```python
-        from frequenz.channels import Broadcast
+        import asyncio
+
+        from frequenz.channels import BroadcastChannel
         from frequenz.channels.experimental import GroupingLatestValueCache
 
-        channel = Broadcast[tuple[int, str]](name="lvc_test")
+        sender, receiver = BroadcastChannel[tuple[int, str]](name="lvc_test")
 
-        cache = GroupingLatestValueCache(channel.new_receiver(), key=lambda x: x[0])
-        sender = channel.new_sender()
+        cache = GroupingLatestValueCache(receiver, key=lambda x: x[0])
 
         assert cache.get(6) is None
         assert 6 not in cache
 
         await sender.send((6, "twenty-six"))
+        await asyncio.sleep(0)  # Let the cache process the message
 
         assert 6 in cache
         assert cache.get(6) == (6, "twenty-six")
diff --git a/src/frequenz/channels/experimental/_pipe.py b/src/frequenz/channels/experimental/_pipe.py
index b54f1917..120b2a04 100644
--- a/src/frequenz/channels/experimental/_pipe.py
+++ b/src/frequenz/channels/experimental/_pipe.py
@@ -28,30 +28,29 @@ class Pipe(typing.Generic[ChannelMessageT]):
     Example:
         ```python
         import asyncio
-        from contextlib import closing, aclosing, AsyncExitStack
+        from contextlib import closing, AsyncExitStack
 
-        from frequenz.channels import Broadcast
+        from frequenz.channels import BroadcastChannel
         from frequenz.channels.experimental import Pipe
 
         async def main() -> None:
-            # Channels, receivers and Pipe are in AsyncExitStack
+            # Senders, receivers and Pipe are in AsyncExitStack
             # to close and stop them at the end.
             async with AsyncExitStack() as stack:
-                source_channel = await stack.enter_async_context(
-                    aclosing(Broadcast[int](name="source channel"))
+                source_sender, source_receiver = BroadcastChannel[int](
+                    name="source channel"
                 )
-                source_receiver = stack.enter_context(closing(source_channel.new_receiver()))
+                stack.push_async_callback(source_sender.aclose)
+                stack.enter_context(closing(source_receiver))
 
-                forwarding_channel = await stack.enter_async_context(
-                    aclosing(Broadcast[int](name="forwarding channel"))
-                )
-                await stack.enter_async_context(
-                    Pipe(source_receiver, forwarding_channel.new_sender())
+                forwarding_sender, receiver = BroadcastChannel[int](
+                    name="forwarding channel"
                 )
+                stack.push_async_callback(forwarding_sender.aclose)
+                stack.enter_context(closing(receiver))
 
-                receiver = stack.enter_context(closing(forwarding_channel.new_receiver()))
+                await stack.enter_async_context(Pipe(source_receiver, forwarding_sender))
 
-                source_sender = source_channel.new_sender()
                 await source_sender.send(10)
                 assert await receiver.receive() == 10
 
diff --git a/src/frequenz/channels/experimental/_relay_sender.py b/src/frequenz/channels/experimental/_relay_sender.py
index 5960229d..c412b33c 100644
--- a/src/frequenz/channels/experimental/_relay_sender.py
+++ b/src/frequenz/channels/experimental/_relay_sender.py
@@ -24,16 +24,13 @@ class RelaySender(Sender[SenderMessageT_contra]):
 
     Example:
         ```python
-        from frequenz.channels import Broadcast
+        from frequenz.channels import BroadcastChannel
         from frequenz.channels.experimental import RelaySender
 
-        channel1: Broadcast[int] = Broadcast(name="channel1")
-        channel2: Broadcast[int] = Broadcast(name="channel2")
+        sender1, receiver1 = BroadcastChannel[int](name="channel1")
+        sender2, receiver2 = BroadcastChannel[int](name="channel2")
 
-        receiver1 = channel1.new_receiver()
-        receiver2 = channel2.new_receiver()
-
-        tee_sender = RelaySender(channel1.new_sender(), channel2.new_sender())
+        tee_sender = RelaySender(sender1, sender2)
 
         await tee_sender.send(5)
         assert await receiver1.receive() == 5
diff --git a/src/frequenz/channels/experimental/_with_previous.py b/src/frequenz/channels/experimental/_with_previous.py
index 9d7420f0..822df390 100644
--- a/src/frequenz/channels/experimental/_with_previous.py
+++ b/src/frequenz/channels/experimental/_with_previous.py
@@ -30,12 +30,11 @@ class WithPrevious(Generic[ChannelMessageT]):
 
     Example: Receiving only messages that are different from the previous one.
         ```python
-        from frequenz.channels import Broadcast
+        from frequenz.channels import BroadcastChannel
         from frequenz.channels.experimental import WithPrevious
 
-        channel = Broadcast[int](name="example")
-        receiver = channel.new_receiver().filter(WithPrevious(lambda old, new: old != new))
-        sender = channel.new_sender()
+        sender, receiver = BroadcastChannel[int](name="example")
+        receiver = receiver.filter(WithPrevious(lambda old, new: old != new))
 
         # This message will be received as it is the first message.
         await sender.send(1)
@@ -51,14 +50,13 @@ class WithPrevious(Generic[ChannelMessageT]):
 
     Example: Receiving only messages if they are bigger than the previous one.
         ```python
-        from frequenz.channels import Broadcast
+        from frequenz.channels import BroadcastChannel
         from frequenz.channels.experimental import WithPrevious
 
-        channel = Broadcast[int](name="example")
-        receiver = channel.new_receiver().filter(
+        sender, receiver = BroadcastChannel[int](name="example")
+        receiver = receiver.filter(
             WithPrevious(lambda old, new: new > old, first_is_true=False)
         )
-        sender = channel.new_sender()
 
         # This message will skipped as first_is_true is False.
         await sender.send(1)
diff --git a/src/frequenz/channels/timer.py b/src/frequenz/channels/timer.py
index 26c41958..37c958ec 100644
--- a/src/frequenz/channels/timer.py
+++ b/src/frequenz/channels/timer.py
@@ -21,7 +21,7 @@
     import asyncio
     from datetime import datetime, timedelta
 
-    from frequenz.channels.timer import Timer
+    from frequenz.channels.timer import Timer, TriggerAllMissed
 
 
     async def main() -> None:
@@ -46,13 +46,12 @@ async def main() -> None:
     import asyncio
     from datetime import timedelta
 
-    from frequenz.channels import Anycast, select, selected_from
-    from frequenz.channels.timer import Timer
+    from frequenz.channels import BroadcastChannel, select, selected_from
+    from frequenz.channels.timer import SkipMissedAndDrift, Timer
 
 
     async def main() -> None:
-        channel = Anycast[int](name="data-channel")
-        data_receiver = channel.new_receiver()
+        sender, data_receiver = BroadcastChannel[int](name="data-channel")
 
         timer = Timer(timedelta(seconds=1.0), SkipMissedAndDrift())
 

From f5a7cb45daf9c08ae6bb62dab12a864aa1a2dcc7 Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Wed, 21 Oct 2026 14:43:30 +0000
Subject: [PATCH 14/15] Close a pipe's ends when it stops on its own

A `Pipe` whose receiver stopped, or whose send failed, ended but kept
its sender open, so the receivers of the sender's channel could wait
forever.  Since a broadcast channel now stays open without receivers,
receivers subscribed after a failed send waited forever too.  And the
pipe's receiver stayed attached to its source channel, so
`has_receivers` stayed `True` there.

Close the sender in both cases, and the receiver too when the send
failed.  `stop()` still leaves both open.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 src/frequenz/channels/experimental/_pipe.py |  19 ++-
 tests/experimental/test_pipe.py             | 133 +++++++++++++++++++-
 2 files changed, 149 insertions(+), 3 deletions(-)

diff --git a/src/frequenz/channels/experimental/_pipe.py b/src/frequenz/channels/experimental/_pipe.py
index 120b2a04..7bdc4efa 100644
--- a/src/frequenz/channels/experimental/_pipe.py
+++ b/src/frequenz/channels/experimental/_pipe.py
@@ -15,7 +15,7 @@
 
 from .._generic import ChannelMessageT
 from .._receiver import Receiver
-from .._sender import Sender
+from .._sender import Sender, SenderError
 
 
 class Pipe(typing.Generic[ChannelMessageT]):
@@ -25,6 +25,13 @@ class Pipe(typing.Generic[ChannelMessageT]):
     creates a pipe between them
     by forwarding all the messages received by the receiver to the sender.
 
+    When its receiver stops, the pipe closes its sender and stops.  If sending fails
+    with a [`SenderError`][frequenz.channels.SenderError], for example because the
+    sender's channel has no receivers, the pipe closes both its sender and its
+    receiver, and stops.  [`stop()`][frequenz.channels.experimental.Pipe.stop]
+    leaves both open.  Give the pipe a clone of the sender if you still need to send
+    after the pipe stops on its own.
+
     Example:
         ```python
         import asyncio
@@ -101,4 +108,12 @@ async def stop(self) -> None:
 
     async def _run(self) -> None:
         async for value in self._receiver:
-            await self._sender.send(value)
+            try:
+                await self._sender.send(value)
+            except SenderError:
+                try:
+                    await self._sender.aclose()
+                finally:
+                    self._receiver.close()
+                return
+        await self._sender.aclose()
diff --git a/tests/experimental/test_pipe.py b/tests/experimental/test_pipe.py
index f0158e14..dc73e24c 100644
--- a/tests/experimental/test_pipe.py
+++ b/tests/experimental/test_pipe.py
@@ -7,7 +7,17 @@
 import typing
 from contextlib import AsyncExitStack, aclosing, closing
 
-from frequenz.channels import Broadcast, Receiver
+import pytest
+
+from frequenz.channels import (
+    Broadcast,
+    BroadcastChannel,
+    NoReceiversError,
+    Receiver,
+    Sender,
+    SenderClosedError,
+    SenderError,
+)
 from frequenz.channels.experimental import Pipe
 
 T = typing.TypeVar("T")
@@ -25,6 +35,13 @@ async def receive_timeout(recv: Receiver[T], timeout: float = 0.1) -> T | type[T
         return Timeout
 
 
+async def wait_stopped(pipe: Pipe[int]) -> None:
+    """Wait until the task of a started pipe ends."""
+    task = pipe._task  # pylint: disable=protected-access
+    assert task is not None
+    await asyncio.wait_for(task, timeout=1.0)
+
+
 async def test_pipe() -> None:
     """Test pipe."""
     channel1: Broadcast[int] = Broadcast(name="channel1")
@@ -79,3 +96,117 @@ async def test_pipe_with_async_exit_stack() -> None:
         source_sender = source_channel.new_sender()
         await source_sender.send(10)
         assert await receiver.receive() == 10
+
+
+async def test_pipe_closes_both_ends_when_sending_fails() -> None:
+    """Ensure a pipe closes its sender and receiver when it can't send."""
+    source_sender, source_receiver = BroadcastChannel[int](name="source")
+    sender, receiver = BroadcastChannel[int](name="forwarding")
+    receiver.close()
+
+    pipe = Pipe(source_receiver, sender)
+    await pipe.start()
+    with pytest.raises(NoReceiversError):
+        await sender.send(0)
+
+    await source_sender.send(1)
+    await wait_stopped(pipe)
+
+    assert not source_sender.has_receivers
+    with pytest.raises(SenderClosedError):
+        await sender.send(2)
+
+
+async def test_pipe_closes_sender_when_receiver_stops() -> None:
+    """Ensure a pipe closes its sender when its receiver stops."""
+    source_sender, source_receiver = BroadcastChannel[int](name="source")
+    sender, receiver = BroadcastChannel[int](name="forwarding")
+
+    pipe = Pipe(source_receiver, sender)
+    await pipe.start()
+    await source_sender.send(1)
+    assert await receive_timeout(receiver) == 1
+
+    await source_sender.aclose()
+    await wait_stopped(pipe)
+
+    with pytest.raises(SenderClosedError):
+        await sender.send(2)
+    assert not receiver.has_senders
+
+
+async def test_pipe_stop_leaves_both_ends_open() -> None:
+    """Ensure stopping a pipe leaves its sender and receiver open."""
+    source_sender, source_receiver = BroadcastChannel[int](name="source")
+    sender, receiver = BroadcastChannel[int](name="forwarding")
+
+    pipe = Pipe(source_receiver, sender)
+    await pipe.start()
+    await asyncio.sleep(0)
+    await pipe.stop()
+
+    await sender.send(1)
+    assert await receive_timeout(receiver) == 1
+    await source_sender.send(2)
+    assert await receive_timeout(source_receiver) == 2
+
+
+class UncloseableReceiver(Receiver[int]):
+    """A receiver whose `close()` fails."""
+
+    def __init__(self, receiver: Receiver[int]) -> None:
+        """Wrap the given receiver."""
+        self._receiver = receiver
+
+    async def ready(self) -> bool:
+        """Wait until the wrapped receiver is ready."""
+        return await self._receiver.ready()
+
+    def consume(self) -> int:
+        """Return the wrapped receiver's message."""
+        return self._receiver.consume()
+
+    def close(self) -> None:
+        """Fail to close."""
+        raise NotImplementedError("close() is not supported")
+
+
+async def test_pipe_closes_sender_when_receiver_cannot_close() -> None:
+    """Ensure a failed send closes the sender even if the receiver can't close."""
+    source_sender, source_receiver = BroadcastChannel[int](name="source")
+    sender, receiver = BroadcastChannel[int](name="forwarding")
+    receiver.close()
+
+    pipe = Pipe(UncloseableReceiver(source_receiver), sender)
+    await pipe.start()
+    await source_sender.send(1)
+    with pytest.raises(NotImplementedError):
+        await wait_stopped(pipe)
+
+    with pytest.raises(SenderClosedError):
+        await sender.send(2)
+
+
+class UncloseableSender(Sender[int]):
+    """A sender whose sends and `aclose()` fail."""
+
+    async def send(self, message: int, /) -> None:
+        """Fail to send."""
+        raise SenderError("Sending failed", self)
+
+    async def aclose(self) -> None:
+        """Fail to close."""
+        raise RuntimeError("Closing failed")
+
+
+async def test_pipe_closes_receiver_when_sender_cannot_close() -> None:
+    """Ensure a failed send closes the receiver even if the sender can't close."""
+    source_sender, source_receiver = BroadcastChannel[int](name="source")
+
+    pipe = Pipe(source_receiver, UncloseableSender())
+    await pipe.start()
+    await source_sender.send(1)
+    with pytest.raises(RuntimeError, match="Closing failed"):
+        await wait_stopped(pipe)
+
+    assert not source_sender.has_receivers

From e723c153bd8e22fdc4a55ddfa395be2ce5eb0b9f Mon Sep 17 00:00:00 2001
From: Sahas Subramanian <sahas.subramanian@proton.me>
Date: Mon, 19 Oct 2026 15:13:09 +0000
Subject: [PATCH 15/15] Update release notes

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
---
 RELEASE_NOTES.md | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
index 96a0240b..f332a36d 100644
--- a/RELEASE_NOTES.md
+++ b/RELEASE_NOTES.md
@@ -8,10 +8,38 @@
 
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
+- Garbage-collected broadcast senders now count as closed.  A `BroadcastChannel`, or a `Broadcast` created with `auto_close=True`, now closes as soon as its last sender is closed or garbage-collected, so keep a reference to every sender that is still needed.  On all broadcast channels, `sender_count` no longer counts garbage-collected senders.
+
+- A `BroadcastChannel`, or a `Broadcast` created with `auto_close=True`, no longer closes when a message is sent while it has no receivers.  `send()` now drops the message and raises a `NoReceiversError`, a `SenderError` without a cause.  The channel stays open: `subscribe()` keeps working, and sending works again once a new receiver is attached.
+
+- Code that waited for such a channel to close after a send without receivers, for example by checking for a `ChannelClosedError` cause or for `is_closed`, should catch `NoReceiversError` instead.  It can then stop sending, or close all of the channel's senders to close the channel.
+
+- An experimental `Pipe` now closes its sender when its receiver stops.  When sending fails with a `SenderError`, for example because the sender's `BroadcastChannel` has no receivers, it closes its receiver too.  Before, the pipe stopped but left both open.  `Pipe.stop()` still leaves both open.
+
+- Give a `Pipe` a clone of its sender if you still need to send after the pipe stops on its own.
+
+- For errors from a `BroadcastChannel` that are caused by a `ChannelClosedError`, `isinstance(error.__cause__.channel, Broadcast)` is now `False`.
+
+- `BroadcastSender.clone()` now raises a `SenderClosedError` if the sender is closed, and a `SenderError` if the channel is closed, like `send()` and `subscribe()` do.
+
+- `BroadcastSender.subscribe()` now uses the `limit` and `warn_on_overflow` values the `BroadcastChannel` was created with as defaults, instead of `50` and `True`.
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->
 
+- `NoReceiversError` is a new `SenderError`.  An open sender of a `BroadcastChannel`, or of a `Broadcast` created with `auto_close=True`, raises it from `send()` while the channel is open but has no receivers.  The message is dropped.  With `resend_latest=True`, new receivers get the last message that was sent successfully instead.
+
+- `BroadcastSender.has_receivers` and `BroadcastReceiver.has_senders` tell whether the other side of a broadcast channel is still there, so producers can stop producing messages that nobody would receive.
+
 ## Bug Fixes
 
 <!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+
+- Closing a broadcast receiver now wakes up tasks waiting for messages on it, instead of leaving them blocked until the next message arrived or the channel was closed.
+
+- A closed broadcast receiver is no longer reported as ready while messages are left in its buffer.  So `select()` reports it as stopped once instead of over and over, and no longer fails with a `SelectError` for a `filter()` on it.
+
+- Sending to or closing a broadcast channel no longer raises `RuntimeError: Event loop is closed` when a receiver was left waiting on an event loop that has since been closed.
+
+- Creating a `BroadcastChannel` no longer emits a deprecation warning telling users to use `BroadcastChannel` instead.