improve lifetimes

Author: carllerche

async-stream/Cargo.toml

@@ -11,4 +11,5 @@ proc-macro-hack = "0.5.8"
 
 [dev-dependencies]
 tokio = "=0.2.0-alpha.1"
+tokio-test = "=0.2.0-alpha.1"
 futures-util-preview = "=0.3.0-alpha.17"

async-stream/src/yielder.rs

@@ -30,7 +30,7 @@ thread_local!(static STORE: Cell<*mut ()> = Cell::new(ptr::null_mut()));
 
 // ===== impl Sender =====
 
-impl<T: Unpin + 'static> Sender<T> {
+impl<T: Unpin> Sender<T> {
     pub fn send(&mut self, value: T) -> impl Future<Output = ()> {
         Send { value: Some(value) }
     }
@@ -40,7 +40,7 @@ struct Send<T> {
     value: Option<T>,
 }
 
-impl<T: Unpin + 'static> Future for Send<T> {
+impl<T: Unpin> Future for Send<T> {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {

async-stream/tests/basic.rs

@@ -2,8 +2,10 @@
 
 use async_stream::stream;
 
-use tokio::prelude::*;
 use futures_util::pin_mut;
+use tokio::prelude::*;
+use tokio::sync::mpsc;
+use tokio_test::assert_ok;
 
 #[tokio::test]
 async fn noop_stream() {
@@ -81,3 +83,43 @@ async fn return_stream() {
     assert_eq!(2, values[1]);
     assert_eq!(3, values[2]);
 }
+
+#[tokio::test]
+async fn consume_channel() {
+    let (mut tx, mut rx) = mpsc::channel(10);
+
+    let s = stream! {
+        while let Some(v) = rx.recv().await {
+            yield v;
+        }
+    };
+
+    pin_mut!(s);
+
+    for i in 0..3 {
+        assert_ok!(tx.send(i).await);
+        assert_eq!(Some(i), s.next().await);
+    }
+
+    drop(tx);
+    assert_eq!(None, s.next().await);
+}
+
+#[tokio::test]
+async fn borrow_self() {
+    struct Data(String);
+
+    impl Data {
+        fn stream<'a>(&'a self) -> impl Stream<Item = &str> + 'a {
+            stream! {
+                yield &self.0[..];
+            }
+        }
+    }
+
+    let data = Data("hello".to_string());
+    let s = data.stream();
+    pin_mut!(s);
+
+    assert_eq!(Some("hello"), s.next().await);
+}