From 909d3ec0ff205cc5be1e31b8c3c22a4f0916bb47 Mon Sep 17 00:00:00 2001
From: Francis Murillo <francis.murillo@protonmail.com>
Date: Sun, 29 Aug 2021 22:30:13 +0800
Subject: [PATCH] stream: add `From<Receiver<T>>` impl for receiver streams
 (#4080)

---
 tokio-stream/src/wrappers/broadcast.rs      | 6 ++++++
 tokio-stream/src/wrappers/mpsc_bounded.rs   | 6 ++++++
 tokio-stream/src/wrappers/mpsc_unbounded.rs | 6 ++++++
 tokio-stream/src/wrappers/watch.rs          | 8 +++++++-
 4 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/tokio-stream/src/wrappers/broadcast.rs b/tokio-stream/src/wrappers/broadcast.rs
index 3bddbb741..c8346a68e 100644
--- a/tokio-stream/src/wrappers/broadcast.rs
+++ b/tokio-stream/src/wrappers/broadcast.rs
@@ -71,3 +71,9 @@ impl<T> fmt::Debug for BroadcastStream<T> {
         f.debug_struct("BroadcastStream").finish()
     }
 }
+
+impl<T: 'static + Clone + Send> From<Receiver<T>> for BroadcastStream<T> {
+    fn from(recv: Receiver<T>) -> Self {
+        Self::new(recv)
+    }
+}
diff --git a/tokio-stream/src/wrappers/mpsc_bounded.rs b/tokio-stream/src/wrappers/mpsc_bounded.rs
index e4f900009..b5362680e 100644
--- a/tokio-stream/src/wrappers/mpsc_bounded.rs
+++ b/tokio-stream/src/wrappers/mpsc_bounded.rs
@@ -57,3 +57,9 @@ impl<T> AsMut<Receiver<T>> for ReceiverStream<T> {
         &mut self.inner
     }
 }
+
+impl<T> From<Receiver<T>> for ReceiverStream<T> {
+    fn from(recv: Receiver<T>) -> Self {
+        Self::new(recv)
+    }
+}
diff --git a/tokio-stream/src/wrappers/mpsc_unbounded.rs b/tokio-stream/src/wrappers/mpsc_unbounded.rs
index bc5f40cdc..54597b7f6 100644
--- a/tokio-stream/src/wrappers/mpsc_unbounded.rs
+++ b/tokio-stream/src/wrappers/mpsc_unbounded.rs
@@ -51,3 +51,9 @@ impl<T> AsMut<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
         &mut self.inner
     }
 }
+
+impl<T> From<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
+    fn from(recv: UnboundedReceiver<T>) -> Self {
+        Self::new(recv)
+    }
+}
diff --git a/tokio-stream/src/wrappers/watch.rs b/tokio-stream/src/wrappers/watch.rs
index 1daca1010..bd3a18a58 100644
--- a/tokio-stream/src/wrappers/watch.rs
+++ b/tokio-stream/src/wrappers/watch.rs
@@ -59,7 +59,7 @@ async fn make_future<T: Clone + Send + Sync>(
     (result, rx)
 }
 
-impl<T: 'static + Clone + Unpin + Send + Sync> WatchStream<T> {
+impl<T: 'static + Clone + Send + Sync> WatchStream<T> {
     /// Create a new `WatchStream`.
     pub fn new(rx: Receiver<T>) -> Self {
         Self {
@@ -94,3 +94,9 @@ impl<T> fmt::Debug for WatchStream<T> {
         f.debug_struct("WatchStream").finish()
     }
 }
+
+impl<T: 'static + Clone + Send + Sync> From<Receiver<T>> for WatchStream<T> {
+    fn from(recv: Receiver<T>) -> Self {
+        Self::new(recv)
+    }
+}