I'd like to be able to do something along these lines in order to close a Receiver stream asynchronously:
extern crate futures;
extern crate tokio;
use futures::future::lazy;
use futures::stream::AndThen;
use futures::sync::mpsc::Receiver;
use futures::{Future, Sink, Stream};
use std::sync::{Arc, Mutex};
use tokio::timer::{Delay, Interval};
fn main() {
tokio::run(lazy(|| {
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();
// This is the part I'd like to be able to do
// After one second, close the `Receiver` so that future
// calls to the `Sender` don't call the callback above in the
// closure passed to `rx.and_then`
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();
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);
}
tokio::spawn(
Interval::new_interval(std::time::Duration::from_millis(10))
.take(10)
.map_err(|e| {
eprintln!("Interval error?! {:?}", e);
})
.fold((tx, 0), |(tx, i), _| {
tx.send(i as u32)
.map_err(|e| eprintln!("Send error?! {:?}", e))
.map(move |tx| (tx, i + 1))
})
.map(|_| ()),
);
Ok(())
}));
}
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.