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};
pub struct WsClient {
tx: Sender<Instruct>,
url: &'static str,
}
impl WsClient {
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,
})
}
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) => { 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) => { 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())); }
};
} 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 => { 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) => { 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.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))
};
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);
}
}