use std::collections::HashMap;
use std::sync::Arc;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::thread;
use std::time::{Duration, SystemTime};

use log::{debug, error, info};
use ws::{CloseCode, Handler, Sender as WsCmd};

use crate::err::{OkResult, ZbusErr};
use crate::message::{Message, Request, Response};
use crate::rpc::IOHandlers;
use crate::wsocket::{Instruct, WsClientHandler};

//wsClient-> wsconnect
pub struct WsClient {
    tx: Sender<Instruct>,
    url: &'static str,
}

impl WsClient {
    /// rpcServer 会用到处理request
    pub fn connect(url: &'static str, process: Option<IOHandlers>) -> Result<Self, ZbusErr> {
        let (tx, rx) = channel();
        handle_instruct(url, rx, tx.clone(), process);
        tx.send(Instruct::Connect);
        Ok(Self {
            tx: tx.clone(),
            url,
        })
    }
    //TODO  后面还可以再改进成，发指令再连接
    pub fn reconnect(&self) -> OkResult {
        if self.is_close() {
            self.tx.send(Instruct::Connect);
            Ok(())
        } else {
            Err(ZbusErr::err("reconnect is err"))
        }
    }

    pub fn is_close(&self) -> bool {
        let (tx, rx) = channel();
        self.tx.send(Instruct::IsClose(Some(tx)));
        if let Ok(result) = rx.recv_timeout(Duration::from_secs(20)) {
            if let Ok(()) = result {
                return true;
            };
        };
        false
    }
    pub fn handler(&self) -> WsClientHandler {
        WsClientHandler {
            tx: self.tx.clone(),
        }
    }
}


impl Drop for WsClient {
    fn drop(&mut self) {
//在这里关闭连接
        self.tx.send(Instruct::Close(None));
        self.tx.send(Instruct::Exit);
        error!("client is over");
        thread::sleep(Duration::from_secs(3));
    }
}


