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

(cherry picked from commit 6d1ae62868)

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),

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();
+}