use futures::{prelude::*, future, stream};
use tokio::timer::Delay;
use std::time::{Duration, Instant};
use std::sync::{Mutex, Arc};

extern crate either;
use either::*;

#[derive(Debug)]
enum Error {
    UploadFailed (u32),
    Timeout,
}

fn upload(val: u32) -> impl Future<Item = u32, Error = Error> {
    println!("uploading {}", val);

    // simulating 100ms upload time
    Delay::new(Instant::now() + Duration::from_millis(100))
        .map_err(drop)
        .then(move |_| {
            if val == 20 {
                future::err::<u32, Error>(Error::UploadFailed (val))
            } else {
                future::ok::<u32, Error>(val)
            }
        })
}

fn do_stuffs(vals: impl Iterator<Item = u32>) -> impl Future<Item = (), Error = Error> {
    stream::iter_ok(vals)
        .and_then(upload)
        .for_each(|val| {
            println!("Done uploading {}", val);
            Ok(())
        })
}

fn main() {
    let uploads = do_stuffs(0..42);

    let timeout = Delay::new(Instant::now() + Duration::from_secs(3))
        .then(|_| {
            println!("Timeout reached!");
            future::err::<(), Error>(Error::Timeout)
        });

    let work = uploads.select(timeout)
        .map(|x| x.0)
        .map_err(|x| x.0);


    let mut runtime = tokio::runtime::Runtime::new().unwrap();
    let result = runtime.block_on(work);
    println!("Result: {:?}", result);
}