GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

query nym-api from the nym-api

Browse Source
This commit is contained in:
Simon Wicky
2023-08-03 14:50:35 +02:00
parent 0d81c7d765
commit 7ca4c3b7b1
3 changed files with 45 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-api/src/main.rs
+10 -1
View File
@@ -22,6 +22,7 @@ use nym_bin_common::logging::setup_logging;
use nym_contract_cache::cache::NymContractCache;
use nym_sphinx::receiver::SphinxMessageReceiver;
use nym_task::TaskManager;
use nym_validator_client::NymApiClient;
use rand::rngs::OsRng;
use std::error::Error;
use support::{http, nyxd};
@@ -124,12 +125,20 @@ async fn start_nym_api_tasks(
if config.network_monitor.enabled {
// if network monitor is enabled, the storage MUST BE available
let storage = maybe_storage.unwrap();
let url = format!(
"http://{}:{}",
rocket.config().address,
rocket.config().port
)
.parse()
.unwrap();
let nym_api_client = NymApiClient::new(url);
network_monitor::start::<SphinxMessageReceiver>(
&config.network_monitor,
nym_contract_cache_state,
storage,
nyxd_client.clone(),
nym_api_client,
&shutdown,
)
.await;
nym-api/src/network_monitor/mod.rs
+17 -1
View File
@@ -22,6 +22,7 @@ use nym_sphinx::acknowledgements::AckKey;
use nym_sphinx::params::PacketType;
use nym_sphinx::receiver::MessageReceiver;
use nym_task::TaskManager;
use nym_validator_client::NymApiClient;
use std::sync::Arc;
pub(crate) mod gateways_reader;
@@ -36,10 +37,12 @@ pub(crate) fn setup<'a>(
nym_contract_cache_state: &NymContractCache,
storage: &NymApiStorage,
nyxd_client: nyxd::Client,
nym_api_client: NymApiClient,
) -> NetworkMonitorBuilder<'a> {
NetworkMonitorBuilder::new(
config,
nyxd_client,
nym_api_client,
storage.to_owned(),
nym_contract_cache_state.to_owned(),
)
@@ -48,6 +51,7 @@ pub(crate) fn setup<'a>(
pub(crate) struct NetworkMonitorBuilder<'a> {
config: &'a config::NetworkMonitor,
nyxd_client: nyxd::Client,
nym_api_client: NymApiClient,
node_status_storage: NymApiStorage,
validator_cache: NymContractCache,
}
@@ -56,12 +60,14 @@ impl<'a> NetworkMonitorBuilder<'a> {
pub(crate) fn new(
config: &'a config::NetworkMonitor,
nyxd_client: nyxd::Client,
nym_api_client: NymApiClient,
node_status_storage: NymApiStorage,
validator_cache: NymContractCache,
) -> Self {
NetworkMonitorBuilder {
config,
nyxd_client,
nym_api_client,
node_status_storage,
validator_cache,
}
@@ -108,6 +114,7 @@ impl<'a> NetworkMonitorBuilder<'a> {
Arc::clone(&encryption_keypair),
self.config.debug.gateway_sending_rate,
bandwidth_controller,
self.nym_api_client,
self.config.debug.disabled_credentials_mode,
);
@@ -181,6 +188,7 @@ fn new_packet_sender(
local_sphinx: Arc<encryption::KeyPair>,
max_sending_rate: usize,
bandwidth_controller: BandwidthController<nyxd::Client, PersistentStorage>,
nym_api_client: nym_validator_client::NymApiClient,
disabled_credentials_mode: bool,
) -> PacketSender {
PacketSender::new(
@@ -192,6 +200,7 @@ fn new_packet_sender(
config.debug.max_concurrent_gateway_clients,
max_sending_rate,
bandwidth_controller,
nym_api_client,
disabled_credentials_mode,
)
}
@@ -224,9 +233,16 @@ pub(crate) async fn start<R: MessageReceiver + Send + 'static>(
nym_contract_cache_state: &NymContractCache,
storage: &NymApiStorage,
nyxd_client: nyxd::Client,
nym_api_client: NymApiClient,
shutdown: &TaskManager,
) {
let monitor_builder = setup(config, nym_contract_cache_state, storage, nyxd_client);
let monitor_builder = setup(
config,
nym_contract_cache_state,
storage,
nyxd_client,
nym_api_client,
);
info!("Starting network monitor...");
let runnables: NetworkMonitorRunnables<R> = monitor_builder.build().await;
runnables.spawn_tasks(shutdown);
nym-api/src/network_monitor/monitor/sender.rs
+18 -2
View File
@@ -21,6 +21,7 @@ use nym_gateway_client::error::GatewayClientError;
use nym_gateway_client::{AcknowledgementReceiver, GatewayClient, MixnetMessageReceiver};
use nym_sphinx::forwarding::packet::MixPacket;
use nym_task::TaskClient;
use nym_validator_client::NymApiClient;
use pin_project::pin_project;
use std::mem;
use std::num::NonZeroUsize;
@@ -133,6 +134,7 @@ pub(crate) struct PacketSender {
active_gateway_clients: ActiveGatewayClients,
fresh_gateway_client_data: Arc<FreshGatewayClientData>,
nym_api_client: NymApiClient,
gateway_connection_timeout: Duration,
max_concurrent_clients: usize,
max_sending_rate: usize,
@@ -151,6 +153,7 @@ impl PacketSender {
max_concurrent_clients: usize,
max_sending_rate: usize,
bandwidth_controller: BandwidthController<nyxd::Client, PersistentStorage>,
nym_api_client: NymApiClient,
disabled_credentials_mode: bool,
) -> Self {
PacketSender {
@@ -163,6 +166,7 @@ impl PacketSender {
bandwidth_controller,
disabled_credentials_mode,
}),
nym_api_client,
gateway_connection_timeout,
max_concurrent_clients,
max_sending_rate,
@@ -186,6 +190,7 @@ impl PacketSender {
identity: identity::PublicKey,
sphinx: encryption::PublicKey,
fresh_gateway_client_data: &FreshGatewayClientData,
nym_api_client: NymApiClient,
) -> (
GatewayClientHandle,
(MixnetMessageReceiver, AcknowledgementReceiver),
@@ -209,6 +214,7 @@ impl PacketSender {
ack_sender,
fresh_gateway_client_data.gateway_response_timeout,
Some(fresh_gateway_client_data.bandwidth_controller.clone()),
nym_api_client,
nym_task::TaskClient::dummy(),
);
@@ -285,13 +291,19 @@ impl PacketSender {
identity: identity::PublicKey,
sphinx: encryption::PublicKey,
fresh_gateway_client_data: &FreshGatewayClientData,
nym_api_client: NymApiClient,
gateway_connection_timeout: Duration,
) -> Option<(
GatewayClientHandle,
(MixnetMessageReceiver, AcknowledgementReceiver),
)> {
let (new_client, (message_receiver, ack_receiver)) =
Self::new_gateway_client_handle(address, identity, sphinx, fresh_gateway_client_data);
let (new_client, (message_receiver, ack_receiver)) = Self::new_gateway_client_handle(
address,
identity,
sphinx,
fresh_gateway_client_data,
nym_api_client,
);
// Put this in timeout in case the gateway has incorrectly set their ulimit and our connection
// gets stuck in their TCP queue and just hangs on our end but does not terminate
@@ -349,6 +361,7 @@ impl PacketSender {
gateway_connection_timeout: Duration,
packets: GatewayPackets,
fresh_gateway_client_data: Arc<FreshGatewayClientData>,
nym_api_client: NymApiClient,
client: Option<GatewayClientHandle>,
max_sending_rate: usize,
) -> Option<GatewayClientHandle> {
@@ -370,6 +383,7 @@ impl PacketSender {
packets.pub_key,
packets.encryption_key,
&fresh_gateway_client_data,
nym_api_client,
gateway_connection_timeout,
)
.await?;
@@ -497,6 +511,7 @@ impl PacketSender {
// we're not interacting with right now)
drop(guard);
let nym_api_client = &self.nym_api_client;
// can't chain it all nicely together as there's no adapter method defined on Stream directly
// for ForEachConcurrentClientUse
let used_clients = ForEachConcurrentClientUse::new(
@@ -507,6 +522,7 @@ impl PacketSender {
gateway_connection_timeout,
packets,
fresh_data,
nym_api_client.clone(),
client,
max_sending_rate,
)