async_service_oneshot_manage.rs

use futures::sync::oneshot::{channel, Receiver, Sender};
use futures::{
    future::{lazy, ok, Future},
    Async, Poll,
};
use tokio;
use tokio_timer::sleep;

use std::time::Duration;

pub struct Service {
    rx: Receiver<bool>,
}

impl Service {
    fn new(rx: Receiver<bool>) -> Self {
        Service { rx }
    }
}
impl Future for Service {
    type Item = ();
    type Error = ();
    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        print!(".");
        match self.rx.try_recv() {
            Ok(_r) => {
                if let Some(_v) = _r {
                    return Ok(Async::Ready(()));
                }
                futures::task::current().notify();
                Ok(Async::NotReady)
            }
            Err(_) => Ok(Async::NotReady),
        }
    }
}

fn start(rx: Receiver<bool>) -> impl Future<Item = (), Error = ()> {
    Service::new(rx).and_then(|_| ok(()))
}

fn stop(tx: Sender<bool>) -> impl Future<Item = (), Error = ()> {
    sleep(Duration::from_millis(1))
        .and_then(move |_| {
            print!("\nStopping\n");
            tx.send(false).unwrap();
            ok(())
        })
        .map_err(|_| ())
}

fn main() {
    tokio::run(lazy(|| {
        let (tx, rx) = channel();
        print!("Starting\n");
        tokio::spawn(start(rx));
        print!("Started\n");
        tokio::spawn(stop(tx));
        ok(())
    }));
}