I am trying to understand how futures::sync::mpsc::Receiver
works. In the below example, the receiver thread sleeps for two seconds and the sender sends every second.
I expect that the sender will need to be blocked because of the wait and then send when the buffer is released.
What I see instead is that it is deadlocked after a time. Increasing the buffer of the channel only extends the time until it is blocked.
What should I do to make the sender send data when the buffer is available and put some backpressure to the sender in such cases? futures::sync::mpsc::channel
has its own documentation, but I do not understand how to use it properly.
extern crate futures;
extern crate tokio_core;
use std::{thread, time};
use futures::sync::mpsc;
use futures::{Future, Sink, Stream};
use tokio_core::reactor::Core;
#[derive(Debug)]
struct Stats {
pub success: usize,
pub failure: usize,
}
fn main() {
let mut core = Core::new().expect("Failed to create core");
let remote = core.remote();
let (tx, rx) = mpsc::channel(1);
thread::spawn(move || loop {
let tx = tx.clone();
let delay = time::Duration::from_secs(1);
thread::sleep(delay);
let f = ::futures::done::<(), ()>(Ok(()));
remote.spawn(|_| {
f.then(|res| {
println!("Sending");
tx.send(res).wait();
println!("Sent");
Ok(())
})
});
});
let mut stats = Stats {
success: 0,
failure: 0,
};
let f2 = rx.for_each(|res| {
println!("Received");
let delay = time::Duration::from_secs(2);
thread::sleep(delay);
match res {
Ok(_) => stats.success += 1,
Err(_) => stats.failure += 1,
}
println!("stats = {:?}", stats);
Ok(())
});
core.run(f2).expect("Core failed to run");
}