Adding upstream version 110.0.1.upstream/110.0.1upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 29 insertions, 0 deletions

diff --git a/third_party/rust/tokio-stream/tests/watch.rs b/third_party/rust/tokio-stream/tests/watch.rs
new file mode 100644
index 0000000000..a56254edef
--- /dev/null
+++ b/third_party/rust/tokio-stream/tests/watch.rs
@@ -0,0 +1,29 @@
+#![cfg(feature = "sync")]
+
+use tokio::sync::watch;
+use tokio_stream::wrappers::WatchStream;
+use tokio_stream::StreamExt;
+
+#[tokio::test]
+async fn message_not_twice() {
+    let (tx, rx) = watch::channel("hello");
+
+    let mut counter = 0;
+    let mut stream = WatchStream::new(rx).map(move |payload| {
+        println!("{}", payload);
+        if payload == "goodbye" {
+            counter += 1;
+        }
+        if counter >= 2 {
+            panic!("too many goodbyes");
+        }
+    });
+
+    let task = tokio::spawn(async move { while stream.next().await.is_some() {} });
+
+    // Send goodbye just once
+    tx.send("goodbye").unwrap();
+
+    drop(tx);
+    task.await.unwrap();
+}