-
Notifications
You must be signed in to change notification settings - Fork 27
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add StreamExt::take_until
#99
Comments
I tried to express this combinator using an existing API async fn run() {
use {
futures_lite::{stream, StreamExt},
std::{pin, time::Duration},
tokio::time,
};
// an origin stream
let stream = stream::iter(0..5).then(|i| async move {
time::sleep(Duration::from_secs(1)).await;
i
});
// wrap items in option, I need some mark that the original stream is finished
let stream = stream.map(Some).chain(stream::once(None));
// instead of `take_until` it will be a stream with a single item
let cancel = stream::once_future(async {
tokio::signal::ctrl_c().await.unwrap();
println!("cancelled");
None
});
// merge two streams
let stream = stream
.or(cancel)
.take_while(|opt| opt.is_some())
.map(|opt| opt.unwrap());
// use the final stream
let mut stream = pin::pin!(stream);
while let Some(item) = stream.next().await {
println!("item {item}");
}
} Overall it's pretty acceptable to me. It would be especially cool to replace But I think it would be simpler and more convenient to have the first class method. |
Inspired by your idea, I proposed this PR. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
How about adding a similar method?
https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html#method.take_until
This can be convenient for a stream cancellation.
The text was updated successfully, but these errors were encountered: