use core_poll::CorePoll;
use std::net::TcpStream;
use std::thread::spawn;
use std::sync::mpsc::Sender;
use std::sync::mpsc::Receiver;
use std::sync::mpsc::channel;
use std::os::unix::io::AsRawFd;
use std;
use super::threadpool::ThreadPool;
use std::os::unix::io::FromRawFd;
use std::sync::Arc;
use std::sync::RwLock;

pub enum Ctrl {
    ADD(i32),
    RM(i32),
    SHUTDOWN,
}
/// Executor to manage socket poll and execute response to handle message
pub struct Executor {
    ctrl_sender: Sender<Ctrl>,
}

impl Executor {
    pub fn new<T>(maxconn: i32,
                  threadnum: usize,
                  handler: fn(fd: i32,
                              socket: &mut TcpStream,
                              commander: Sender<Ctrl>,
                              data: Arc<RwLock<T>>),
                  data: T)
                  -> Self
        where T: 'static + Send + Sync
    {
        let (tx, rx): (Sender<Ctrl>, Receiver<Ctrl>) = channel();
        let tx_clone = tx.clone();
        let thpool = ThreadPool::new(threadnum);
        spawn(move || {
            let mut epoll = CorePoll::new(maxconn);
            let data = Arc::new(RwLock::new(data));
            loop {
                match rx.try_recv() {
                    Ok(ctrl) => {
                        match ctrl {
                            Ctrl::ADD(fd) => {
                                epoll.add(fd);
                            }
                            Ctrl::RM(fd) => {
                                epoll.remove(fd);
                            }
                            Ctrl::SHUTDOWN => {
                                epoll.close();
                                // drop(epoll);
                                // return;
                            }
                        }
                    }
                    Err(_) => {}
                }

                let list = epoll.wait();

                for fd in list {
                    let cloned = tx_clone.clone();
                    let cloned_data = data.clone();
                    thpool.execute(move || {
                        unsafe {
                            let mut stream: TcpStream = TcpStream::from_raw_fd(fd);
                            handler(fd, &mut stream, cloned, cloned_data);
                            std::mem::forget(stream);
                        }
                    });
                }
            }

        });

        Executor { ctrl_sender: tx }
    }
    /// Add socket to epoll list
    pub fn add(&self, sock: TcpStream) {
        self.ctrl_sender.send(Ctrl::ADD(sock.as_raw_fd())).unwrap();
        std::mem::forget(sock);
    }

    /// Remove socket from epoll listener
    pub fn remove(&self, sock: TcpStream) {
        self.ctrl_sender.send(Ctrl::RM(sock.as_raw_fd())).unwrap();
        std::mem::forget(sock);
    }

    /// Shutdown executor
    pub fn shutwodn(&self) {
        self.ctrl_sender.send(Ctrl::SHUTDOWN).unwrap();
    }
}