Add async_stream to create streams via generators

.travis.yml

@@ -97,6 +97,13 @@ matrix:
         - cargo build --manifest-path futures-channel/Cargo.toml --no-default-features --features alloc
         - cargo build --manifest-path futures-util/Cargo.toml --no-default-features --features alloc
 
+    - name: cargo test (async-stream feature)
+      rust: nightly
+      script:
+        - cargo test --manifest-path futures/Cargo.toml --features async-stream,nightly --test async_stream_tests
+        - cargo clean --manifest-path futures/testcrate/Cargo.toml
+        - cargo test --manifest-path futures/testcrate/Cargo.toml
+
     - name: cargo build --target=thumbv6m-none-eabi
       rust: nightly
       install:

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.17"
+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.34", 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,55 @@
+
+**Note that these are experimental APIs.**
+
+## Usage
+
+Add this to your `Cargo.toml`:
+
+```toml
+[dependencies]
+futures-preview = { version = "=0.3.0-alpha.16", 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
+#![feature(async_await, stmt_expr_attributes, proc_macro_hygiene)]
+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, closures, blocks, `#[async_stream]` functions and `async_stream_block!` macros.
+
+### \#\[async_stream\]
+
+Creates streams via generators.
+
+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
+#![feature(async_await, generators)]
+use futures::prelude::*;
+use futures::async_stream;
+
+// Returns a stream of i32
+#[async_stream(item = i32)]
+fn foo(stream: impl Stream<Item = String>) {
+    #[for_await]
+    for x in stream {
+        yield x.parse().unwrap();
+    }
+}
+```
+
+`#[async_stream]` must have an item type specified via `item = some::Path` and the values output from the stream must be yielded via the `yield` expression.
+
+[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,48 @@
+// 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(),
+        )
+    };
+    ($span:expr, $msg:expr) => {
+        return proc_macro::TokenStream::from(
+            syn::Error::new_spanned($span, $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."
+            ),
+        ))
+    };
+}

futures-select-macro/Cargo.toml

@@ -22,4 +22,4 @@ std = []
 proc-macro2 = "0.4"
 proc-macro-hack = "0.5.3"
 quote = "0.6"
-syn = { version = "0.15.25", features = ["full"] }
+syn = { version = "0.15.34", features = ["full"] }

futures-util/Cargo.toml

@@ -28,6 +28,7 @@ sink = ["futures-sink-preview"]
 io = ["std", "futures-io-preview", "memchr"]
 channel = ["std", "futures-channel-preview"]
 select-macro = ["async-await", "futures-select-macro-preview", "proc-macro-hack", "proc-macro-nested", "rand"]
+async-stream = ["std", "async-await"]
 
 [dependencies]
 futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.17", default-features = false }

futures-util/src/async_stream/mod.rs

@@ -0,0 +1,72 @@
+//! Await
+//!
+//! This module contains a number of functions and combinators for working
+//! with `async`/`await` code.
+
+use futures_core::future::Future;
+use futures_core::stream::Stream;
+use std::future::{self, get_task_context, set_task_context};
+use std::marker::PhantomData;
+use std::ops::{Generator, GeneratorState};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Wrap a generator in a stream.
+///
+/// This function returns a `GenStream` underneath, but hides it in `impl Trait` to give
+/// better error messages (`impl Stream` rather than `GenStream<[closure.....]>`).
+pub fn from_generator<U, T>(x: T) -> impl Stream<Item = U>
+where
+    T: Generator<Yield = Poll<U>, Return = ()>,
+{
+    GenStream { gen: x, done: false, _phantom: PhantomData }
+}
+
+/// A wrapper around generators used to implement `Stream` for `async`/`await` code.
+#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
+struct GenStream<U, T: Generator<Yield = Poll<U>, Return = ()>> {
+    gen: T,
+    // If resuming after Complete, std generators panic. This is natural when using the generators,
+    // but in the streams, we may want to call `poll_next` after `poll_next` returns `None`
+    // because it is possible to call `next` after `next` returns `None` in many iterators.
+    done: bool,
+    _phantom: PhantomData<U>,
+}
+
+impl<U, T: Generator<Yield = Poll<U>, Return = ()>> Stream for GenStream<U, T> {
+    type Item = U;
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if self.done {
+            return Poll::Ready(None);
+        }
+        // Safe because we're !Unpin + !Drop mapping to a ?Unpin value
+        let Self { gen, done, .. } = unsafe { Pin::get_unchecked_mut(self) };
+        let gen = unsafe { Pin::new_unchecked(gen) };
+        set_task_context(cx, || match gen.resume() {
+            GeneratorState::Yielded(x) => x.map(Some),
+            GeneratorState::Complete(()) => {
+                *done = true;
+                Poll::Ready(None)
+            }
+        })
+    }
+}
+
+/// Polls a stream in the current thread-local task waker.
+pub fn poll_next_with_tls_context<S>(s: Pin<&mut S>) -> Poll<Option<S::Item>>
+where
+    S: Stream,
+{
+    get_task_context(|cx| S::poll_next(s, cx))
+}
+
+// The `await!` called in `async_stream` needs to be adjust to yield `Poll`,
+// but for this purpose, we don't want the user to use `#![feature(gen_future)]`.
+/// Polls a future in the current thread-local task waker.
+#[inline]
+pub fn poll_with_tls_context<F>(f: Pin<&mut F>) -> Poll<F::Output>
+where
+    F: Future
+{
+    future::poll_with_tls_context(f)
+}

