use crate::config::{Context, ZMQCfg}; use crate::ip::IpEvent; use crate::utils::gethostname; use std::sync::Arc; use tokio::sync::mpsc::Sender; use tokio::sync::RwLock; const ZMQPROTO: &str = "tcp"; pub async fn zconnect<'a>( zmqcfg: &ZMQCfg, zmqtype: zmq::SocketType, ) -> Result { let zctx = zmq::Context::new(); let zmqhost = &zmqcfg.hostname; let zmqport = &zmqcfg.port; let socket = zctx.socket(zmqtype).unwrap(); let connectstring = format!("{ZMQPROTO}://{zmqhost}:{zmqport}"); socket.connect(&connectstring.as_str())?; Ok(socket) } pub async fn zmqinit(ctx: &Arc>, ipeventtx: &Sender) -> zmq::Socket { let ctxarc = Arc::clone(&ctx); let zmqreqsocket; let zmqsubsocket; { let zmqctx = ctxarc.read().await; zmqreqsocket = zconnect(&zmqctx.cfg.zmq.get("reqrep").unwrap(), zmq::REQ) .await .unwrap(); zmqsubsocket = zconnect(&zmqctx.cfg.zmq.get("pubsub").unwrap(), zmq::SUB) .await .unwrap(); } listenpubsub(&ctx, ipeventtx.clone(), zmqsubsocket).await; return zmqreqsocket; } async fn listenpubsub(ctx: &Arc>, txpubsub: Sender, socket: zmq::Socket) { let prefix; { let ctx = ctx.read().await; prefix = format!( "{sub} ", sub = ctx.cfg.zmq.get("pubsub").unwrap().subscription ); socket .set_subscribe(ctx.cfg.zmq.get("pubsub").unwrap().subscription.as_bytes()) .expect("failed setting subscription"); } tokio::spawn(async move { loop { let msgs: Option = match socket.recv_string(0) { Ok(s) => match s { Ok(ss) => Some(ss), Err(e) => { println!("{e:?}"); None } }, Err(e) => { println!("{e:?}"); None } }; match msgs { Some(ss) => { let msg = ss.strip_prefix(prefix.as_str()).unwrap(); let tosend: IpEvent = serde_json::from_str(msg).unwrap(); if tosend.ipdata.hostname != gethostname(true) || tosend.msgtype == "init".to_string() { txpubsub.send(tosend).await.unwrap(); } } None => {} }; } }); } pub async fn send_to_ipbl_zmq(reqsocket: &zmq::Socket, ip: &IpEvent, _ret: &mut Vec) { let msg = format!("{val}", val = serde_json::to_string(&ip).unwrap()); match reqsocket.send(&msg, 0) { Ok(_) => {} Err(e) => { println!("{e:?}") } }; match reqsocket.recv_string(0) { Ok(o) => match o { Ok(_) => {} Err(ee) => { println!("{ee:?}") } }, Err(e) => { println!("{e:?}") } }; }