/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2026 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::time::Instant;
use darkfi::{
rpc::{
client::RpcClient,
jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult},
util::JsonValue,
},
system::{ExecutorPtr, Publisher, StoppableTask},
Error, Result,
};
use tracing::{error, info};
use url::Url;
use crate::DamCli;
impl DamCli {
/// Auxiliary function to ping configured damd daemon for liveness.
pub async fn ping(&self) -> Result<()> {
info!("Executing ping request to damd...");
let latency = Instant::now();
let rep = self.damd_daemon_request("ping", &JsonValue::Array(vec![])).await?;
let latency = latency.elapsed();
info!("Got reply: {rep:?}");
info!("Latency: {latency:?}");
Ok(())
}
/// Auxiliary function to execute a request towards the configured damd daemon JSON-RPC endpoint.
pub async fn damd_daemon_request(&self, method: &str, params: &JsonValue) -> Result {
let req = JsonRequest::new(method, params.clone());
let rep = self.rpc_client.request(req).await?;
Ok(rep)
}
/// Subscribes to damd's JSON-RPC notification endpoints.
pub async fn subscribe(&self, endpoint: &str, method: &str, ex: &ExecutorPtr) -> Result<()> {
info!("Subscribing to receive notifications for: {method}");
let endpoint = Url::parse(endpoint)?;
let _method = String::from(method);
let publisher = Publisher::new();
let subscription = publisher.clone().subscribe().await;
let _publisher = publisher.clone();
let _ex = ex.clone();
StoppableTask::new().start(
// Weird hack to prevent lifetimes hell
async move {
let rpc_client = RpcClient::new(endpoint, _ex).await?;
let req = JsonRequest::new(&_method, JsonValue::Array(vec![]));
rpc_client.subscribe(req, _publisher).await
},
|res| async move {
match res {
Ok(()) => { /* Do nothing */ }
Err(e) => {
error!("[subscribe] JSON-RPC server error: {e:?}");
publisher
.notify(JsonResult::Error(JsonError::new(
ErrorCode::InternalError,
None,
0,
)))
.await;
}
}
},
Error::RpcServerStopped,
ex.clone(),
);
info!("Detached subscription to background");
info!("All is good. Waiting for new notifications...");
let e = loop {
match subscription.receive().await {
JsonResult::Notification(n) => {
info!("Got notification from subscription");
if n.method != method {
break Error::UnexpectedJsonRpc(format!(
"Got foreign notification from damd: {}",
n.method
))
}
// Verify parameters
if !n.params.is_array() {
break Error::UnexpectedJsonRpc(
"Received notification params are not an array".to_string(),
)
}
let params = n.params.get::>().unwrap();
if params.is_empty() {
break Error::UnexpectedJsonRpc(
"Notification parameters are empty".to_string(),
)
}
for param in params {
let param = param.get::().unwrap();
println!("Notification: {param}");
}
}
JsonResult::Error(e) => {
// Some error happened in the transmission
break Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}"))
}
x => {
// And this is weird
break Error::UnexpectedJsonRpc(format!(
"Got unexpected data from JSON-RPC: {x:?}"
))
}
}
};
Err(e)
}
}