ipblc/old/zmqcom.rs

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:?}")
        }
    };
}