107 lines
3.1 KiB
Rust
107 lines
3.1 KiB
Rust
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<zmq::Socket, zmq::Error> {
|
|
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<RwLock<Context>>, ipeventtx: &Sender<IpEvent>) -> 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<RwLock<Context>>, txpubsub: Sender<IpEvent>, 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<String> = 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<String>) {
|
|
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:?}")
|
|
}
|
|
};
|
|
}
|