How to cancel an infinite stream from within the stream itself?
For cases where you want to cancel a stream from outside of the stream, see stream-cancel.
For your specific case, it's easiest to convert your collection into a stream and zip it together with the interval timer. This way, the resulting stream naturally stops when the collection is empty:
use futures::{future, stream, Stream}; // 0.1.29
use std::time::Duration;
use tokio; // 0.1.22
use tokio_timer::Interval; // 0.2.11
fn main() {
tokio::run({
let some_vars = vec![1, 2, 3, 4, 5, 6, 7, 8];
let timer =
Interval::new_interval(Duration::from_millis(100)).map_err(|e| panic!("Error: {}", e));
let some_vars = stream::iter_ok(some_vars.into_iter().rev());
let combined = timer.zip(some_vars);
combined.for_each(move |(_, item)| {
eprintln!("Woke up");
tokio::spawn(future::lazy(move || {
println!("{:?}", item);
Ok(())
}));
Ok(())
})
});
}
Otherwise, you can stop the stream by using and_then
to both remove the value from the collection and control if the stream should continue:
use futures::{future, Stream}; // 0.1.29
use std::time::Duration;
use tokio; // 0.1.22
use tokio_timer::Interval; // 0.2.11
fn main() {
tokio::run({
let mut some_vars = vec![1, 2, 3, 4, 5, 6, 7, 8];
let timer =
Interval::new_interval(Duration::from_millis(100)).map_err(|e| panic!("Error: {}", e));
let limited = timer.and_then(move |_| {
if some_vars.len() <= 4 {
Err(())
} else {
some_vars.pop().ok_or(())
}
});
limited.for_each(move |item| {
eprintln!("Woke up");
tokio::spawn(future::lazy(move || {
println!("{:?}", item);
Ok(())
}));
Ok(())
})
});
}