futures-util/src/lib.rs

@@ -3,6 +3,7 @@
 
 #![cfg_attr(feature = "async-await", feature(async_await))]
 #![cfg_attr(feature = "cfg-target-has-atomic", feature(cfg_target_has_atomic))]
+#![cfg_attr(feature = "async-stream", feature(gen_future, generator_trait, generators))]
 
 #![cfg_attr(not(feature = "std"), no_std)]
 #![warn(missing_docs, missing_debug_implementations, rust_2018_idioms, unreachable_pub)]
@@ -20,6 +21,9 @@ compile_error!("The `cfg-target-has-atomic` feature requires the `nightly` featu
 #[cfg(all(feature = "async-await", not(feature = "nightly")))]
 compile_error!("The `async-await` feature requires the `nightly` feature as an explicit opt-in to unstable features");
 
+#[cfg(all(feature = "async-stream", not(feature = "nightly")))]
+compile_error!("The `async-stream` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+
 #[cfg(feature = "alloc")]
 extern crate alloc;
 
@@ -38,6 +42,14 @@ pub mod async_await;
 #[doc(hidden)]
 pub use self::async_await::*;
 
+#[cfg(feature = "async-stream")]
+#[macro_use]
+#[doc(hidden)]
+pub mod async_stream;
+#[cfg(feature = "async-stream")]
+#[doc(hidden)]
+pub use self::async_stream::*;
+
 #[cfg(feature = "select-macro")]
 #[doc(hidden)]
 pub mod rand_reexport { // used by select!

futures/Cargo.toml

@@ -28,6 +28,7 @@ futures-executor-preview = { path = "../futures-executor", version = "=0.3.0-alp
 futures-io-preview = { path = "../futures-io", version = "=0.3.0-alpha.17", default-features = false }
 futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.17", default-features = false }
 futures-util-preview = { path = "../futures-util", version = "=0.3.0-alpha.17", default-features = false, features = ["sink"] }
+futures-async-macro-preview = { path = "../futures-async-macro", version = "=0.3.0-alpha.17", optional = true }
 
 [dev-dependencies]
 pin-utils = "0.1.0-alpha.4"
@@ -44,6 +45,7 @@ async-await = ["futures-util-preview/async-await", "futures-util-preview/select-
 compat = ["std", "futures-util-preview/compat"]
 io-compat = ["compat", "futures-util-preview/io-compat"]
 cfg-target-has-atomic = ["futures-core-preview/cfg-target-has-atomic", "futures-channel-preview/cfg-target-has-atomic", "futures-util-preview/cfg-target-has-atomic"]
+async-stream = ["std", "async-await", "futures-util-preview/async-stream", "futures-async-macro-preview"]
 
 [package.metadata.docs.rs]
 all-features = true

futures/src/lib.rs

@@ -40,6 +40,9 @@ compile_error!("The `async-await` feature requires the `nightly` feature as an e
 #[cfg(all(feature = "cfg-target-has-atomic", not(feature = "nightly")))]
 compile_error!("The `cfg-target-has-atomic` feature requires the `nightly` feature as an explicit opt-in to unstable features");
 
+#[cfg(all(feature = "async-stream", not(feature = "nightly")))]
+compile_error!("The `async-stream` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+
 #[doc(hidden)] pub use futures_core::core_reexport;
 
 #[doc(hidden)] pub use futures_core::future::Future;
@@ -72,6 +75,13 @@ pub use futures_util::{
     // Async-await
     join, try_join, pending, poll,
 };
+// Async stream
+#[cfg(feature = "async-stream")]
+#[doc(hidden)]
+pub use futures_util::async_stream;
+#[cfg(feature = "async-stream")]
+#[doc(hidden)] // https://github.com/rust-lang/rust/issues/58696
+pub use futures_async_macro::*;
 
 #[cfg_attr(
     feature = "cfg-target-has-atomic",

futures/testcrate/Cargo.toml

@@ -2,16 +2,17 @@
 name = "testcrate"
 version = "0.1.0"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
+edition = "2018"
+publish = false
 
 [lib]
 path = "lib.rs"
 
-[dependencies.futures]
-features = ["std", "nightly"]
-path = ".."
+[dependencies]
+futures-preview = { path = "..", features = ["async-stream", "nightly"] }
 
 [dev-dependencies]
-compiletest_rs = "0.3.7"
+compiletest = { version = "0.3.21", package = "compiletest_rs", features = ["stable"] }
 
 [[test]]
 name = "ui"

futures/testcrate/lib.rs

@@ -0,0 +1 @@
+

futures/testcrate/tests/ui.rs

@@ -2,18 +2,24 @@ fn run_mode(mode: &'static str) {
     use std::env;
     use std::path::PathBuf;
 
-    let mut config = compiletest_rs::Config::default();
+    let mut config = compiletest::Config::default();
     config.mode = mode.parse().expect("invalid mode");
     let mut me = env::current_exe().unwrap();
     me.pop();
-    config.target_rustcflags = Some(format!("-L {}", me.display()));
+    config.target_rustcflags = Some(format!(
+        "--edition=2018 \
+         -Z unstable-options \
+         --extern futures \
+         -L {}",
+        me.display()
+    ));
     let src = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
     config.src_base = src.join(mode);
 
     me.pop();
     me.pop();
     config.build_base = me.join("tests").join(mode);
-    compiletest_rs::run_tests(&config);
+    compiletest::run_tests(&config);
 }
 
 fn main() {

futures/testcrate/ui/bad-input.rs

@@ -0,0 +1,22 @@
+
+#![feature(async_await, generators)]
+
+use futures::*;
+
+#[async_stream(item = i32)]
+fn foo() {
+    #[for_await(bar)]
+    for i in stream::iter(vec![1, 2]) {
+        yield i;
+    }
+}
+
+#[async_stream(baz, item = i32)]
+fn bar() {
+    #[for_await]
+    for i in stream::iter(vec![1, 2]) {
+        yield i;
+    }
+}
+
+fn main() {}

futures/testcrate/ui/bad-input.stderr

@@ -0,0 +1,14 @@
+error: attribute must be of the form `#[for_await]`
+ --> $DIR/bad-input.rs:8:5
+  |
+8 |     #[for_await(bar)]
+  |     ^^^^^^^^^^^^^^^^^
+
+error: expected `item`
+  --> $DIR/bad-input.rs:14:16
+   |
+14 | #[async_stream(baz, item = i32)]
+   |                ^^^
+
+error: aborting due to 2 previous errors
+

futures/testcrate/ui/bad-item-type.rs

@@ -0,0 +1,24 @@
+#![feature(async_await, generators)]
+
+use futures::*;
+
+#[async_stream(item = Option<i32>)]
+fn foobar() {
+    let val = Some(42);
+    if val.is_none() {
+        yield None;
+        return;
+    }
+    let val = val.unwrap();
+    yield val;
+}
+
+#[async_stream(item = (i32, i32))]
+fn tuple() {
+    if false {
+        yield 3;
+    }
+    yield (1, 2)
+}
+
+fn main() {}

futures/testcrate/ui/bad-item-type.stderr

@@ -0,0 +1,131 @@
+error[E0308]: mismatched types
+  --> $DIR/bad-item-type.rs:13:11
+   |
+13 |     yield val;
+   |           ^^^
+   |           |
+   |           expected enum `std::option::Option`, found integer
+   |           help: try using a variant of the expected type: `Some(val)`
+   |
+   = note: expected type `std::option::Option<_>`
+              found type `{integer}`
+
+error[E0698]: type inside generator must be known in this context
+ --> $DIR/bad-item-type.rs:7:9
+  |
+7 |     let val = Some(42);
+  |         ^^^ cannot infer type for `{integer}`
+  |
+note: the type is part of the generator because of this `yield`
+ --> $DIR/bad-item-type.rs:5:1
+  |
+5 | #[async_stream(item = Option<i32>)]
+  | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error[E0698]: type inside generator must be known in this context
+  --> $DIR/bad-item-type.rs:12:9
+   |
+12 |     let val = val.unwrap();
+   |         ^^^ cannot infer type for `{integer}`
+   |
+note: the type is part of the generator because of this `yield`
+  --> $DIR/bad-item-type.rs:5:1
+   |
+5  | #[async_stream(item = Option<i32>)]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error[E0698]: type inside generator must be known in this context
+  --> $DIR/bad-item-type.rs:13:11
+   |
+13 |     yield val;
+   |           ^^^ cannot infer type for `{integer}`
+   |
+note: the type is part of the generator because of this `yield`
+  --> $DIR/bad-item-type.rs:5:1
+   |
+5  | #[async_stream(item = Option<i32>)]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error[E0308]: mismatched types
+  --> $DIR/bad-item-type.rs:21:11
+   |
+21 |     yield (1, 2)
+   |           ^^^^^^ expected integer, found tuple
+   |
+   = note: expected type `{integer}`
+              found type `({integer}, {integer})`
+
+error[E0271]: type mismatch resolving `<impl futures_core::stream::Stream as futures_core::stream::Stream>::Item == (i32, i32)`
+  --> $DIR/bad-item-type.rs:16:23
+   |
+16 | #[async_stream(item = (i32, i32))]
+   |                       ^^^^^^^^^^ expected integer, found tuple
+   |
+   = note: expected type `{integer}`
+              found type `(i32, i32)`
+   = note: the return type of a function must have a statically known size
+
+error[E0698]: type inside generator must be known in this context
+  --> $DIR/bad-item-type.rs:16:1
+   |
+16 | #[async_stream(item = (i32, i32))]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ cannot infer type for `{integer}`
+   |
+note: the type is part of the generator because of this `yield`
+  --> $DIR/bad-item-type.rs:16:1
+   |
+16 | #[async_stream(item = (i32, i32))]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error[E0698]: type inside generator must be known in this context
+  --> $DIR/bad-item-type.rs:19:15
+   |
+19 |         yield 3;
+   |               ^ cannot infer type for `{integer}`
+   |
+note: the type is part of the generator because of this `yield`
+  --> $DIR/bad-item-type.rs:16:1
+   |
+16 | #[async_stream(item = (i32, i32))]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error[E0698]: type inside generator must be known in this context
+  --> $DIR/bad-item-type.rs:21:12
+   |
+21 |     yield (1, 2)
+   |            ^ cannot infer type for `{integer}`
+   |
+note: the type is part of the generator because of this `yield`
+  --> $DIR/bad-item-type.rs:16:1
+   |
+16 | #[async_stream(item = (i32, i32))]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error[E0698]: type inside generator must be known in this context
+  --> $DIR/bad-item-type.rs:21:15
+   |
+21 |     yield (1, 2)
+   |               ^ cannot infer type for `{integer}`
+   |
+note: the type is part of the generator because of this `yield`
+  --> $DIR/bad-item-type.rs:16:1
+   |
+16 | #[async_stream(item = (i32, i32))]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error[E0698]: type inside generator must be known in this context
+  --> $DIR/bad-item-type.rs:21:11
+   |
+21 |     yield (1, 2)
+   |           ^^^^^^ cannot infer type for `{integer}`
+   |
+note: the type is part of the generator because of this `yield`
+  --> $DIR/bad-item-type.rs:16:1
+   |
+16 | #[async_stream(item = (i32, i32))]
+   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+error: aborting due to 11 previous errors
+
+Some errors have detailed explanations: E0271, E0308, E0698.
+For more information about an error, try `rustc --explain E0271`.

futures/testcrate/ui/bad-return-type.rs

@@ -1,33 +1,11 @@
-#![feature(proc_macro, generators)]
+#![feature(async_await, generators)]
 
-#[async]
-fn foobar() -> Result<Option<i32>, ()> {
-    let val = Some(42);
-    if val.is_none() {
-        return Ok(None)
-    }
-    let val = val.unwrap();
-    Ok(val)
-}
+use futures::*;
 
 #[async_stream(item = Option<i32>)]
-fn foobars() -> Result<(), ()> {
-    let val = Some(42);
-    if val.is_none() {
-        stream_yield!(None);
-        return Ok(())
-    }
-    let val = val.unwrap();
-    stream_yield!(val);
-    Ok(())
-}
+fn foo() -> i32 {} // ERROR
 
-#[async]
-fn tuple() -> Result<(i32, i32), ()> {
-    if false {
-        return Ok(3);
-    }
-    Ok((1, 2))
-}
+#[async_stream(item = (i32, i32))]
+fn tuple() -> () {} // OK
 
 fn main() {}

futures/testcrate/ui/bad-return-type.stderr

@@ -1,90 +1,8 @@
-error[E0308]: mismatched types
-  --> $DIR/bad-return-type.rs:14:8
-   |
-14 |     Ok(val)
-   |        ^^^
-   |        |
-   |        expected enum `std::option::Option`, found integral variable
-   |        help: try using a variant of the expected type: `Some(val)`
-   |
-   = note: expected type `std::option::Option<i32>`
-              found type `{integer}`
+error: async stream functions must return the unit type
+ --> $DIR/bad-return-type.rs:6:13
+  |
+6 | fn foo() -> i32 {} // ERROR
+  |             ^^^
 
-error[E0308]: mismatched types
-  --> $DIR/bad-return-type.rs:25:5
-   |
-25 |     stream_yield!(val);
-   |     ^^^^^^^^^^^^^^^^^^^
-   |     |
-   |     expected enum `std::option::Option`, found integral variable
-   |     help: try using a variant of the expected type: `Some(e)`
-   |
-   = note: expected type `std::option::Option<_>`
-              found type `{integer}`
-   = note: this error originates in a macro outside of the current crate (in Nightly builds, run with -Z external-macro-backtrace for more info)
+error: aborting due to previous error
 
-error[E0907]: type inside generator must be known in this context
-  --> $DIR/bad-return-type.rs:19:9
-   |
-19 |     let val = Some(42);
-   |         ^^^
-   |
-note: the type is part of the generator because of this `yield`
-  --> $DIR/bad-return-type.rs:25:5
-   |
-25 |     stream_yield!(val);
-   |     ^^^^^^^^^^^^^^^^^^^
-   = note: this error originates in a macro outside of the current crate (in Nightly builds, run with -Z external-macro-backtrace for more info)
-
-error[E0907]: type inside generator must be known in this context
-  --> $DIR/bad-return-type.rs:24:9
-   |
-24 |     let val = val.unwrap();
-   |         ^^^
-   |
-note: the type is part of the generator because of this `yield`
-  --> $DIR/bad-return-type.rs:25:5
-   |
-25 |     stream_yield!(val);
-   |     ^^^^^^^^^^^^^^^^^^^
-   = note: this error originates in a macro outside of the current crate (in Nightly builds, run with -Z external-macro-backtrace for more info)
-
-error[E0907]: type inside generator must be known in this context
-  --> $DIR/bad-return-type.rs:25:5
-   |
-25 |     stream_yield!(val);
-   |     ^^^^^^^^^^^^^^^^^^^
-   |
-note: the type is part of the generator because of this `yield`
-  --> $DIR/bad-return-type.rs:25:5
-   |
-25 |     stream_yield!(val);
-   |     ^^^^^^^^^^^^^^^^^^^
-   = note: this error originates in a macro outside of the current crate (in Nightly builds, run with -Z external-macro-backtrace for more info)
-
-error[E0907]: type inside generator must be known in this context
-  --> $DIR/bad-return-type.rs:25:5
-   |
-25 |     stream_yield!(val);
-   |     ^^^^^^^^^^^^^^^^^^^
-   |
-note: the type is part of the generator because of this `yield`
-  --> $DIR/bad-return-type.rs:25:5
-   |
-25 |     stream_yield!(val);
-   |     ^^^^^^^^^^^^^^^^^^^
-   = note: this error originates in a macro outside of the current crate (in Nightly builds, run with -Z external-macro-backtrace for more info)
-
-error[E0308]: mismatched types
-  --> $DIR/bad-return-type.rs:32:19
-   |
-32 |         return Ok(3);
-   |                   ^ expected tuple, found integral variable
-   |
-   = note: expected type `(i32, i32)`
-              found type `{integer}`
-
-error: aborting due to 7 previous errors
-
-Some errors occurred: E0308, E0907.
-For more information about an error, try `rustc --explain E0308`.

futures/testcrate/ui/forget-semicolon.rs

@@ -0,0 +1,11 @@
+#![feature(async_await, generators)]
+
+use futures::*;
+
+#[async_stream(item = ())]
+fn foo() {
+    yield;
+    Some(())
+}
+
+fn main() {}

futures/testcrate/ui/forget-semicolon.stderr

@@ -0,0 +1,14 @@
+error[E0308]: mismatched types
+ --> $DIR/forget-semicolon.rs:8:5
+  |
+8 |     Some(())
+  |     ^^^^^^^^- help: try adding a semicolon: `;`
+  |     |
+  |     expected (), found enum `std::option::Option`
+  |
+  = note: expected type `()`
+             found type `std::option::Option<()>`
+
+error: aborting due to previous error
+
+For more information about this error, try `rustc --explain E0308`.

futures/testcrate/ui/missing-item.rs

@@ -1,9 +1,8 @@
-#![allow(warnings)]
-#![feature(proc_macro, generators)]
+#![feature(async_await, generators)]
+
+use futures::*;
 
 #[async_stream]
-fn foos(a: String) -> Result<(), u32> {
-    Ok(())
-}
+fn foo(a: String) {}
 
 fn main() {}

futures/testcrate/ui/missing-item.stderr

@@ -1,10 +1,8 @@
-error: custom attribute panicked
- --> $DIR/missing-item.rs:8:1
+error: unexpected end of input, expected `item`
+ --> $DIR/missing-item.rs:5:1
   |
-8 | #[async_stream]
+5 | #[async_stream]
   | ^^^^^^^^^^^^^^^
-  |
-  = help: message: #[async_stream] requires item type to be specified
 
 error: aborting due to previous error
 

futures/testcrate/ui/move-captured-variable.rs

@@ -1,12 +1,14 @@
-#![feature(proc_macro, generators)]
+#![feature(async_await, generators, proc_macro_hygiene)]
+
+use futures::*;
 
 fn foo<F: FnMut()>(_f: F) {}
 
 fn main() {
     let a = String::new();
     foo(|| {
-        async_block! {
-            Ok::<String, i32>(a)
+        async_stream_block! {
+            yield a
         };
     });
 }

futures/testcrate/ui/move-captured-variable.stderr

@@ -1,13 +1,17 @@
-error[E0507]: cannot move out of captured outer variable in an `FnMut` closure
-  --> $DIR/move-captured-variable.rs:12:9
+error[E0507]: cannot move out of `a`, a captured variable in an `FnMut` closure
+  --> $DIR/move-captured-variable.rs:10:9
    |
-10 |       let a = String::new();
+8  |       let a = String::new();
    |           - captured outer variable
-11 |       foo(|| {
-12 | /         async_block! {
-13 | |             Ok::<String, i32>(a)
-14 | |         };
-   | |__________^ cannot move out of captured outer variable in an `FnMut` closure
+9  |       foo(|| {
+10 | /         async_stream_block! {
+11 | |             yield a
+   | |                   -
+   | |                   |
+   | |                   move occurs because `a` has type `std::string::String`, which does not implement the `Copy` trait
+   | |                   move occurs due to use in generator
+12 | |         };
+   | |__________^ move out of `a` occurs here
 
 error: aborting due to previous error
 

futures/testcrate/ui/nested.rs

@@ -0,0 +1,15 @@
+#![feature(async_await, generators)]
+
+use futures::*;
+
+#[async_stream(item = i32)]
+fn _stream1() {
+    let _ = async {
+        #[for_await]
+        for i in stream::iter(vec![1, 2]) {
+            yield i * i;
+        }
+    };
+}
+
+fn main() {}

futures/testcrate/ui/nested.stderr

@@ -0,0 +1,8 @@
+error[E0727]: `async` generators are not yet supported
+  --> $DIR/nested.rs:10:13
+   |
+10 |             yield i * i;
+   |             ^^^^^^^^^^^
+
+error: aborting due to previous error
+

futures/testcrate/ui/type_error.rs

@@ -1,9 +1,11 @@
-#![feature(proc_macro, generators)]
+#![feature(async_await, generators)]
 
-#[async]
-fn foo() -> Result<i32, i32> {
+use futures::*;
+
+#[async_stream(item = i32)]
+fn foo() {
     let a: i32 = "a"; //~ ERROR: mismatched types
-    Ok(1)
+    yield 1;
 }
 
 fn main() {}

futures/testcrate/ui/type_error.stderr

@@ -1,7 +1,7 @@
 error[E0308]: mismatched types
- --> $DIR/type_error.rs:9:18
+ --> $DIR/type_error.rs:7:18
   |
-9 |     let a: i32 = "a"; //~ ERROR: mismatched types
+7 |     let a: i32 = "a"; //~ ERROR: mismatched types
   |                  ^^^ expected i32, found reference
   |
   = note: expected type `i32`

futures/testcrate/ui/unresolved-type.rs

@@ -1,13 +1,8 @@
-#![feature(proc_macro, generators)]
+#![feature(async_await, generators)]
 
-#[async]
-fn foo() -> Result<Left, u32> {
-    Err(3)
-}
+use futures::*;
 
 #[async_stream(item = Left)]
-fn foos() -> Result<(), u32> {
-    Err(3)
-}
+fn foo() {}
 
 fn main() {}

futures/testcrate/ui/unresolved-type.stderr

@@ -1,34 +1,20 @@
 error[E0412]: cannot find type `Left` in this scope
- --> $DIR/unresolved-type.rs:8:20
+ --> $DIR/unresolved-type.rs:5:23
   |
-8 | fn foo() -> Result<Left, u32> {
-  |                    ^^^^ not found in this scope
+5 | #[async_stream(item = Left)]
+  |                       ^^^^ not found in this scope
+help: there is an enum variant `core::fmt::Alignment::Left` and 7 others; try using the variant's enum
   |
-  = help: there is an enum variant `futures::future::Either::Left`, try using `futures::future::Either`?
-  = help: there is an enum variant `std::fmt::rt::v1::Alignment::Left`, try using `std::fmt::rt::v1::Alignment`?
+5 | #[async_stream(item = core::fmt::Alignment)]
+  |                       ^^^^^^^^^^^^^^^^^^^^
+5 | #[async_stream(item = core::fmt::rt::v1::Alignment)]
+  |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+5 | #[async_stream(item = futures::core_reexport::fmt::Alignment)]
+  |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+5 | #[async_stream(item = futures::core_reexport::fmt::rt::v1::Alignment)]
+  |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+and 3 other candidates
 
-error[E0412]: cannot find type `Left` in this scope
-  --> $DIR/unresolved-type.rs:12:23
-   |
-12 | #[async_stream(item = Left)]
-   |                       ^^^^ not found in this scope
-   |
-   = help: there is an enum variant `futures::future::Either::Left`, try using `futures::future::Either`?
-   = help: there is an enum variant `std::fmt::rt::v1::Alignment::Left`, try using `std::fmt::rt::v1::Alignment`?
-
-error[E0907]: type inside generator must be known in this context
-  --> $DIR/unresolved-type.rs:12:1
-   |
-12 | #[async_stream(item = Left)]
-   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-   |
-note: the type is part of the generator because of this `yield`
-  --> $DIR/unresolved-type.rs:12:1
-   |
-12 | #[async_stream(item = Left)]
-   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-error: aborting due to 3 previous errors
+error: aborting due to previous error
 
-Some errors occurred: E0412, E0907.
-For more information about an error, try `rustc --explain E0412`.
+For more information about this error, try `rustc --explain E0412`.

futures/testcrate/ui/update-all-references.sh

@@ -21,3 +21,4 @@
 MY_DIR=$(dirname $0)
 cd $MY_DIR
 find . -name '*.rs' | xargs ./update-references.sh
+cd -

futures/testcrate/ui/update-references.sh

@@ -21,7 +21,7 @@
 
 MYDIR=$(dirname $0)
 
-BUILD_DIR="../../target/tests/ui"
+BUILD_DIR="../target/tests/ui"
 
 while [[ "$1" != "" ]]; do
     STDERR_NAME="${1/%.rs/.stderr}"
@@ -38,5 +38,3 @@ while [[ "$1" != "" ]]; do
         cp $BUILD_DIR/$STDERR_NAME $MYDIR/$STDERR_NAME
     fi
 done
-
-

futures/tests/async_stream/elisions.rs

@@ -0,0 +1,66 @@
+use futures::executor::block_on;
+use futures::*;
+
+struct Ref<'a, T>(&'a T);
+
+#[async_stream(item = i32)]
+fn references(x: &i32) {
+    yield *x;
+}
+
+#[async_stream(item = i32)]
+fn new_types(x: Ref<'_, i32>) {
+    yield *x.0;
+}
+
+struct Foo(i32);
+
+impl Foo {
+    #[async_stream(item = &i32)]
+    fn foo(&self) {
+        yield &self.0
+    }
+}
+
+#[async_stream(item = &i32)]
+fn single_ref(x: &i32) {
+    yield x
+}
+
+#[async_stream(item = ())]
+fn check_for_name_colision<'_async0, T>(_x: &T, _y: &'_async0 i32) {
+    yield
+}
+
+#[test]
+fn main() {
+    block_on(async {
+        let x = 0;
+        let foo = Foo(x);
+
+        #[for_await]
+        for y in references(&x) {
+            assert_eq!(y, x);
+        }
+
+        #[for_await]
+        for y in new_types(Ref(&x)) {
+            assert_eq!(y, x);
+        }
+
+        #[for_await]
+        for y in single_ref(&x) {
+            assert_eq!(y, &x);
+        }
+
+        #[for_await]
+        for y in foo.foo() {
+            assert_eq!(y, &x);
+        }
+
+        #[for_await]
+        for y in check_for_name_colision(&x, &x) {
+            assert_eq!(y, ());
+        }
+    });
+}

futures/tests_disabled/async_await/mod.rs → futures/tests/async_stream/mod.rs

@@ -1,3 +1,4 @@
 mod elisions;
+mod nested;
 mod pinned;
 mod smoke;

futures/tests/async_stream/nested.rs

@@ -0,0 +1,13 @@
+
+use futures::*;
+
+#[async_stream(item = i32)] // impl Generator<Yield = Poll<U>, Return = ()>
+fn _stream1() {
+    let _ = async { // impl Generator<Yield = (), Return = U>
+        #[for_await]
+        for i in stream::iter(vec![1, 2]) {
+            future::lazy(|_| i * i).await;
+        }
+    };
+    future::lazy(|_| ()).await;
+}

futures/tests/async_stream/pinned.rs

@@ -0,0 +1,45 @@
+use futures::executor::block_on;
+use futures::*;
+
+#[async_stream(item = u64)]
+fn stream1() {
+    fn integer() -> u64 {
+        1
+    }
+    let x = &integer();
+    yield 0;
+    yield *x;
+}
+
+fn stream1_block() -> impl Stream<Item = u64> {
+    async_stream_block! {
+        #[for_await]
+        for item in stream1() {
+            yield item
+        }
+    }
+}
+
+async fn uses_async_for() -> Vec<u64> {
+    let mut v = vec![];
+    #[for_await]
+    for i in stream1() {
+        v.push(i);
+    }
+    v
+}
+
+async fn uses_async_for_block() -> Vec<u64> {
+    let mut v = vec![];
+    #[for_await]
+    for i in stream1_block() {
+        v.push(i);
+    }
+    v
+}
+
+#[test]
+fn main() {
+    assert_eq!(block_on(uses_async_for()), vec![0, 1]);
+    assert_eq!(block_on(uses_async_for_block()), vec![0, 1]);
+}

futures/tests/async_stream/smoke.rs

@@ -0,0 +1,133 @@
+//! A bunch of ways to use async/await syntax.
+//!
+//! This is mostly a test f r this repository itself, not necessarily serving
+//! much more purpose than that.
+
+use futures::executor::block_on;
+use futures::*;
+
+async fn future1() -> i32 {
+    let mut cnt = 0;
+    #[for_await]
+    for x in stream::iter(vec![1, 2, 3, 4]) {
+        cnt += x;
+    }
+    cnt
+}
+
+#[async_stream(item = u64)]
+fn stream1() {
+    yield 0;
+    yield 1;
+}
+
+#[async_stream(item = T)]
+fn stream2<T: Clone>(t: T) {
+    yield t.clone();
+    yield t.clone();
+}
+
+#[async_stream(item = i32)]
+fn stream3() {
+    let mut cnt = 0;
+    #[for_await]
+    for x in stream::iter(vec![1, 2, 3, 4]) {
+        cnt += x;
+        yield x;
+    }
+    yield cnt;
+}
+
+mod foo {
+    pub struct _Foo(pub i32);
+}
+
+#[async_stream(item = foo::_Foo)]
+fn _stream5() {
+    yield foo::_Foo(0);
+    yield foo::_Foo(1);
+}
+
+#[async_stream(item = i32)]
+fn _stream6() {
+    #[for_await]
+    for foo::_Foo(i) in _stream5() {
+        yield i * i;
+    }
+}
+
+#[async_stream(item = ())]
+fn _stream7() {
+    yield ();
+}
+
+#[async_stream(item = [u32; 4])]
+fn _stream8() {
+    yield [1, 2, 3, 4];
+}
+
+struct A(i32);
+
+impl A {
+    #[async_stream(item = i32)]
+    fn a_foo(self) {
+        yield self.0
+    }
+
+    #[async_stream(item = i32)]
+    fn _a_foo2(self: Box<Self>) {
+        yield self.0
+    }
+}
+
+async fn loop_in_loop() -> bool {
+    let mut cnt = 0;
+    let vec = vec![1, 2, 3, 4];
+    #[for_await]
+    for x in stream::iter(vec.clone()) {
+        #[for_await]
+        for y in stream::iter(vec.clone()) {
+            cnt += x * y;
+        }
+    }
+
+    let sum = (1..5).map(|x| (1..5).map(|y| x * y).sum::<i32>()).sum::<i32>();
+    cnt == sum
+}
+
+#[test]
+fn main() {
+    // https://github.com/alexcrichton/futures-await/issues/45
+    #[async_stream(item = ())]
+    fn _stream10() {
+        yield;
+    }
+
+    block_on(async {
+        let mut v = 0..=1;
+        #[for_await]
+        for x in stream1() {
+            assert_eq!(x, v.next().unwrap());
+        }
+
+        let mut v = [(), ()].iter();
+        #[for_await]
+        for x in stream2(()) {
+            assert_eq!(x, *v.next().unwrap());
+        }
+
+        let mut v = [1, 2, 3, 4, 10].iter();
+        #[for_await]
+        for x in stream3() {
+            assert_eq!(x, *v.next().unwrap());
+        }
+
+        #[for_await]
+        for x in A(11).a_foo() {
+            assert_eq!(x, 11);
+        }
+    });
+
+    assert_eq!(block_on(future1()), 10);
+    assert_eq!(block_on(loop_in_loop()), true);
+}

futures/tests/async_stream_tests.rs

@@ -0,0 +1,5 @@
+#![feature(async_await)]
+#![cfg_attr(all(feature = "async-stream", feature = "nightly"), feature(generators, stmt_expr_attributes, proc_macro_hygiene))]
+
+#[cfg(all(feature = "async-stream", feature = "nightly"))]
+mod async_stream;