use futures;
use std::{
    io::{Cursor, Error, ErrorKind},
    net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4},
};
use tokio::{
    self,
    codec::{BytesCodec, Framed},
    net::TcpStream,
    prelude::future::Either,
    prelude::*,
};

pub fn stream_map<S, I>(stream: S) -> impl Stream<Item = (I, u32), Error = std::io::Error>
where
    S: Stream<Item = I, Error = std::io::Error>,
    I: AsyncRead + AsyncWrite + 'static,
{
    use std::ops::Add;
    stream
        .map(|client| (Framed::new(client, BytesCodec::new()).split(), 0))
        .and_then(|((w, r), version)| {
            let framed = r.reunite(w).unwrap().into_inner();
            if version == 255 {
                return Box::new(futures::future::err(std::io::Error::from_raw_os_error(0)));
            }
            match version {
                0 => Box::new(futures::future::ok((framed, version))),
                1 => Box::new(futures::future::ok((framed, 5))),
                2 => Box::new(
                    tokio::timer::Delay::new(
                        std::time::Instant::now().add(std::time::Duration::from_secs(1)),
                    )
                    .map_err(|_| std::io::Error::from_raw_os_error(0))
                    .and_then(|_| futures::future::ok((framed, 10))),
                ),
                _ => unreachable!(),
            }
        })
}