use std::fmt::Debug;
/**
 * rust-daemon
 * Server implementation
 *
 * https://github.com/ryankurte/rust-daemon
 * Copyright 2018 Ryan Kurte
 */
use std::fs;
use std::io::Error as IoError;
use std::sync::{Arc, Mutex};

use futures::sync::mpsc;
use futures::sync::oneshot;

use tokio::prelude::*;
use tokio::spawn;
use tokio_uds::{UnixListener, UnixStream};

use serde::{Deserialize, Serialize};

use uuid::Uuid;

use client::Client;
use error::DaemonError;
use user::User;

/// Server implements a daemon server
/// This creates an IPC socket and listens for messages from connected clients
pub struct Server<REQ, RESP> {
    path: String,
    connections: Arc<Mutex<Vec<Connection<REQ, RESP>>>>,
    incoming: Option<mpsc::UnboundedReceiver<Request<REQ, RESP>>>,
    exit: oneshot::Sender<()>,
}

impl<REQ, RESP> Server<REQ, RESP>
where
    for<'de> REQ: Serialize + Deserialize<'de> + Clone + Send + Debug + 'static,
    for<'de> RESP: Serialize + Deserialize<'de> + Clone + Send + Debug + 'static,
{
    /// Create a new server with the defined Request and Response types
    /// This starts a new listening thread using the provided runtime handle
    /// and provides a Source of Requests from connected clients
    pub fn new(path: String) -> Result<Server<REQ, RESP>, DaemonError> {
        trace!("[daemon server] creating server (socket: {})", path);
        let _res = fs::remove_file(&path);

        // Create listener and client list
        let listener = UnixListener::bind(&path)?;
        let connections = Arc::new(Mutex::new(Vec::new()));

        let (tx, rx) = mpsc::unbounded::<Request<REQ, RESP>>();
        let (exit_tx, exit_rx) = oneshot::channel::<()>();

        // Handle incoming connections
        let client_list = connections.clone();
        let tokio_server = listener
            .incoming()
            .for_each(move |socket| {
                // Fetch user info
                let p = socket.peer_cred().unwrap();
                let u = User::from_uid(p.uid).unwrap();

                // Create client connection
                let mut client = Client::<_, RESP, REQ>::from(socket);

                // Set client thread exit
                let (client_exit_tx, client_exit_rx) = oneshot::channel::<()>();
                client.exit = Some(client_exit_tx);

                let conn = Connection::<REQ, RESP>::new(u, client.clone());

                trace!(
                    "[daemon server] new connection user: '{}' id: '{}'",
                    conn.user.name,
                    conn.id
                );

                // Add to client list
                client_list.lock().unwrap().push(conn.clone());
                let incoming = tx.clone();

                // Handle incoming requests
                let rx_handle = client
                    .for_each(move |req| {
                        trace!("[daemon server] client rx: {:?}", req);
                        let r = Request {
                            inner: conn.clone(),
                            data: req,
                        };
                        incoming.clone().send(r).wait().unwrap();

                        Ok(())
                    })
                    .map_err(|e| panic!("error: {}", e))
                    .select2(client_exit_rx)
                    .then(|_| {
                        info!("[daemon server] closing client handler");
                        Ok(())
                    });
                    
                spawn(rx_handle);

                Ok(())
            })
            .map_err(|err| {
                error!("[daemon server] accept error: {}", err);
            })
            .select2(exit_rx)
            .then(|_| {
                info!("[daemon server] closing listener");
                Ok(())
            });;

        let s = Server {
            path: path,
            connections,
            incoming: Some(rx),
            exit: exit_tx,
        };

        spawn(tokio_server);

        Ok(s)
    }

    pub fn incoming(&mut self) -> Option<mpsc::UnboundedReceiver<Request<REQ, RESP>>> {
        self.incoming.take()
    }

    pub fn close(self) {
        trace!("[daemon server] closing socket server");

        // Send listener exit signal
        let _ = self.exit.send(());

        // Send exit signals to client listeners
        let mut connections = self.connections.lock().unwrap();
        let _results: Vec<_> = connections.drain(0..).map(|c| c.client.exit() ).collect();

        // Remove daemon port
        let _e = fs::remove_file(&self.path);
    }
}

/// Request is a request from a client
/// This contains the Request data and is a Sink for Response data
pub struct Request<REQ, RESP> {
    inner: Connection<REQ, RESP>,
    data: REQ,
}

/// Allows requests to be cloned
impl<REQ, RESP> Request<REQ, RESP>
where
    REQ: Clone,
{
    // Fetch the data for a given request
    pub fn data(&self) -> REQ {
        self.data.clone()
    }
}

/// Sink implementation allows responses to requests
impl<REQ, RESP> Sink for Request<REQ, RESP>
where
    RESP: Clone + Serialize,
{
    type SinkItem = RESP;
    type SinkError = IoError;

    fn start_send(
        &mut self,
        item: Self::SinkItem,
    ) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
        trace!("[daemon request] start send");
        self.inner.client.start_send(item)
    }

    fn poll_complete(&mut self) -> Result<Async<()>, Self::SinkError> {
        trace!("[daemon request] send complete");
        self.inner.client.poll_complete()
    }
}

/// Connection defined a connected client with associated user and a randomly allocated ID
#[derive(Clone)]
struct Connection<REQ, RESP> {
    pub id: String,
    pub user: User,
    pub client: Client<UnixStream, RESP, REQ>,
}

impl<REQ, RESP> Connection<REQ, RESP>
where
    for<'de> REQ: Serialize + Deserialize<'de> + Clone + Send + 'static,
    for<'de> RESP: Serialize + Deserialize<'de> + Clone + Send + 'static,
{
    pub fn new(user: User, client: Client<UnixStream, RESP, REQ>) -> Connection<REQ, RESP> {
        Connection {
            id: Uuid::new_v4().to_string(),
            user,
            client,
        }
    }
}