use std::time::{Instant, Duration};
use tokio::{
    net::{TcpListener, TcpStream},
    timer::{Delay, Interval},
    runtime::Runtime,
};
use tokio_io::AsyncRead;

use futures::{
    prelude::*,
    sync::oneshot,
};

fn handle_client(mut client: TcpStream) -> impl Future<Item = (), Error = ()> {
    let mut buf = Vec::with_capacity(32);
    while let Ok(Async::Ready(_len)) = client.read_buf(&mut buf) {
    println!("Handling client!");
    }
    Ok(()).into_future()

}

fn dummy_service() -> impl Future<Item = (), Error = ()> {
    // dummy service that finishes after 1 second
    Delay::new(Instant::now() + Duration::from_secs(1))
        .map_err(drop)
}

fn server() -> impl Future<Item = (), Error = ()> {
    let addr = "127.0.0.1:1337".parse().unwrap();

    let listener = TcpListener::bind(&addr)
        .expect("Failed to bind");

    let accept_loop = listener.incoming()
        .for_each(|client| {
            tokio::spawn(handle_client(client));
            Ok(())
        })
        .map_err(drop);

    accept_loop
}

fn clients() -> impl Future<Item = (), Error = ()> {
    let addr = "127.0.0.1:1337".parse().unwrap();

    // creates a client every 100ms
    Interval::new_interval(Duration::from_millis(100))
        .map_err(drop)
        .for_each(move |_| {
            TcpStream::connect(&addr)
                .map(drop)
                .map_err(drop)
        })

}

fn main() {
    let mut rt = Runtime::new().unwrap();

    rt.spawn(clients());

    let (tx, rx) = oneshot::channel();

    let service = dummy_service()
        .and_then(|_| tx.send(()));

    rt.spawn(service);

    let server_with_shutdown = server()
        .select(rx.map_err(drop))
        .map(drop)
        .map_err(drop);

    let _ = rt.block_on(server_with_shutdown);
}