nodes refactor, some log changes again

Author: erhant

compute/src/handlers/workflow.rs

@@ -129,11 +129,7 @@ impl WorkflowHandler {
 
                 // convert payload to message
                 let payload_str = serde_json::json!(payload).to_string();
-                log::info!(
-                    "Publishing result for task {}\n{}",
-                    task.task_id,
-                    payload_str
-                );
+                log::info!("Publishing result for task {}", task.task_id);
                 DriaMessage::new(payload_str, Self::RESPONSE_TOPIC)
             }
             Err(err) => {

compute/src/node.rs

@@ -81,9 +81,7 @@ impl DriaComputeNode {
         let (p2p_client, p2p_commander, message_rx) = DriaP2PClient::new(
             keypair,
             config.p2p_listen_addr.clone(),
-            available_nodes.bootstrap_nodes.clone().into_iter(),
-            available_nodes.relay_nodes.clone().into_iter(),
-            available_nodes.rpc_nodes.clone().into_iter(),
+            &available_nodes,
             protocol,
         )?;
 

monitor/src/main.rs

@@ -33,9 +33,7 @@ async fn main() -> eyre::Result<()> {
     let (client, commander, msg_rx) = DriaP2PClient::new(
         keypair,
         listen_addr,
-        nodes.bootstrap_nodes.into_iter(),
-        nodes.relay_nodes.into_iter(),
-        nodes.rpc_nodes.into_iter(),
+        &nodes,
         DriaP2PProtocol::new_major_minor(network.protocol_name()),
     )?;
 

p2p/src/client.rs

@@ -10,7 +10,7 @@ use std::time::Duration;
 use tokio::sync::mpsc;
 
 use crate::behaviour::{DriaBehaviour, DriaBehaviourEvent};
-use crate::DriaP2PProtocol;
+use crate::{DriaNodes, DriaP2PProtocol};
 
 use super::commands::DriaP2PCommand;
 use super::DriaP2PCommander;
@@ -46,9 +46,7 @@ impl DriaP2PClient {
     pub fn new(
         keypair: Keypair,
         listen_addr: Multiaddr,
-        bootstraps: impl Iterator<Item = Multiaddr>,
-        relays: impl Iterator<Item = Multiaddr>,
-        rpcs: impl Iterator<Item = Multiaddr>,
+        nodes: &DriaNodes,
         protocol: DriaP2PProtocol,
     ) -> Result<(
         DriaP2PClient,
@@ -89,15 +87,18 @@ impl DriaP2PClient {
             .set_mode(Some(libp2p::kad::Mode::Server));
 
         // initiate bootstrap
-        for addr in bootstraps {
+        for addr in &nodes.bootstrap_nodes {
             log::info!("Dialling bootstrap: {:#?}", addr);
             if let Some(peer_id) = addr.iter().find_map(|p| match p {
                 Protocol::P2p(peer_id) => Some(peer_id),
                 _ => None,
             }) {
                 swarm.dial(addr.clone())?;
                 log::info!("Adding {} to Kademlia routing table", addr);
-                swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
+                swarm
+                    .behaviour_mut()
+                    .kademlia
+                    .add_address(&peer_id, addr.clone());
             } else {
                 log::warn!("Missing peerID in address: {}", addr);
             }
@@ -115,17 +116,29 @@ impl DriaP2PClient {
         // listen on all interfaces for incoming connections
         log::info!("Listening p2p network on: {}", listen_addr);
         swarm.listen_on(listen_addr)?;
-        for addr in relays {
+
+        // listen on relay addresses with p2p circuit
+        for addr in &nodes.relay_nodes {
             log::info!("Listening to relay: {}", addr);
             swarm.listen_on(addr.clone().with(Protocol::P2pCircuit))?;
         }
 
         // dial rpc nodes
-        for rpc_addr in rpcs {
+        for rpc_addr in &nodes.rpc_nodes {
             log::info!("Dialing RPC node: {}", rpc_addr);
-            swarm.dial(rpc_addr)?;
+            swarm.dial(rpc_addr.clone())?;
         }
 
+        // add rpcs as explicit peers
+        // TODO: may not be necessary
+        // for rpc_peer_id in &nodes.rpc_peerids {
+        //     log::info!("Adding {} as explicit peer.", rpc_peer_id);
+        //     swarm
+        //         .behaviour_mut()
+        //         .gossipsub
+        //         .add_explicit_peer(rpc_peer_id);
+        // }
+
         // create commander
         let (cmd_tx, cmd_rx) = mpsc::channel(COMMAND_CHANNEL_BUFSIZE);
         let commander = DriaP2PCommander::new(cmd_tx, protocol.clone());

p2p/src/nodes.rs

@@ -29,6 +29,26 @@ impl DriaNodes {
         }
     }
 
+    pub fn with_relay_nodes(mut self, addresses: impl IntoIterator<Item = Multiaddr>) -> Self {
+        self.relay_nodes.extend(addresses);
+        self
+    }
+
+    pub fn with_bootstrap_nodes(mut self, addresses: impl IntoIterator<Item = Multiaddr>) -> Self {
+        self.bootstrap_nodes.extend(addresses);
+        self
+    }
+
+    pub fn with_rpc_nodes(mut self, addresses: impl IntoIterator<Item = Multiaddr>) -> Self {
+        self.rpc_nodes.extend(addresses);
+        self
+    }
+
+    pub fn with_rpc_peer_ids(mut self, addresses: impl IntoIterator<Item = PeerId>) -> Self {
+        self.rpc_peerids.extend(addresses);
+        self
+    }
+
     /// Parses static bootstrap & relay nodes from environment variables.
     ///
     /// The environment variables are:

p2p/tests/listen_test.rs

@@ -1,4 +1,4 @@
-use dkn_p2p::{DriaP2PClient, DriaP2PProtocol};
+use dkn_p2p::{DriaNodes, DriaP2PClient, DriaP2PProtocol};
 use eyre::Result;
 use libp2p_identity::Keypair;
 
@@ -12,13 +12,18 @@ async fn test_listen_topic_once() -> Result<()> {
         .is_test(true)
         .try_init();
 
+    let listen_addr = "/ip4/0.0.0.0/tcp/4001".parse()?;
+
+    // prepare nodes
+    let nodes = DriaNodes::new(dkn_p2p::DriaNetworkType::Community)
+    .with_bootstrap_nodes(["/ip4/44.206.245.139/tcp/4001/p2p/16Uiu2HAm4q3LZU2T9kgjKK4ysy6KZYKLq8KiXQyae4RHdF7uqSt4".parse()?])
+    .with_relay_nodes(["/ip4/34.201.33.141/tcp/4001/p2p/16Uiu2HAkuXiV2CQkC9eJgU6cMnJ9SMARa85FZ6miTkvn5fuHNufa".parse()?]);
+
     // spawn P2P client in another task
     let (client, mut commander, mut msg_rx) = DriaP2PClient::new(
         Keypair::generate_secp256k1(),
-        "/ip4/0.0.0.0/tcp/4001".parse()?,
-        vec!["/ip4/44.206.245.139/tcp/4001/p2p/16Uiu2HAm4q3LZU2T9kgjKK4ysy6KZYKLq8KiXQyae4RHdF7uqSt4".parse()?].into_iter(),
-        vec!["/ip4/34.201.33.141/tcp/4001/p2p/16Uiu2HAkuXiV2CQkC9eJgU6cMnJ9SMARa85FZ6miTkvn5fuHNufa".parse()?].into_iter(),
-        vec![].into_iter(),
+        listen_addr,
+        &nodes,
         DriaP2PProtocol::default(),
     )
     .expect("could not create p2p client");