deps: update pea2pea to 0.40

Signed-off-by: ljedrz <[email protected]>

.crawler/Cargo.toml

@@ -49,7 +49,7 @@ optional = true
 version = "0.12"
 
 [dependencies.pea2pea]
-version = "0.39"
+version = "0.40"
 
 [dependencies.postgres-native-tls]
 version = "0.5"

.crawler/src/crawler.rs

@@ -92,7 +92,7 @@ impl Crawler {
             ..Default::default()
         };
 
-        let pea2pea_node = Pea2PeaNode::new(Some(config)).await.unwrap();
+        let pea2pea_node = Pea2PeaNode::new(config).await.unwrap();
         let client_state = Default::default();
         let node = Self {
             synth_node: SynthNode::new(pea2pea_node, client_state),
@@ -132,7 +132,7 @@ impl Crawler {
         task::spawn(async move {
             loop {
                 info!(parent: node.node().span(), "crawling the network for more peers; asking peers for their peers");
-                node.send_broadcast(ClientMessage::PeerRequest).unwrap();
+                node.broadcast(ClientMessage::PeerRequest).unwrap();
 
                 // Disconnect from peers that we've collected sufficient information on or that have become stale.
                 let addrs_to_disconnect = node.known_network.addrs_to_disconnect();
@@ -159,7 +159,7 @@ impl Crawler {
                             let connection_init_timestamp = OffsetDateTime::now_utc();
                             if node_clone.node().connect(addr).await.is_ok() {
                                 // Immediately ask for the new peer's peers.
-                                let _ = node_clone.send_direct_message(addr, ClientMessage::PeerRequest);
+                                let _ = node_clone.unicast(addr, ClientMessage::PeerRequest);
                                 node_clone.known_network.connected_to_node(addr, connection_init_timestamp, true);
                             } else {
                                 node_clone.known_network.connected_to_node(addr, connection_init_timestamp, false);
@@ -343,7 +343,7 @@ impl Crawler {
             .choose_multiple(&mut self.rng(), SHARED_PEER_COUNT);
 
         debug!(parent: self.node().span(), "sending a PeerResponse to {}", source);
-        self.send_direct_message(source, ClientMessage::PeerResponse(peers, None))?;
+        self.unicast(source, ClientMessage::PeerResponse(peers, None))?;
 
         Ok(())
     }
@@ -371,7 +371,7 @@ impl Crawler {
                                 node_clone.known_network.connected_to_node(addr, connection_init_timestamp, true);
 
                                 // Immediately ask for the new peer's peers.
-                                let _ = node_clone.send_direct_message(addr, ClientMessage::PeerRequest);
+                                let _ = node_clone.unicast(addr, ClientMessage::PeerRequest);
                             } else {
                                 node_clone.known_network.connected_to_node(addr, connection_init_timestamp, false);
                             }
@@ -406,7 +406,7 @@ impl Crawler {
         );
 
         debug!(parent: self.node().span(), "sending a Pong to {}", source);
-        self.send_direct_message(source, msg)?;
+        self.unicast(source, msg)?;
 
         Ok(())
     }

.integration/Cargo.toml

@@ -31,7 +31,7 @@ optional = true
 version = "0.1"
 
 [dependencies.pea2pea]
-version = "0.39"
+version = "0.40"
 
 [dependencies.peak_alloc]
 version = "0.1"

.integration/src/test_node.rs

@@ -74,7 +74,7 @@ impl TestNode {
             ..Default::default()
         };
 
-        let pea2pea_node = Pea2PeaNode::new(Some(config)).await.unwrap();
+        let pea2pea_node = Pea2PeaNode::new(config).await.unwrap();
         let client_state = Default::default();
         let node = TestNode::new(pea2pea_node, client_state);
         node.enable_disconnect().await;
@@ -106,7 +106,7 @@ impl TestNode {
             loop {
                 if node.node().num_connected() != 0 {
                     info!(parent: node.node().span(), "sending out Pings");
-                    node.send_broadcast(ping_msg.clone()).unwrap();
+                    node.broadcast(ping_msg.clone()).unwrap();
                 }
                 tokio::time::sleep(Duration::from_secs(PING_INTERVAL_SECS)).await;
             }
@@ -121,7 +121,7 @@ impl TestNode {
                 let num_connections = node.node().num_connected() + node.node().num_connecting();
                 if num_connections < DESIRED_CONNECTIONS && node.node().num_connected() != 0 {
                     info!(parent: node.node().span(), "I'd like to have {} more peers; asking peers for their peers", DESIRED_CONNECTIONS - num_connections);
-                    node.send_broadcast(ClientMessage::PeerRequest).unwrap();
+                    node.broadcast(ClientMessage::PeerRequest).unwrap();
                 }
                 tokio::time::sleep(Duration::from_secs(PEER_INTERVAL_SECS)).await;
             }
@@ -179,7 +179,7 @@ impl TestNode {
         let msg = ClientMessage::PeerResponse(peers, None);
         info!(parent: self.node().span(), "sending a PeerResponse to {}", source);
 
-        self.send_direct_message(source, msg)?;
+        self.unicast(source, msg)?;
 
         Ok(())
     }
@@ -222,7 +222,7 @@ impl TestNode {
 
         info!(parent: self.node().span(), "sending a Pong to {}", source);
 
-        self.send_direct_message(source, msg)?;
+        self.unicast(source, msg)?;
 
         Ok(())
     }

.integration/tests/common/mod.rs

@@ -51,7 +51,7 @@ pub async fn spawn_test_node_with_nonce(local_nonce: ClientNonce) -> TestNode {
         ..Default::default()
     };
 
-    let pea2pea_node = pea2pea::Node::new(Some(config)).await.unwrap();
+    let pea2pea_node = pea2pea::Node::new(config).await.unwrap();
     let client_state = ClientState {
         local_nonce,
         ..Default::default()

.integration/tests/network/manual_testing.rs

@@ -39,7 +39,7 @@ async fn spawn_inert_node_at_port() {
         ..Default::default()
     };
 
-    let test_node = TestNode::new(Pea2PeaNode::new(Some(config)).await.unwrap(), Default::default());
+    let test_node = TestNode::new(Pea2PeaNode::new(config).await.unwrap(), Default::default());
     test_node.enable_handshake().await;
     test_node.enable_reading().await;
     test_node.enable_writing().await;

.integration/tests/network/perf.rs

@@ -91,7 +91,7 @@ async fn measure_peer_request_time() {
     for i in 0..NUM_ITERATIONS {
         let start = Instant::now();
         test_node
-            .send_direct_message(client_addr, Message::PeerRequest)
+            .unicast(client_addr, Message::PeerRequest)
             .unwrap()
             .await
             .unwrap()
@@ -130,12 +130,7 @@ async fn measure_ping_time() {
     let init_recv_count = test_node.node().stats().received().0 as usize;
     for i in 0..NUM_ITERATIONS {
         let start = Instant::now();
-        test_node
-            .send_direct_message(client_addr, ping.clone())
-            .unwrap()
-            .await
-            .unwrap()
-            .unwrap();
+        test_node.unicast(client_addr, ping.clone()).unwrap().await.unwrap().unwrap();
         wait_until!(1, test_node.node().stats().received().0 as usize == init_recv_count + i + 1);
         avg_request_time += start.elapsed();
     }

.synthetic_node/Cargo.toml

@@ -28,7 +28,7 @@ version = "0.3"
 version = "0.12"
 
 [dependencies.pea2pea]
-version = "0.39"
+version = "0.40"
 
 [dependencies.rand]
 version = "0.8"

.synthetic_node/src/lib.rs

@@ -247,7 +247,7 @@ impl Disconnect for SynthNode {
     async fn handle_disconnect(&self, disconnecting_addr: SocketAddr) {
         // Send a disconnect message before breaking the connection.
         let disconnect_msg = ClientMessage::Disconnect(DisconnectReason::NoReasonGiven);
-        if let Ok(rx) = self.send_direct_message(disconnecting_addr, disconnect_msg) {
+        if let Ok(rx) = self.unicast(disconnecting_addr, disconnect_msg) {
             let _ = rx.await;
         }