Add async_stream to create streams via generators

Author: taiki-e

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
   "futures",
+  "futures-async-macro",
   "futures-core",
   "futures-channel",
   "futures-executor",

futures-async-macro/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "futures-async-macro-preview"
+edition = "2018"
+version = "0.3.0-alpha.15"
+authors = ["Alex Crichton <[email protected]>"]
+license = "MIT OR Apache-2.0"
+repository = "https://github.com/rust-lang-nursery/futures-rs"
+homepage = "https://rust-lang-nursery.github.io/futures-rs"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.14/futures_async_macro"
+description = """
+Definition of the `#[async_stream]` macro for the `futures-rs` crate as well as a few other assorted macros.
+"""
+
+[lib]
+name = "futures_async_macro"
+proc-macro = true
+
+[features]
+std = []
+
+[dependencies]
+proc-macro2 = "0.4"
+quote = "0.6"
+syn = { version = "0.15.31", features = ["full", "fold"] }

futures-async-macro/LICENSE-APACHE

@@ -0,0 +1 @@
+../LICENSE-APACHE

futures-async-macro/LICENSE-MIT

@@ -0,0 +1 @@
+../LICENSE-MIT

futures-async-macro/README.md

@@ -0,0 +1,51 @@
+
+**Note that these are experimental APIs.**
+
+## Usage
+
+Add this to your `Cargo.toml`:
+
+```toml
+[dependencies]
+futures-preview = { version = "0.3.0-alpha.14", features = ["async-stream", "nightly"] }
+```
+
+### \#\[for_await\]
+
+Processes streams using a for loop.
+
+This is a reimplement of [futures-await]'s `#[async]` for loops for futures 0.3 and is an experimental implementation of [the idea listed as the next step of async/await](https://github.com/rust-lang/rfcs/blob/master/text/2394-async_await.md#for-await-and-processing-streams).
+
+```rust
+use futures::for_await;
+use futures::prelude::*;
+
+#[for_await]
+for value in stream::iter(1..=5) {
+    println!("{}", value);
+}
+```
+
+`value` has the `Item` type of the stream passed in. Note that async for loops can only be used inside of `async` functions or `#[async_stream]` functions.
+
+### \#\[async_stream\]
+
+This is a reimplement of [futures-await]'s `#[async_stream]` for futures 0.3 and is an experimental implementation of [the idea listed as the next step of async/await](https://github.com/rust-lang/rfcs/blob/master/text/2394-async_await.md#generators-and-streams).
+
+```rust
+use futures::prelude::*;
+use futures::{async_stream, stream_yield};
+
+// Returns a stream of i32
+#[async_stream]
+fn foo(stream: impl Stream<Item = String>) -> i32 {
+    #[for_await]
+    for x in stream {
+        stream_yield!(x.parse().unwrap());
+    }
+}
+```
+
+`#[async_stream]` have an item type specified via `-> some::Path` and the values output from the stream must be yielded via the `stream_yield!` macro.
+
+[futures-await]: https://github.com/alexcrichton/futures-await

futures-async-macro/src/elision.rs

