get tests passing

Author: carllerche

async-stream-impl/src/lib.rs

@@ -8,12 +8,15 @@ use syn::parse::{Parse, ParseStream, Result};
 use syn::visit_mut::VisitMut;
 
 struct AsyncStreamImpl {
+    yielder: syn::Ident,
     stmts: Vec<syn::Stmt>,
+    num_yield: u32,
 }
 
 struct Scrub {
     yielder: syn::Ident,
     unit: Box<syn::Expr>,
+    num_yield: u32,
 }
 
 impl Parse for AsyncStreamImpl {
@@ -25,6 +28,7 @@ impl Parse for AsyncStreamImpl {
         let mut scrub = Scrub {
             yielder,
             unit: syn::parse_quote!(()),
+            num_yield: 0,
         };
 
         while !input.is_empty() {
@@ -33,14 +37,22 @@ impl Parse for AsyncStreamImpl {
             stmts.push(stmt);
         }
 
-        Ok(AsyncStreamImpl { stmts })
+        let Scrub { yielder, num_yield, .. } = scrub;
+
+        Ok(AsyncStreamImpl {
+            yielder,
+            stmts,
+            num_yield,
+        })
     }
 }
 
 impl VisitMut for Scrub {
     fn visit_expr_mut(&mut self, i: &mut syn::Expr) {
         match i {
             syn::Expr::Yield(expr) => {
+                self.num_yield += 1;
+
                 let value_expr = if let Some(ref e) = expr.expr {
                     e
                 } else {
@@ -58,9 +70,22 @@ impl VisitMut for Scrub {
 #[proc_macro_hack]
 pub fn async_stream_impl(input: TokenStream) -> TokenStream {
     let AsyncStreamImpl {
+        yielder,
         stmts,
+        num_yield,
     } = syn::parse_macro_input!(input as AsyncStreamImpl);
 
-    quote!(#(#stmts)*).into()
-    // quote!(#input).into()
+    if num_yield == 0 {
+        quote!({
+            if false {
+                #yielder.send(()).await;
+            }
+
+            #(#stmts)*
+        }).into()
+    } else {
+        quote!({
+            #(#stmts)*
+        }).into()
+    }
 }

async-stream/src/yielder.rs

@@ -43,7 +43,7 @@ struct Send<T> {
 impl<T: Unpin + 'static> Future for Send<T> {
     type Output = ();
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {
         if self.value.is_none() {
             return Poll::Ready(());
         }

async-stream/tests/basic.rs

@@ -1,6 +1,83 @@
+#![feature(async_await)]
+
 use async_stream::stream;
 
-#[test]
-fn smoke() {
-    stream!();
+use tokio::prelude::*;
+use futures_util::pin_mut;
+
+#[tokio::test]
+async fn noop_stream() {
+    let s = stream! {};
+    pin_mut!(s);
+
+    while let Some(_) = s.next().await {
+        unreachable!();
+    }
+}
+
+#[tokio::test]
+async fn empty_stream() {
+    let mut ran = false;
+
+    {
+        let r = &mut ran;
+        let s = stream! {
+            *r = true;
+            println!("hello world!");
+        };
+        pin_mut!(s);
+
+        while let Some(_) = s.next().await {
+            unreachable!();
+        }
+    }
+
+    assert!(ran);
+}
+
+#[tokio::test]
+async fn yield_single_value() {
+    let s = stream! {
+        yield "hello";
+    };
+
+    let values: Vec<_> = s.collect().await;
+
+    assert_eq!(1, values.len());
+    assert_eq!("hello", values[0]);
+}
+
+#[tokio::test]
+async fn yield_multi_value() {
+    let s = stream! {
+        yield "hello";
+        yield "world";
+        yield "dizzy";
+    };
+
+    let values: Vec<_> = s.collect().await;
+
+    assert_eq!(3, values.len());
+    assert_eq!("hello", values[0]);
+    assert_eq!("world", values[1]);
+    assert_eq!("dizzy", values[2]);
+}
+
+#[tokio::test]
+async fn return_stream() {
+    fn build_stream() -> impl Stream<Item = u32> {
+        stream! {
+            yield 1;
+            yield 2;
+            yield 3;
+        }
+    }
+
+    let s = build_stream();
+
+    let values: Vec<_> = s.collect().await;
+    assert_eq!(3, values.len());
+    assert_eq!(1, values[0]);
+    assert_eq!(2, values[1]);
+    assert_eq!(3, values[2]);
 }