{"action":"opened","issue":{"url":"https://api.github.com/repos/rust-lang-nursery/futures-rs/issues/1227","repository_url":"https://api.github.com/repos/rust-lang-nursery/futures-rs","labels_url":"https://api.github.com/repos/rust-lang-nursery/futures-rs/issues/1227/labels{/name}","comments_url":"https://api.github.com/repos/rust-lang-nursery/futures-rs/issues/1227/comments","events_url":"https://api.github.com/repos/rust-lang-nursery/futures-rs/issues/1227/events","html_url":"https://github.com/rust-lang-nursery/futures-rs/issues/1227","id":354135135,"node_id":"MDU6SXNzdWUzNTQxMzUxMzU=","number":1227,"title":"Can't loop on Stream with async-await","user":{"login":"xrl","id":693059,"node_id":"MDQ6VXNlcjY5MzA1OQ==","avatar_url":"https://avatars2.githubusercontent.com/u/693059?v=4","gravatar_id":"","url":"https://api.github.com/users/xrl","html_url":"https://github.com/xrl","followers_url":"https://api.github.com/users/xrl/followers","following_url":"https://api.github.com/users/xrl/following{/other_user}","gists_url":"https://api.github.com/users/xrl/gists{/gist_id}","starred_url":"https://api.github.com/users/xrl/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/xrl/subscriptions","organizations_url":"https://api.github.com/users/xrl/orgs","repos_url":"https://api.github.com/users/xrl/repos","events_url":"https://api.github.com/users/xrl/events{/privacy}","received_events_url":"https://api.github.com/users/xrl/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2018-08-26T22:36:04Z","updated_at":"2018-08-26T22:36:04Z","closed_at":null,"author_association":"CONTRIBUTOR","body":"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?\r\n\r\nHere's my Stream example, uses repeat and add a value to it:\r\n\r\n```\r\n#![feature(async_await,generators,futures_api,pin,gen_future)]\r\nuse std::future;\r\nuse std::task::Poll;\r\nuse std::mem::PinMut;\r\nuse futures::stream::{ repeat, Stream, Repeat };\r\nuse futures_core::stream::StreamObj;\r\nuse futures_util::stream::StreamExt;\r\n\r\npub fn poll_next_in_task_cx<S>(s: PinMut<S>) -> Poll<Option<S::Item>>\r\n    where\r\n        S: Stream\r\n{\r\n    future::get_task_cx(|cx| s.poll_next(cx))\r\n}\r\n\r\nmacro_rules! await_item {\r\n    ($e:expr) => { {\r\n        let mut pinned = $e;\r\n        loop {\r\n            if let Poll::Ready(x) =\r\n                poll_next_in_task_cx(unsafe {\r\n                    PinMut::new_unchecked(&mut pinned)\r\n                })\r\n            {\r\n                break x;\r\n            }\r\n            // FIXME(cramertj) prior to stabilizing await, we have to ensure that this\r\n            // can't be used to create a generator on stable via `|| await!()`.\r\n            yield\r\n        }\r\n    } }\r\n}\r\n\r\nstruct Response<'a> {\r\n    stream: StreamObj<'a, u8>\r\n}\r\n\r\nfn test_response<'a>(s: &'a mut Repeat<u8>) -> Response<'a> {\r\n    Response {\r\n        stream: StreamObj::new(s)\r\n    }\r\n}\r\n\r\nasync fn do_stuff() {\r\n    let mut s = repeat(10);\r\n    let response_a_tron = test_response(&mut s);\r\n    let item = await_item!(\r\n        response_a_tron.stream.map(|v| v+1)\r\n    );\r\n    println!(\"item: {:?}\", item);\r\n}\r\n\r\npub fn main() {\r\n    fahrenheit::run(do_stuff());\r\n}\r\n```\r\n\r\nwhich runs and prints:\r\n\r\n```\r\n item: Some(11)\r\n```\r\n\r\nso 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:\r\n\r\n```\r\nasync fn do_stuff() {\r\n    let mut s = repeat(10);\r\n    let response_a_tron = test_response(&mut s);\r\n    loop {\r\n        let item = await_item!(\r\n            response_a_tron.stream.map(|v| v+1)\r\n        );\r\n        println!(\"item: {:?}\", item);\r\n    }\r\n}\r\n```\r\n\r\nerrors with:\r\n\r\n```\r\nerror[E0382]: use of moved value: `response_a_tron.stream`\r\n  --> src/main.rs:49:13\r\n   |\r\n49 |             response_a_tron.stream.map(|v| v+1)\r\n   |             ^^^^^^^^^^^^^^^^^^^^^^ value moved here in previous iteration of loop\r\n   |\r\n   = note: move occurs because `response_a_tron.stream` has type `futures_core::stream::stream_obj::StreamObj<'_, u8>`, which does not implement the `Copy` trait\r\n```\r\n\r\nwhich is fair, I guess. and I can't even change it to be `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]>``.\r\n\r\nAnd unwinding this exploration a bit, what I want is an iterator like behavior out of a stream. Something like \r\n\r\n```\r\nfor value in await_item!(response_a_tron.stream...) {\r\n  println!(\"val: {:?}\", value);\r\n}\r\n```\r\n\r\nIs this doable? Is there a desire for this behavior in futures? Any prior art I should consider?"}}
{"actor":{"display_login":"xrl"}}