From 97de03edffe6337399bd4172072e4932b70b91d1 Mon Sep 17 00:00:00 2001
From: Evan Simmons <evan.simmons@prevoty.com>
Date: Fri, 8 Apr 2022 11:51:57 -0700
Subject: [PATCH] Add subscribe method to broadcast::Receiver

---
 tokio/src/sync/broadcast.rs | 15 +++++++++++----
 1 file changed, 11 insertions(+), 4 deletions(-)

diff --git a/tokio/src/sync/broadcast.rs b/tokio/src/sync/broadcast.rs
index e54fe5c8151..e34b64a906d 100644
--- a/tokio/src/sync/broadcast.rs
+++ b/tokio/src/sync/broadcast.rs
@@ -552,7 +552,7 @@ impl<T> Sender<T> {
     /// ```
     pub fn subscribe(&self) -> Receiver<T> {
         let shared = self.shared.clone();
-        new_receiver(shared)
+        new_receiver(shared, None)
     }
 
     /// Returns the number of active receivers
@@ -642,7 +642,8 @@ impl<T> Sender<T> {
     }
 }
 
-fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
+/// Create a new `Receiver` which reads starting from the tail if `next_pos` is not specified.
+fn new_receiver<T>(shared: Arc<Shared<T>>, next_pos: Option<u64>) -> Receiver<T> {
     let mut tail = shared.tail.lock();
 
     if tail.rx_cnt == MAX_RECEIVERS {
@@ -651,10 +652,9 @@ fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
 
     tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");
 
-    let next = tail.pos;
+    let next = next_pos.unwrap_or(tail.pos);
 
     drop(tail);
-
     Receiver { shared, next }
 }
 
@@ -1022,6 +1022,13 @@ impl<T> Drop for Receiver<T> {
     }
 }
 
+impl<T> Clone for Receiver<T> {
+    fn clone(&self) -> Self {
+        let shared = self.shared.clone();
+        new_receiver(shared, Some(self.next))
+    }
+}
+
 impl<'a, T> Recv<'a, T> {
     fn new(receiver: &'a mut Receiver<T>) -> Recv<'a, T> {
         Recv {