title: How can I implement a blocking queue mechanism with futures::sync::mpsc::channel?
body: <p>I am trying to understand how <a href="https://docs.rs/futures/0.1.25/futures/sync/mpsc/struct.Receiver.html" rel="nofollow noreferrer"><code>futures::sync::mpsc::Receiver</code></a> works. In the below example, the  receiver thread sleeps for two seconds and the sender sends every second. </p>
  
  <p>I expect that the sender will need to be blocked because of the wait and then send when the buffer is released. </p>
  
  <p>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.</p>
  
  <p>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? <a href="https://docs.rs/futures/0.1.25/futures/sync/mpsc/fn.channel.html" rel="nofollow noreferrer"><code>futures::sync::mpsc::channel</code></a> has its own documentation, but I do not understand how to use it properly.</p>
  
  <pre><code>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::&lt;(), ()&gt;(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(_) =&gt; stats.success += 1,
              Err(_) =&gt; stats.failure += 1,
          }
          println!("stats = {:?}", stats);
  
          Ok(())
      });
  
      core.run(f2).expect("Core failed to run");
  }
  </code></pre>