@@ -0,0 +1,101 @@
+use proc_macro2::Span;
+use syn::fold::Fold;
+use syn::punctuated::Punctuated;
+use syn::token::Comma;
+use syn::{ArgSelfRef, FnArg, GenericParam, Lifetime, LifetimeDef, TypeReference};
+
+pub(super) fn unelide_lifetimes(
+    generics: &mut Punctuated<GenericParam, Comma>,
+    args: Vec<FnArg>,
+) -> Vec<FnArg> {
+    let mut folder = UnelideLifetimes::new(generics);
+    args.into_iter().map(|arg| folder.fold_fn_arg(arg)).collect()
+}
+
+struct UnelideLifetimes<'a> {
+    generics: &'a mut Punctuated<GenericParam, Comma>,
+    lifetime_index: usize,
+    lifetime_name: String,
+    count: u32,
+}
+
+impl<'a> UnelideLifetimes<'a> {
+    fn new(generics: &'a mut Punctuated<GenericParam, Comma>) -> UnelideLifetimes<'a> {
+        let lifetime_index = lifetime_index(generics);
+        let lifetime_name = lifetime_name(generics);
+        UnelideLifetimes { generics, lifetime_index, lifetime_name, count: 0 }
+    }
+
+    // Constitute a new lifetime
+    fn new_lifetime(&mut self) -> Lifetime {
+        let lifetime_name = format!("{}{}", self.lifetime_name, self.count);
+        let lifetime = Lifetime::new(&lifetime_name, Span::call_site());
+
+        let idx = self.lifetime_index + self.count as usize;
+        self.generics.insert(idx, GenericParam::Lifetime(LifetimeDef::new(lifetime.clone())));
+        self.count += 1;
+
+        lifetime
+    }
+
+    // Take an Option<Lifetime> and guarantee its an unelided lifetime
+    fn expand_lifetime(&mut self, lifetime: Option<Lifetime>) -> Lifetime {
+        match lifetime {
+            Some(l) => self.fold_lifetime(l),
+            None => self.new_lifetime(),
+        }
+    }
+}
+
+impl Fold for UnelideLifetimes<'_> {
+    // Handling self arguments
+    fn fold_arg_self_ref(&mut self, arg: ArgSelfRef) -> ArgSelfRef {
+        let ArgSelfRef { and_token, lifetime, mutability, self_token } = arg;
+        let lifetime = Some(self.expand_lifetime(lifetime));
+        ArgSelfRef { and_token, lifetime, mutability, self_token }
+    }
+
+    // If the lifetime is `'_`, replace it with a new unelided lifetime
+    fn fold_lifetime(&mut self, lifetime: Lifetime) -> Lifetime {
+        if lifetime.ident == "_" {
+            self.new_lifetime()
+        } else {
+            lifetime
+        }
+    }
+
+    // If the reference's lifetime is elided, replace it with a new unelided lifetime
+    fn fold_type_reference(&mut self, ty_ref: TypeReference) -> TypeReference {
+        let TypeReference { and_token, lifetime, mutability, elem } = ty_ref;
+        let lifetime = Some(self.expand_lifetime(lifetime));
+        let elem = Box::new(self.fold_type(*elem));
+        TypeReference { and_token, lifetime, mutability, elem }
+    }
+}
+
+fn lifetime_index(generics: &Punctuated<GenericParam, Comma>) -> usize {
+    generics
+        .iter()
+        .take_while(|param| if let GenericParam::Lifetime(_) = param { true } else { false })
+        .count()
+}
+
+// Determine the prefix for all lifetime names. Ensure it doesn't
+// overlap with any existing lifetime names.
+fn lifetime_name(generics: &Punctuated<GenericParam, Comma>) -> String {
+    let mut lifetime_name = String::from("'_async");
+    let existing_lifetimes: Vec<String> = generics
+        .iter()
+        .filter_map(|param| {
+            if let GenericParam::Lifetime(LifetimeDef { lifetime, .. }) = param {
+                Some(lifetime.to_string())
+            } else {
+                None
+            }
+        })
+        .collect();
+    while existing_lifetimes.iter().any(|name| name.starts_with(&lifetime_name)) {
+        lifetime_name.push('_');
+    }
+    lifetime_name
+}

futures-async-macro/src/error.rs

@@ -0,0 +1,52 @@
+// Make error messages a little more readable than `panic!`.
+macro_rules! error {
+    ($msg:expr) => {
+        return proc_macro::TokenStream::from(
+            syn::Error::new(proc_macro2::Span::call_site(), $msg).to_compile_error(),
+        )
+    };
+}
+
+// TODO: Should we give another name?
+// `assert!` that call `error!` instead of `panic!`.
+macro_rules! assert_ {
+    ($e:expr, $msg:expr) => {
+        if !$e {
+            error!($msg)
+        }
+    };
+}
+
+pub(super) fn expr_compile_error(e: &syn::Error) -> syn::Expr {
+    syn::parse2(e.to_compile_error()).unwrap()
+}
+
+// Long error messages and error messages that are called multiple times
+
+macro_rules! args_is_not_empty {
+    ($name:expr) => {
+        concat!("attribute must be of the form `#[", $name, "]`")
+    };
+}
+
+macro_rules! outside_of_async_error {
+    ($tokens:expr, $name:expr) => {
+        $crate::error::expr_compile_error(&syn::Error::new_spanned(
+            $tokens,
+            concat!(
+                $name,
+                " cannot be allowed outside of \
+                 async closures, blocks, functions, async_stream blocks, and functions."
+            ),
+        ))
+    };
+}
+
+macro_rules! outside_of_async_stream_error {
+    ($tokens:expr, $name:expr) => {
+        $crate::error::expr_compile_error(&syn::Error::new_spanned(
+            $tokens,
+            concat!($name, " cannot be allowed outside of async_stream blocks, and functions."),
+        ))
+    };
+}