thread sleep future

sleep.rs

extern crate futures;

use futures::{Future, Poll, Async};

use std::time::Duration;
use std::thread;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

struct Sleep {
    duration: Duration,
    done_flag: Arc<AtomicBool>,
    handle: Option<thread::JoinHandle<()>>,
}

impl Sleep {
    fn new(duration: Duration) -> Sleep {
        Sleep {
            duration,
            done_flag: Arc::new(AtomicBool::new(false)),
            handle: None,
        }
    }
}

impl futures::Future for Sleep {
    type Item = ();
    type Error = ();

    fn poll(&mut self) -> Poll<(), ()> {
        match self.handle.take() {
            None => {
                let current_task = futures::task::current();
                let flag_clone = self.done_flag.clone();
                let duration = self.duration;
                self.handle = Some(thread::spawn(move || {
                    thread::sleep(duration);
                    flag_clone.store(true, Ordering::Release);
                    current_task.notify();
                }));
                Ok(Async::NotReady)
            }
            Some(handle) => {
                if self.done_flag.load(Ordering::Acquire) {
                    // The sleep is done. Clean up the sleeping thread.
                    handle.join().unwrap();
                    Ok(Async::Ready(()))
                } else {
                    // The sleep isn't done yet. Put the handle back.
                    self.handle = Some(handle);
                    Ok(Async::NotReady)
                }
            }
        }
    }
}

fn main() {
    println!("beginning");
    Sleep::new(Duration::from_secs(1))
        .then(|_| {println!("middle"); Ok::<(), ()>(())})
        .then(|_| Sleep::new(Duration::from_secs(1)))
        .wait()
        .unwrap();
    println!("end");
}