Futures-rs: Can't loop on Stream with async-await

[xrl]

I don't know how to write a Stream-backed async code. I was able to pull out a single item from a stream but I don't know how to call it more than once, in a looping fashion. Am I missing something?

Here's my Stream example, uses repeat and add a value to it:

#![feature(async_await,generators,futures_api,pin,gen_future)]
use std::future;
use std::task::Poll;
use std::mem::PinMut;
use futures::stream::{ repeat, Stream, Repeat };
use futures_core::stream::StreamObj;
use futures_util::stream::StreamExt;

pub fn poll_next_in_task_cx<S>(s: PinMut<S>) -> Poll<Option<S::Item>>
    where
        S: Stream
{
    future::get_task_cx(|cx| s.poll_next(cx))
}

macro_rules! await_item {
    ($e:expr) => { {
        let mut pinned = $e;
        loop {
            if let Poll::Ready(x) =
                poll_next_in_task_cx(unsafe {
                    PinMut::new_unchecked(&mut pinned)
                })
            {
                break x;
            }
            // FIXME(cramertj) prior to stabilizing await, we have to ensure that this
            // can't be used to create a generator on stable via `|| await!()`.
            yield
        }
    } }
}

struct Response<'a> {
    stream: StreamObj<'a, u8>
}

async fn do_stuff() {
    let mut s = repeat(10).map(|v| v+1);
    let response_a_tron = Response {
        stream: StreamObj::new(&mut s)
    };
    let item = await_item!(response_a_tron.stream);
    println!("item: {:?}", item);
}

pub fn main() {
    fahrenheit::run(do_stuff());
}

which runs and prints:

 item: Some(11)

so that's cool. but now I don't know how to model looping/repeating. What's the expected path forward? If I put the await_item! inside of a loop I get a problem with values being moved in to the loop:

async fn do_stuff() {
    let mut s = repeat(10);
    let response_a_tron = test_response(&mut s);
    loop {
        let item = await_item!(
            response_a_tron.stream.map(|v| v+1)
        );
        println!("item: {:?}", item);
    }
}

errors with:

error[E0382]: use of moved value: `response_a_tron.stream`
  --> src/main.rs:49:13
   |
49 |             response_a_tron.stream.map(|v| v+1)
   |             ^^^^^^^^^^^^^^^^^^^^^^ value moved here in previous iteration of loop
   |
   = note: move occurs because `response_a_tron.stream` has type `futures_core::stream::stream_obj::StreamObj<'_, u8>`, which does not implement the `Copy` trait

which is fair, I guess. and I can't even change it to a borrow with let item = await_item!( &response_a_tron.stream.map(|v| v+1) ); because:

the trait `futures_core::stream::Stream` is not implemented for `&futures_util::stream::map::Map<futures_core::stream::stream_obj::StreamObj<'_, u8>, [closure@src/main.rs:49:41: 49:48]>`

Unwinding this exploration a bit, what I want is an iterator like behavior out of a stream. Something like

async fn do_stuff() {
    let mut s = repeat(10).map(|v| v+1);
    let response_a_tron = Response {
        stream: StreamObj::new(&mut s)
    };
    for item in await_item!(response_a_tron.stream) {
        println!("item: {:?}", item);
    };
}

which currently evalutes to a single value and the program exits, so it kinda works, but I only get one values and it's done.

Is a true iterator behavior doable? Is there a desire for this behavior in futures? Any prior art I should consider?

[tinaun]

what you want to use is the .next() method.

while let Some(item) = await!(stream.next()) {
    // etc
}

once the basics of futures have been finalized, i'd love to have nicer syntax like asnync for and friends for working with streams.

[xrl]

@tinaun thanks for the example, that got me going again! I had seen Next but didn't understand how to use it.

[nanreh]

With updated await syntax:

while let Some(a) = (stream.next()).await {
   // etc
}