sync: close the `broadcast::Sender` in `broadcast::Sender::new()` (#7629)

(cherry picked from commit 6d1ae6286880c828c13efb5f11b60c18fb94f947)

diff --git a/tokio/src/sync/broadcast.rs b/tokio/src/sync/broadcast.rs
index b48493be26cf14250601c32cb46d8996f40a8fdc..a69eca661eade75fc0ceb6d40bf61b9ab294002f 100644 (file)
--- a/tokio/src/sync/broadcast.rs
+++ b/tokio/src/sync/broadcast.rs
@@ -522,7 +522,7 @@ impl<T> Sender<T> {
             tail: Mutex::new(Tail {
                 pos: 0,
                 rx_cnt: receiver_count,
-                closed: false,
+                closed: receiver_count == 0,
                 waiters: LinkedList::new(),
             }),
             num_tx: AtomicUsize::new(1),

diff --git a/tokio/tests/sync_broadcast.rs b/tokio/tests/sync_broadcast.rs
index 3af96bdb5d5b578d1104732cf494ffc547dc075d..d485c63076fc5e19456b97af51fe2aa99ced524f 100644 (file)
--- a/tokio/tests/sync_broadcast.rs
+++ b/tokio/tests/sync_broadcast.rs
@@ -656,3 +656,15 @@ async fn receiver_recv_is_cooperative() {
         _ = tokio::task::yield_now() => {},
     }
 }
+
+#[tokio::test]
+async fn broadcast_sender_new_must_be_closed() {
+    let capacity = 1;
+    let tx: broadcast::Sender<()> = broadcast::Sender::new(capacity);
+
+    tx.send(()).unwrap_err();
+
+    let _rx = tx.subscribe();
+
+    tx.send(()).unwrap();
+}