//! This crate provides a drop in replacement for the `TcpClient` and `TcpServer` from
//! `tokio-proto` that operates over Unix domain sockets instead of a TCP connection. This crate
//! is missing the `threads()` method from `TcpServer` since Unix domain sockets do not support
//! `SO_REUSEADDR`.

#![cfg(unix)]

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

use std::io;
use std::path::{Path, PathBuf};
use std::marker::PhantomData;
use std::sync::Arc;

use futures::stream::Stream;
use futures::future::{Then, Future};
use tokio_core::reactor::{Core, Handle};
use tokio_proto::{BindClient, BindServer};
use tokio_service::{NewService, Service};
use tokio_uds::{UnixListener, UnixStream};

/// A builder for Unix Socket servers.
///
/// Setting up a server needs, at minimum:
///
/// - A server protocol implementation
/// - An path
/// - A service to provide
///
/// In addition to those basics, the builder provides some additional
/// configuration, which is expected to grow over time.
///
/// See the crate docs for an example.
pub struct UnixServer<Kind, P> {
    _kind: PhantomData<Kind>,
    proto: Arc<P>,
    path: PathBuf,
}

impl<Kind, P> UnixServer<Kind, P>
    where P: BindServer<Kind, UnixStream> + Send + Sync + 'static
{
    /// Starts building a server for the given protocol and path, with
    /// default configuration.
    ///
    /// Generally, a protocol is implemented *not* by implementing the
    /// `BindServer` trait directly, but instead by implementing one of the
    /// protocol traits:
    ///
    /// - `tokio_proto::pipeline::ServerProto`
    /// - `tokio_proto::multiplex::ServerProto`
    /// - `tokio_proto::streaming::pipeline::ServerProto`
    /// - `tokio_proto::streaming::multiplex::ServerProto`
    ///
    /// See the crate documentation for more details on those traits.
    pub fn new<T>(protocol: P, path: T) -> UnixServer<Kind, P>
        where T: AsRef<Path>
    {
        UnixServer {
            _kind: PhantomData,
            proto: Arc::new(protocol),
            path: path.as_ref().into(),
        }
    }

    /// Set the path for the server.
    pub fn path<T>(&mut self, path: T)
        where T: AsRef<Path>
    {
        self.path = path.as_ref().into();
    }

    /// Start up the server, providing the given service on it.
    ///
    /// This method will block the current thread until the server is shut down.
    pub fn serve<S>(&self, new_service: S)
        where S: NewService + Send + Sync + 'static,
              S::Instance: 'static,
              P::ServiceError: 'static,
              P::ServiceResponse: 'static,
              P::ServiceRequest: 'static,
              S::Request: From<P::ServiceRequest>,
              S::Response: Into<P::ServiceResponse>,
              S::Error: Into<P::ServiceError>
    {
        let new_service = Arc::new(new_service);
        self.with_handle(move |_| new_service.clone())
    }

    /// Start up the server, providing the given service on it, and providing
    /// access to the event loop handle.
    ///
    /// The `new_service` argument is a closure that is given an event loop
    /// handle, and produces a value implementing `NewService`. That value is in
    /// turn used to make a new service instance for each incoming connection.
    ///
    /// This method will block the current thread until the server is shut down.
    pub fn with_handle<F, S>(&self, new_service: F)
        where F: Fn(&Handle) -> S + Send + Sync + 'static,
              S: NewService + Send + Sync + 'static,
              S::Instance: 'static,
              P::ServiceError: 'static,
              P::ServiceResponse: 'static,
              P::ServiceRequest: 'static,
              S::Request: From<P::ServiceRequest>,
              S::Response: Into<P::ServiceResponse>,
              S::Error: Into<P::ServiceError>
    {
        let proto = self.proto.clone();
        let new_service = Arc::new(new_service);
        let path = self.path.clone();

        serve(proto, path, &*new_service);
    }
}

fn serve<P, Kind, F, S, T>(binder: Arc<P>, path: T, new_service: &F)
    where P: BindServer<Kind, UnixStream>,
          F: Fn(&Handle) -> S,
          S: NewService + Send + Sync,
          T: AsRef<Path>,
          S::Instance: 'static,
          P::ServiceError: 'static,
          P::ServiceResponse: 'static,
          P::ServiceRequest: 'static,
          S::Request: From<P::ServiceRequest>,
          S::Response: Into<P::ServiceResponse>,
          S::Error: Into<P::ServiceError>
{
    struct WrapService<S, Request, Response, Error> {
        inner: S,
        _marker: PhantomData<fn() -> (Request, Response, Error)>,
    }

    impl<S, Request, Response, Error> Service for WrapService<S, Request, Response, Error>
        where S: Service,
              S::Request: From<Request>,
              S::Response: Into<Response>,
              S::Error: Into<Error>
    {
        type Request = Request;
        type Response = Response;
        type Error = Error;
        type Future = Then<S::Future,
             Result<Response, Error>,
             fn(Result<S::Response, S::Error>) -> Result<Response, Error>>;

        fn call(&self, req: Request) -> Self::Future {
            fn change_types<A, B, C, D>(r: Result<A, B>) -> Result<C, D>
                where A: Into<C>,
                      B: Into<D>
            {
                match r {
                    Ok(e) => Ok(e.into()),
                    Err(e) => Err(e.into()),
                }
            }

            self.inner.call(S::Request::from(req)).then(change_types)
        }
    }

    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let new_service = new_service(&handle);
    let listener = UnixListener::bind(path, &handle).unwrap();

    let server = listener.incoming().for_each(move |(socket, _)| {
        // Create the service
        let service = try!(new_service.new_service());

        // Bind it!
        binder.bind_server(&handle,
                           socket,
                           WrapService {
                               inner: service,
                               _marker: PhantomData,
                           });

        Ok(())
    });

    core.run(server).unwrap();
}

/// Builds client connections to external services.
///
/// To connect to a service, you need a *client protocol* implementation; see
/// the crate documentation for guidance.
pub struct UnixClient<Kind, P> {
    _kind: PhantomData<Kind>,
    proto: Arc<P>,
}

impl<Kind, P> UnixClient<Kind, P>
    where P: BindClient<Kind, UnixStream>
{
    /// Create a builder for the given client protocol.
    ///
    /// To connect to a service, you need a *client protocol* implementation;
    /// see the crate documentation for guidance.
    pub fn new(protocol: P) -> UnixClient<Kind, P> {
        UnixClient {
            _kind: PhantomData,
            proto: Arc::new(protocol),
        }
    }

    /// Establish a connection to the given path.
    pub fn connect<T>(&self, path: T, handle: &Handle) -> io::Result<P::BindClient>
        where T: AsRef<Path>
    {
        Ok(self.proto.bind_client(handle, UnixStream::connect(path, handle)?))
    }
}