example.rs

extern crate futures;
extern crate tokio;

use std::time::Duration;
use std::sync::Arc;
use tokio::timer::Delay;
use std::sync::Mutex;
use std::sync::Condvar;
use tokio::spawn;
use futures::future::lazy;
use futures::Future;
use std::time::Instant;
use num_cpus;
use tokio::runtime::Builder;

const REQUESTS_COUNT: u64 = 100;
const PROCESSING_DURATION: Duration = Duration::from_millis(100);

fn main() {
    let cores = num_cpus::get();
    let mut runtime = Builder::new().core_threads(2).build().expect("failed to start new Runtime");
    runtime.spawn(lazy(|| {
        let sync = Arc::new((Mutex::new(0), Condvar::new()));
        for id in 0..REQUESTS_COUNT {
            println!("send request {}", id);
            let start = Instant::now();
            let task_sync = sync.clone();
            let task =  Delay::new(Instant::now() + PROCESSING_DURATION)
                .map_err(|e| { println!("err {}", e); })
                .map(move |_| {
                    println!("request {} takes {:?}", id, start.elapsed());
                })
                .map(move |_| {
                    let &(ref lock, ref cvar) = &*task_sync;
    
                    let mut processed = lock.lock().unwrap();
                    *processed += 1;
    
                    cvar.notify_one();
                });
            spawn(task);
        }
    
        println!("wait for completion");
        let &(ref lock, ref cvar) = &*sync;
        let mut processed = lock.lock().unwrap();
        println!("requests processed {} / {}", *processed, REQUESTS_COUNT);
        while *processed != REQUESTS_COUNT {
            processed = cvar.wait(processed).unwrap();
            println!("requests processed {} / {}", *processed, REQUESTS_COUNT);
        }
        println!("All requests processed");
        Ok(())
    }));
    runtime.shutdown_on_idle().wait().unwrap();
}