I'd like to be able to do something along these lines:
let (tx, rx) = futures::sync::mpsc::channel(1000);
let arc = Arc::new(Mutex::<Option<AndThen<Receiver<u32>, _, _>>>::new(None));
{
let mut and_then = arc.lock().unwrap();
*and_then = Some(rx.and_then(|num| {
println!("{}", num);
Ok(())
}));
}
let arc_clone = arc.clone();
// after 1 second, try to close the receiver
tokio::spawn(
Delay::new(std::time::Instant::now() + std::time::Duration::from_secs(1))
.map_err(|e| eprintln!("Some delay err {:?}", e))
.and_then(move |_| {
let mut maybe_stream = arc_clone.lock().unwrap();
// take the `AndThen` out of global storage,
// commute it to the inner `Receiver` instance,
// and close the receiver
match maybe_stream.take() {
Some(stream) => stream.into_inner().close(),
None => eprintln!("Can't close non-existent stream"), // line "A"
}
Ok(())
}),
);
{
let mut maybe_stream = arc.lock().unwrap();
let stream = maybe_stream.take().expect("Stream already ripped out"); // line "B"
let rx = stream.for_each(|_| Ok(()));
tokio::spawn(rx);
}
in order to close a Receiver stream asynchronously
However, line A runs because I have to move the stream on line B in order to call .for_each on it. If I don't call .for_each (or something like it), I can't execute the AndThen at all, as far as I know. I can't call .for_each without actually moving the object because for_each is a moving method.
Is it possible for me to do what I'm trying to do? This seems like it should definitely be possible, but maybe I'm missing something obvious.
I'm using futures at 0.1 and tokio at 0.1.