Code shared from the Rust Playground

playground.rs

extern crate tokio; // 0.1.8;
extern crate futures; // 0.1.24;

use futures::sync::mpsc;
use futures::{task, Async, Future, Sink, Stream};

pub fn invoke_bpm_battle_warm_up( // this method is called from the poll parent method
    receiver: &mut Option<mpsc::Receiver<bool>>,
    msg: BattleWarmUp,
    addr: Addr<Bpm>,
) -> Result<Option<bool>, Error> {
    let _ = receiver.get_or_insert_with(|| {
        let (tx, rx) = mpsc::channel::<bool>(1);
        let t = task::current(); // get current task
        let future = addr
            .send(msg)
            .map_err(Into::into)
            .and_then(|r| r)
            .map_err(|_| format_err!("mpsc send error"))
            .and_then(move |b| {
                t.notify(); // notify after future resolves
                // after debugging it leads to run:
                // https://github.com/rust-lang-nursery/futures-rs/blob/0.1.24/src/task_impl/std/mod.rs#L164
                // https://github.com/tokio-rs/tokio/blob/tokio-io-0.1.8/tokio-current-thread/src/scheduler.rs#L724
                // 
                tx.send(b)
                    .map_err(|_| format_err!("mpsc receive error"))
                    .map(|_| ())
            }).map_err(|_| ());

        let _ = tokio::spawn(future);
        rx
    });
    if let Some(ref mut rx) = receiver {
        rx.poll()
            .map_err(|_| format_err!("mpsc error").into())
            .map(|r| match r {
                Async::NotReady => None,
                Async::Ready(o) => o.map(|b| b),
            })
    } else {
        Ok(None) // it will return Ok(Async::NotReady) if there is None data to receive
    }
}


fn main() {
}