extern crate futures;
extern crate tokio_core;
extern crate tokio_proto;
extern crate tokio_service;

use std::str;
use std::io::{self, ErrorKind, Write};

use futures::{future, Future, BoxFuture};
use tokio_core::io::{Io, Codec, Framed, EasyBuf};
use tokio_proto::TcpServer;
use tokio_proto::pipeline::ServerProto;
use tokio_service::Service;


#[derive(Default)]
pub struct IntCodec;

fn parse_u64(from: &[u8])-> Result<u64, io::Error> {
    Ok(str::from_utf8(from)
        .map_err(|e| io::Error::new(ErrorKind::InvalidData, e))?
        .parse()
        .map_err(|e| io::Error::new(ErrorKind::InvalidData, e))?
    )
}

impl Codec for IntCodec {
    type In = u64;
    type Out = u64;

    fn decode(&mut self, buf: &mut EasyBuf)-> Result<Option<u64>, io::Error> {
        println!("{:?}, 64: {:?}", buf, "64".as_bytes());
        if let Some(i) = buf.as_slice().iter().position(|&b| b == b'\n') {
            // \r\nの扱いは？
            let full_line = buf.drain_to(i+1);

            let slice = &full_line.as_slice()[..i-1];
            Ok(Some(parse_u64(slice)?))
        } else {Ok(None)}
    }

    fn decode_eof(&mut self, buf: &mut EasyBuf)-> io::Result<u64> {
        let amt = buf.len();
        Ok(parse_u64(buf.drain_to(amt).as_slice())?)
    }

    fn encode(&mut self, item: u64, into: &mut Vec<u8>)-> io::Result<()> {
        write!(into, "{}\n", item);
        Ok(())
    }
}


pub struct IntProto;

impl<T: Io+ 'static>ServerProto<T> for IntProto {
    type Request = u64;
    type Response = u64;
    type Transport = Framed<T, IntCodec>;
    type BindTransport = Result<Self::Transport, io::Error>;

    fn bind_transport(&self, io: T)-> Self::BindTransport {
        Ok(io.framed(IntCodec))
    }
}



pub struct Doubler;

impl Service for Doubler {
    type Request = u64;
    type Response = u64;
    type Error = io::Error;
    type Future = BoxFuture<u64, io::Error>;

    fn call(&self, req: u64)-> Self::Future {
        println!("{:?}", req);
        future::finished(req * 2).boxed()
    }
}

fn main() {
    let addr = "127.0.0.1:8081".parse().unwrap();
    // This method will block the current thread until the server is shut down.
    TcpServer::new(IntProto, addr)
        .serve(|| Ok(Doubler));
}