fn handle_instruct(url: &'static str, rx: Receiver<Instruct>, tx: Sender<Instruct>, procoss: Option<IOHandlers>) {
    thread::spawn(move || {
        let mut queue: HashMap<String, (SystemTime, Response)> = HashMap::new();
        let reconnect_num = 10;
        let mut ws_cmd: Option<WsCmd> = None;
        'a: loop {
            let tx = tx.clone();
            if let Ok(instruct) = rx.recv_timeout(Duration::from_secs(60)) {
                match instruct {
                    Instruct::Delivery(msg, sender) => {//把response放入队列中
                        let result: OkResult = ws_cmd.as_ref().map_or_else(|| Err(ZbusErr::closed()), |cmd| {
                            let msg = match msg {
                                Message::Request(ref req) => serde_json::to_string(req),
                                Message::Response(ref resp) => serde_json::to_string(resp),
                            };
                            msg.map_or_else(|_| Err(ZbusErr::validate("msg data err")), |json| {
                                debug!("deliver msg is {}", json);
                                cmd.send(json.clone()).map_err(|e| {
                                    error!("deliver msg  {} {} fail", json, e.to_string());
                                    if let ws::ErrorKind::Io(_) = e.kind {
                                        tx.send(Instruct::Closed);
                                        ZbusErr::closed()
                                    } else {
                                        ZbusErr::err(e.to_string())
                                    }
                                })
                            })
                        });
                        sender.map(|sender| sender.send(result));
                    }
                    Instruct::Receive(msg) => {//把response放入队列中
                        match msg {
                            Message::Request(req) => {
                                debug!("handle rpc request");
                                match req.id() {
                                    None => debug!("bad request ,no request id {}", serde_json::to_string(&req).unwrap()),
                                    Some(_) => { procoss.as_ref().map(|io_handler| io_handler.handler_request(req, tx.clone())); }
                                };
                            }// 启动rpcServer
                            Message::Response(resp) => {
                                if let Some(id) = resp.id() {
                                    debug!("put resp {}", id);
                                    queue.insert(String::from(id), (SystemTime::now(), resp));
                                } else {
                                    debug!("error resp {}", serde_json::to_string(&resp).unwrap());
                                }
                            }
                        };
                    }
                    Instruct::Connect => {//连接服务器
                        //连接成功/失败处理
                        //循环处理次数，可10秒重连一次
                        for i in 0..reconnect_num {
                            let (sender, rx) = channel();
                            let tx = tx.clone();
                            thread::spawn(move || {
                                let result = ws::connect(url, |out| MessageHandle {
                                    ws_client: out,
                                    sender: sender.clone(),
                                    tx: tx.clone(),
                                });
                                if let Err(e) = result {
                                    error!("ws client is close exit reason {}", e);
                                }
                                error!("ws client is close");
                            });
                            if let Ok(ws_client) = rx.recv_timeout(Duration::from_secs(30)) {
                                 ws_cmd = Some(ws_client);
                                continue 'a;
                            } else {
                                thread::sleep(Duration::from_secs(10));
                                error!("{} secs {} nums reconnect", 10, i);
                            }
                        }
                        error!("重新连接失败");
                    }
                    Instruct::Response(msg_id, sender) => {//查询response
                        let result = queue.remove(&msg_id).map_or(Err(ZbusErr::err("未发现")), |(_, resp)| Ok(resp));
                        sender.map(|sender| sender.send(result));
                    }
                    Instruct::Close(sender) => {//关闭连接
                        ws_cmd.as_ref().map(|cmd| {
                            cmd.close(CloseCode::Normal);
                            tx.send(Instruct::Closed);//关闭成功失败都认为关闭成功
                        });
                        sender.map(|sender| sender.send(Ok(())));
                    }
                    Instruct::Closed => {//连接已经关闭
                        ws_cmd = None;
                    }
                    Instruct::IsClose(sender) => {//查询连接是否关闭
                        let result = ws_cmd.as_ref().map_or_else(|| Err(ZbusErr::closed()), |cmd| {
                            cmd.ping(vec![]).map_or_else(|e| {
                                if let ws::ErrorKind::Io(_) = e.kind {
                                    tx.send(Instruct::Closed);
                                    Err(ZbusErr::closed())//关闭
                                } else {
                                    Ok(())
                                }
                            }, |_| Ok(()))
                        });
                        sender.map(|sender| sender.send(result));
                    }
                    Instruct::Exit => {
                        break;
                    }
                    _ => {}
                };
            } else {
                tx.send(Instruct::IsClose(None));
                for (id, (timeout, resp)) in &queue {
                    timeout.elapsed().map(|elapsed| if elapsed.as_secs() > 30 {
                        tx.send(Instruct::Response(id.into(), None));
                    });
                }
            }
        }
        error!(" handle Instruct over,resp nums {} not get", queue.len());
    });
}

struct MessageHandle {
    ws_client: WsCmd,
    tx: Sender<Instruct>,
    sender: Sender<WsCmd>,
}

impl Handler for MessageHandle {
    fn on_open(&mut self, shake: ws::Handshake) -> ws::Result<()> {
        if let Some(addr) = shake.remote_addr()? {
            debug!("Connection with {} now open", addr);
        }
        dbg!("ws connect");
        // self.tx.send(Instruct::Connected(self.ws_client.clone()));
        self.sender.send(self.ws_client.clone());
        Ok(())
    }
    fn on_message(&mut self, msg: ws::Message) -> ws::Result<()> {
        let json = msg.to_string();
        let msg = if json.contains(r#""status":null"#) {
            debug!("request");
            serde_json::from_str::<Request>(&json).map(|msg| Message::Request(msg))
        } else {
            debug!("response");
            serde_json::from_str::<Response>(&json).map(|msg| Message::Response(msg))
        };
        // debug!("{}", json);
        msg.map_or_else(|e| debug!(" recv message err is {} [{}]", e, json), |msg| { self.tx.send(Instruct::Receive(msg)); });

        Ok(())
    }
    fn on_close(&mut self, code: CloseCode, reason: &str) {
        dbg!("ws client is closed");
        self.tx.send(Instruct::Closed);
        //send close
    }
}