/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2023 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ use std::collections::hash_map::Entry; use async_std::sync::Arc; use log::{debug, error, info}; use serde_json::Value; use smol::Executor; use url::Url; use darkfi::util::{async_util, time::NanoTimestamp}; use crate::{ config::{DnvConfig, Node, NodeType}, error::{DnetViewError, DnetViewResult}, model::{ LilithInfo, Model, NetworkInfo, NodeInfo, SelectableObject, Session, SessionInfo, SlotInfo, }, rpc::RpcConnect, util::{make_empty_id, make_info_id, make_node_id, make_session_id}, }; pub struct DataParser { model: Arc, config: DnvConfig, } impl DataParser { pub fn new(model: Arc, config: DnvConfig) -> Arc { Arc::new(Self { model, config }) } pub async fn start_connect_slots(self: Arc, ex: Arc>) -> DnetViewResult<()> { debug!(target: "dnetview", "start_connect_slots() START"); for node in &self.config.nodes { debug!(target: "dnetview", "attempting to spawn..."); ex.clone().spawn(self.clone().try_connect(node.clone())).detach(); } Ok(()) } async fn try_connect(self: Arc, node: Node) -> DnetViewResult<()> { debug!(target: "dnetview", "try_connect() START"); loop { info!("Attempting to poll {}, RPC URL: {}", node.name, node.rpc_url); // Parse node config and execute poll. // On any failure, sleep and retry. match RpcConnect::new(Url::parse(&node.rpc_url)?, node.name.clone()).await { Ok(client) => { if let Err(e) = self.poll(&node, client).await { error!("Poll execution error: {:?}", e); } } Err(e) => { error!("RPC client creation error: {:?}", e); } } self.parse_offline(node.name.clone()).await?; async_util::sleep(2000).await; } } async fn poll(&self, node: &Node, client: RpcConnect) -> DnetViewResult<()> { loop { // Ping the node to verify if its online. if let Err(e) = client.ping().await { return Err(DnetViewError::Darkfi(e)) } // Retrieve node info, based on its type let response = match &node.node_type { NodeType::LILITH => client.lilith_spawns().await, NodeType::NORMAL => client.dnet_info().await, NodeType::CONSENSUS => client.get_consensus_info().await, }; // Parse response match response { Ok(reply) => { debug!("dnetview:: poll() reply {:?}", reply); if reply.as_object().is_none() || reply.as_object().unwrap().is_empty() { return Err(DnetViewError::EmptyRpcReply) } match &node.node_type { NodeType::LILITH => { self.parse_lilith_data( reply.as_object().unwrap().clone(), node.name.clone(), ) .await? } _ => self.parse_data(reply.as_object().unwrap(), node.name.clone()).await?, }; } Err(e) => return Err(e), } // Sleep until next poll async_util::sleep(2000).await; } } // If poll times out, inititalize data structures with empty values. async fn parse_offline(&self, node_name: String) -> DnetViewResult<()> { debug!(target: "dnetview", "parse_offline() START"); let name = "Offline".to_string(); let session_type = Session::Offline; let mut sessions: Vec = Vec::new(); let hosts = Vec::new(); let node_id = make_node_id(&node_name)?; let dnet_id = make_empty_id(&node_id, &session_type, 0)?; let addr = "Null".to_string(); let state = None; let random_id = 0; let remote_id = "Null".to_string(); let log = Vec::new(); let is_empty = true; let slot = SlotInfo::new( dnet_id.clone(), node_id.clone(), addr.clone(), random_id, remote_id, log, is_empty, ); let session_info = SessionInfo::new( dnet_id, node_id.clone(), //name.clone(), addr.clone(), state, slot, is_empty, ); sessions.push(session_info); // TODO: clean this up let node = NodeInfo::new( node_id.clone(), name.clone(), hosts, sessions.clone(), sessions.clone(), is_empty, ); self.update_selectables(node).await?; Ok(()) } async fn parse_data( &self, reply: &serde_json::Map, name: String, ) -> DnetViewResult<()> { let hosts = &reply["hosts"]; let inbound = &reply["inbound"]; let outbound = &reply["outbound"]; let node_id = make_node_id(&name)?; let hosts = self.parse_hosts(hosts).await?; let inbound = self.parse_session(inbound, &node_id, Session::Inbound).await?; let outbound = self.parse_session(outbound, &node_id, Session::Outbound).await?; let node = NodeInfo::new(node_id, name, hosts, inbound.clone(), outbound.clone(), false); self.update_selectables(node).await?; self.update_msgs(inbound.clone(), outbound.clone()).await?; Ok(()) } async fn parse_lilith_data( &self, reply: serde_json::Map, name: String, ) -> DnetViewResult<()> { let spawns: Vec> = serde_json::from_value(reply.get("spawns").unwrap().clone()).unwrap(); let mut networks = vec![]; for spawn in spawns { let name = spawn.get("name").unwrap().as_str().unwrap().to_string(); let id = make_node_id(&name)?; let urls: Vec = serde_json::from_value(spawn.get("urls").unwrap().clone()).unwrap(); let nodes: Vec = serde_json::from_value(spawn.get("hosts").unwrap().clone()).unwrap(); let network = NetworkInfo::new(id, name, urls, nodes); networks.push(network); } let id = make_node_id(&name)?; let lilith = LilithInfo::new(id.clone(), name, networks); let lilith_obj = SelectableObject::Lilith(lilith.clone()); self.model.selectables.lock().await.insert(id, lilith_obj); for network in lilith.networks { let network_obj = SelectableObject::Network(network.clone()); self.model.selectables.lock().await.insert(network.id, network_obj); } Ok(()) } async fn update_msgs( &self, inbounds: Vec, outbounds: Vec, ) -> DnetViewResult<()> { for inbound in inbounds { if !self.model.msg_map.lock().await.contains_key(&inbound.info.dnet_id) { // we don't have this ID: it is a new node self.model .msg_map .lock() .await .insert(inbound.info.dnet_id, inbound.info.log.clone()); } else { // we have this id: append the msg values match self.model.msg_map.lock().await.entry(inbound.info.dnet_id) { Entry::Vacant(e) => { e.insert(inbound.info.log); } Entry::Occupied(mut e) => { for msg in inbound.info.log { e.get_mut().push(msg); } } } } } for outbound in outbounds { if !self.model.msg_map.lock().await.contains_key(&outbound.info.dnet_id) { // we don't have this ID: it is a new node self.model .msg_map .lock() .await .insert(outbound.info.dnet_id, outbound.info.log.clone()); } else { // we have this id: append the msg values match self.model.msg_map.lock().await.entry(outbound.info.dnet_id) { Entry::Vacant(e) => { e.insert(outbound.info.log); } Entry::Occupied(mut e) => { for msg in outbound.info.log { e.get_mut().push(msg); } } } } } Ok(()) } async fn update_selectables(&self, node: NodeInfo) -> DnetViewResult<()> { if node.is_offline { let node_obj = SelectableObject::Node(node.clone()); self.model.selectables.lock().await.insert(node.dnet_id.clone(), node_obj.clone()); } else { let node_obj = SelectableObject::Node(node.clone()); self.model.selectables.lock().await.insert(node.dnet_id.clone(), node_obj.clone()); for inbound in node.inbound { if !inbound.is_empty { let inbound_obj = SelectableObject::Session(inbound.clone()); self.model .selectables .lock() .await .insert(inbound.clone().dnet_id, inbound_obj.clone()); let info_obj = SelectableObject::Connect(inbound.info.clone()); self.model .selectables .lock() .await .insert(inbound.info.clone().dnet_id, info_obj.clone()); } } for outbound in node.outbound { if !outbound.is_empty { let outbound_obj = SelectableObject::Session(outbound.clone()); self.model .selectables .lock() .await .insert(outbound.clone().dnet_id, outbound_obj.clone()); let info_obj = SelectableObject::Connect(outbound.info.clone()); self.model .selectables .lock() .await .insert(outbound.info.clone().dnet_id, info_obj.clone()); } } } Ok(()) } async fn parse_session( &self, reply: &Value, node_id: &String, prefix: Session, ) -> DnetViewResult> { let session_id = make_session_id(&node_id, &prefix)?; let mut session_info: Vec = Vec::new(); // TODO: improve this ugly hack. let mut slot_count = 0; // Dnetview is not enabled. if reply.is_null() { debug!(target: "dnetview", "parse_outbound() reply.is_null() == True"); slot_count += 1; let info_id = make_empty_id(&node_id, &prefix, slot_count)?; let node_id = node_id.to_string(); let addr = "Null".to_string(); let random_id = 0; let remote_id = "Null".to_string(); let log = Vec::new(); let is_empty = false; let slot = SlotInfo::new( info_id.clone(), node_id.clone(), addr, random_id, remote_id, log, is_empty, ); let is_empty = true; let addr = "Null".to_string(); let state = None; let session = SessionInfo::new(session_id.clone(), node_id.clone(), addr, state, slot, is_empty); session_info.push(session); return Ok(session_info) } let sessions = reply.as_array().unwrap(); debug!(target: "dnetview", "parse_outbound() len session{:?}", sessions.len()); for session in sessions { match session.as_object() { Some(obj) => { debug!(target: "dnetview", "parse_outbound() obj {:?}", session); let addr = obj.get("addr").unwrap().as_str().unwrap().to_string(); let state: Option = match obj.get("state") { Some(state) => Some(state.as_str().unwrap().to_string()), None => None, }; let info: serde_json::Map = serde_json::from_value(obj.get("info").unwrap().clone()).unwrap(); let slot_addr = info.get("addr").unwrap().as_str().unwrap().to_string(); let random_id = info.get("random_id").unwrap().as_u64().unwrap(); let remote_id = info.get("remote_id").unwrap().as_str().unwrap().to_string(); let info_id = make_info_id(&random_id)?; let log: Vec<(NanoTimestamp, String, String)> = serde_json::from_value(info.get("log").unwrap().clone()).unwrap(); // ... let node_id = node_id.to_string(); let is_empty = false; let slot = SlotInfo::new( info_id.clone(), node_id.clone(), slot_addr, random_id, remote_id, log, is_empty, ); let session = SessionInfo::new( session_id.clone(), node_id.clone(), addr.clone(), state, slot, is_empty, ); session_info.push(session); } None => { // TODO: clean up empty info boilerplate. slot_count += 1; let info_id = make_empty_id(node_id, &prefix, slot_count)?; let node_id = node_id.to_string(); let addr = "Null".to_string(); let random_id = 0; let remote_id = "Null".to_string(); let log = Vec::new(); let is_empty = true; let slot = SlotInfo::new( info_id.clone(), node_id.clone(), addr.clone(), random_id, remote_id, log, is_empty, ); let is_empty = true; let state = None; let session = SessionInfo::new( session_id.clone(), node_id.clone(), addr.clone(), state, slot, is_empty, ); session_info.push(session); } } } Ok(session_info) } async fn parse_hosts(&self, hosts: &Value) -> DnetViewResult> { match hosts.as_array() { Some(h) => match h.is_empty() { true => Ok(Vec::new()), false => { let hosts: Vec = h.iter().map(|addr| addr.as_str().unwrap().to_string()).collect(); Ok(hosts) } }, None => { if hosts.is_null() { // TODO: this should probs just say null let h = Vec::new(); return Ok(h) } debug!("dnetview::parse_hosts() hosts returns None and !is_null() {}", hosts); Err(DnetViewError::ValueIsNotObject) } } } }