/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2022 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::{
ConnectInfo, LilithInfo, Model, NetworkInfo, NodeInfo, SelectableObject, Session,
SessionInfo,
},
rpc::RpcConnect,
util::{is_empty_session, make_connect_id, make_empty_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.get_info().await,
};
// Parse response
match response {
Ok(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?
}
NodeType::NORMAL => {
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;
}
}
async fn parse_offline(&self, node_name: String) -> DnetViewResult<()> {
let name = "Offline".to_string();
let session_type = Session::Offline;
let node_id = make_node_id(&node_name)?;
let session_id = make_session_id(&node_id, &session_type)?;
let mut connects: Vec = Vec::new();
let mut sessions: Vec = Vec::new();
// initialize with empty values
let id = make_empty_id(&node_id, &session_type, 0)?;
let addr = "Null".to_string();
let state = "Null".to_string();
let parent = node_id.clone();
let msg_log = Vec::new();
let is_empty = true;
let last_msg = "Null".to_string();
let last_status = "Null".to_string();
let remote_node_id = "Null".to_string();
let connect_info = ConnectInfo::new(
id,
addr,
state.clone(),
parent.clone(),
msg_log,
is_empty,
last_msg,
last_status,
remote_node_id,
);
connects.push(connect_info.clone());
let accept_addr = None;
let session_info =
SessionInfo::new(session_id, name, is_empty, parent, connects, accept_addr, None);
sessions.push(session_info);
let node = NodeInfo::new(node_id, node_name, state, sessions.clone(), None, true);
self.update_selectables(sessions, node).await?;
Ok(())
}
async fn parse_data(
&self,
reply: &serde_json::Map,
node_name: String,
) -> DnetViewResult<()> {
let addr = &reply.get("external_addr");
let inbound = &reply["session_inbound"];
let _manual = &reply["session_manual"];
let outbound = &reply["session_outbound"];
let state = &reply["state"];
let mut sessions: Vec = Vec::new();
let node_id = make_node_id(&node_name)?;
let ext_addr = self.parse_external_addr(addr).await?;
let in_session = self.parse_inbound(inbound, &node_id).await?;
let out_session = self.parse_outbound(outbound, &node_id).await?;
//let man_session = self.parse_manual(manual, &node_id).await?;
sessions.push(in_session.clone());
sessions.push(out_session.clone());
//sessions.push(man_session.clone());
let node = NodeInfo::new(
node_id,
node_name,
state.as_str().unwrap().to_string(),
sessions.clone(),
ext_addr,
false,
);
self.update_selectables(sessions.clone(), node).await?;
self.update_msgs(sessions).await?;
//debug!("IDS: {:?}", self.model.ids.lock().await);
//debug!("INFOS: {:?}", self.model.nodes.lock().await);
Ok(())
}
async fn parse_lilith_data(
&self,
reply: serde_json::Map,
name: String,
) -> DnetViewResult<()> {
let urls: Vec = serde_json::from_value(reply.get("urls").unwrap().clone()).unwrap();
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, urls, 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, sessions: Vec) -> DnetViewResult<()> {
for session in sessions {
for connection in session.children {
if !self.model.msg_map.lock().await.contains_key(&connection.id) {
// we don't have this ID: it is a new node
self.model
.msg_map
.lock()
.await
.insert(connection.id, connection.msg_log.clone());
} else {
// we have this id: append the msg values
match self.model.msg_map.lock().await.entry(connection.id) {
Entry::Vacant(e) => {
e.insert(connection.msg_log);
}
Entry::Occupied(mut e) => {
for msg in connection.msg_log {
e.get_mut().push(msg);
}
}
}
}
}
}
Ok(())
}
async fn update_selectables(
&self,
sessions: Vec,
node: NodeInfo,
) -> DnetViewResult<()> {
if node.is_offline {
let node_obj = SelectableObject::Node(node.clone());
self.model.selectables.lock().await.insert(node.id.clone(), node_obj.clone());
} else {
let node_obj = SelectableObject::Node(node.clone());
self.model.selectables.lock().await.insert(node.id.clone(), node_obj.clone());
for session in sessions {
if !session.is_empty {
let session_obj = SelectableObject::Session(session.clone());
self.model
.selectables
.lock()
.await
.insert(session.clone().id, session_obj.clone());
for connect in session.children {
let connect_obj = SelectableObject::Connect(connect.clone());
self.model
.selectables
.lock()
.await
.insert(connect.clone().id, connect_obj.clone());
}
}
}
}
Ok(())
}
async fn parse_external_addr(&self, addr: &Option<&Value>) -> DnetViewResult