extern crate futures;
extern crate tokio_core;
extern crate tokio_io;
extern crate tokio_proto;
extern crate tokio_service;
extern crate tokio_uds_proto;

use futures::{Future, IntoFuture};
use futures::future::FutureResult;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_io::codec::{Framed, LinesCodec};
use tokio_core::reactor::Core;
use tokio_proto::pipeline::{ClientProto, ServerProto};
use tokio_service::Service;
use tokio_uds_proto::{UnixClient, UnixServer};
use std::fs;
use std::io;
use std::thread;
use std::time::Duration;

#[test]
fn test_simple_echo() {
    let mut core = Core::new().unwrap();

    // This brings up our server.
    let path = "./tests/socket";

    thread::spawn(move || {
        UnixServer::new(LineProto, path).serve(|| Ok(EchoService));
    });

    // A bit annoying, but we need to wait for the server to connect
    thread::sleep(Duration::from_millis(100));

    let handle = core.handle();
    let client = UnixClient::new(LineProto).connect(path, &handle).unwrap();
    core.run(client.call("ping".into()).and_then(|response| {
        if response == "ping" {
            Ok(())
        } else {
            Err(io::Error::new(io::ErrorKind::InvalidData, "wrong response"))
        }
    })).unwrap();

    fs::remove_file(path).unwrap();
}

#[test]
#[should_panic(expected = "wrong response")]
fn test_simple_changed_echo() {
    let mut core = Core::new().unwrap();

    // This brings up our server.
    let path = "./tests/socket2";

    thread::spawn(move || {
        UnixServer::new(LineProto, path).serve(|| Ok(EchoService));
    });

    // A bit annoying, but we need to wait for the server to connect
    thread::sleep(Duration::from_millis(100));

    let handle = core.handle();
    let client = UnixClient::new(LineProto).connect(path, &handle).unwrap();
    let res = core.run(client.call("ping".into()).and_then(|response| {
        if response == "pong" {
            Ok(())
        } else {
            Err(io::Error::new(io::ErrorKind::InvalidData, "wrong response"))
        }
    }));

    fs::remove_file(path).unwrap();
    res.unwrap();
}

/// Protocol definition
struct LineProto;

impl<T: AsyncRead + AsyncWrite + 'static> ClientProto<T> for LineProto {
    type Request = String;
    type Response = String;

    /// `Framed<T, LinesCodec>` is the return value of `io.framed(LinesCodec)`
    type Transport = Framed<T, LinesCodec>;
    type BindTransport = Result<Self::Transport, io::Error>;

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

impl<T: AsyncRead + AsyncWrite + 'static> ServerProto<T> for LineProto {
    type Request = String;
    type Response = String;

    /// `Framed<T, LinesCodec>` is the return value of `io.framed(LinesCodec)`
    type Transport = Framed<T, LinesCodec>;
    type BindTransport = Result<Self::Transport, io::Error>;

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

struct EchoService;
impl Service for EchoService {
    type Request = String;
    type Response = String;
    type Error = io::Error;
    type Future = FutureResult<String, io::Error>;

    fn call(&self, req: Self::Request) -> Self::Future {
        Ok(req).into_future()
    }
}