feat: wait for the relayed interfaces before proceeding with the main flow

fbozic · fbozic · commit bae7cb5c3ab7 · 2024-05-15T16:23:43.000+02:00
diff --git a/examples/chat/README.md b/examples/chat/README.md
@@ -22,12 +22,18 @@ publish calimero-network/examples/chat/v0.0.1 ola
 ## Run locally with remote peer dial in
 This examples shows how to run two sessions locally and connect sessions manually by dialing private remote peer from each session. For the gossip message to pass from one local session to second local session it needs to go "the long way" around (local -> remote -> local).
 
+Additional info:
+- Remote instance is running in a private subnet behind NAT.
+- Remote instance PeerId: `12D3KooWP285Hw3CSTdr9oU6Ezz4hDoi6XS5vfDjjNeTJ1uFMGvp`
+- Remote instance address at the relay server: `ip4/3.71.239.80/udp/4001/quic-v1/p2p/12D3KooWAgFah4EZtWnMMGMUddGdJpb5cq2NubNCAD2jA5AZgbXF/p2p-circuit/p2p/12D3KooWP285Hw3CSTdr9oU6Ezz4hDoi6XS5vfDjjNeTJ1uFMGvp`
 Run first chat session in interactive mode with remote peer dial.
 ```
+cargo run -p chat-example -- --mode interactive --port 4002 --secret-key-seed 102 --gossip-topic-names calimero-network/examples/chat/v0.0.1 --dial-peer-addrs /ip4/3.71.239.80/udp/4001/quic-v1/p2p/12D3KooWAgFah4EZtWnMMGMUddGdJpb5cq2NubNCAD2jA5AZgbXF/p2p-circuit/p2p/12D3KooWP285Hw3CSTdr9oU6Ezz4hDoi6XS5vfDjjNeTJ1uFMGvp --relay-address /ip4/3.71.239.80/udp/4001/quic-v1/p2p/12D3KooWAgFah4EZtWnMMGMUddGdJpb5cq2NubNCAD2jA5AZgbXF
 ```
 
 Run second chat session in interactive mode with remote peer dial.
 ```
+cargo run -p chat-example -- --mode interactive --port 4003 --secret-key-seed 103 --gossip-topic-names calimero-network/examples/chat/v0.0.1 --dial-peer-addrs /ip4/3.71.239.80/udp/4001/quic-v1/p2p/12D3KooWAgFah4EZtWnMMGMUddGdJpb5cq2NubNCAD2jA5AZgbXF/p2p-circuit/p2p/12D3KooWP285Hw3CSTdr9oU6Ezz4hDoi6XS5vfDjjNeTJ1uFMGvp --relay-address /ip4/3.71.239.80/udp/4001/quic-v1/p2p/12D3KooWAgFah4EZtWnMMGMUddGdJpb5cq2NubNCAD2jA5AZgbXF
 ```
 
 In any interactive session publish new message manually:
diff --git a/examples/chat/src/main.rs b/examples/chat/src/main.rs
@@ -3,8 +3,10 @@ use std::str::FromStr;
 use clap::Parser;
 use libp2p::gossipsub;
 use libp2p::identity;
+use libp2p::PeerId;
 use multiaddr::Multiaddr;
 use tokio::io::AsyncBufReadExt;
+use tracing::debug;
 use tracing::{error, info};
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::EnvFilter;
@@ -51,7 +53,7 @@ impl FromStr for Mode {
         match mode {
             "interactive" => Ok(Mode::Interactive),
             "echo" => Ok(Mode::Echo),
-            _ => Err("Expected either 'dial' or 'listen'".to_string()),
+            _ => Err("Expected either 'interactive' or 'echo'".to_string()),
         }
     }
 }
@@ -72,21 +74,24 @@ async fn main() -> eyre::Result<()> {
     let keypair = generate_ed25519(opt.secret_key_seed);
 
     let (network_client, mut network_events) =
-        network::run(keypair, opt.port, opt.relay_address.clone()).await?;
+        network::run(keypair.clone(), opt.port, opt.relay_address.clone()).await?;
 
     if let Some(peer_addrs) = opt.dial_peer_addrs {
         for addr in peer_addrs {
+            info!("Dialing peer: {}", addr);
             network_client.dial(addr).await?;
         }
     }
 
     if let Some(topic_names) = opt.gossip_topic_names {
         for topic_name in topic_names {
+            info!("Subscribing to topic: {}", topic_name);
             let topic = gossipsub::IdentTopic::new(topic_name);
             network_client.subscribe(topic).await?;
         }
     }
 
+    let peer_id = keypair.public().to_peer_id();
     match opt.mode {
         Mode::Interactive => {
             let mut stdin = tokio::io::BufReader::new(tokio::io::stdin()).lines();
@@ -97,7 +102,7 @@ async fn main() -> eyre::Result<()> {
                         let Some(event) = event else {
                             break;
                         };
-                        handle_network_event(Mode::Interactive, network_client.clone(), event).await?;
+                        handle_network_event(network_client.clone(), event, peer_id, false).await?;
                     }
                     line = stdin.next_line() => {
                         if let Some(line) = line? {
@@ -109,7 +114,7 @@ async fn main() -> eyre::Result<()> {
         }
         Mode::Echo => {
             while let Some(event) = network_events.recv().await {
-                handle_network_event(Mode::Echo, network_client.clone(), event).await?;
+                handle_network_event(network_client.clone(), event, peer_id, true).await?;
             }
         }
     }
@@ -124,43 +129,50 @@ fn generate_ed25519(secret_key_seed: u8) -> identity::Keypair {
     identity::Keypair::ed25519_from_bytes(bytes).expect("only errors on wrong length")
 }
 
+const LINE_START: &str = ">>>>>>>>>> ";
+
 async fn handle_network_event(
-    mode: Mode,
     network_client: network::client::NetworkClient,
     event: network::types::NetworkEvent,
+    peer_id: PeerId,
+    is_echo: bool,
 ) -> eyre::Result<()> {
     match event {
         network::types::NetworkEvent::IdentifySent { peer_id } => {
-            info!("Identify sent to {:?}", peer_id);
+            debug!("Identify sent to {:?}", peer_id);
         }
         network::types::NetworkEvent::IdentifyReceived {
             peer_id,
             observed_addr,
         } => {
-            info!(
+            debug!(
                 "Identify received from {:?} at {:?}",
                 peer_id, observed_addr
             );
         }
-        network::types::NetworkEvent::Message { id, message } => {
+        network::types::NetworkEvent::Message { message, .. } => {
             let text = String::from_utf8_lossy(&message.data);
-            info!("Message from {:?}: {:?}", id, text);
-
-            match mode {
-                Mode::Echo => {
-                    let text = format!("Echo, original: '{}'", text);
-
-                    match network_client
-                        .publish(message.topic, text.into_bytes())
-                        .await
-                    {
-                        Ok(_) => info!("Echoed message back"),
-                        Err(err) => error!(%err, "Failed to echo message back"),
-                    };
+            println!("{LINE_START} Received message: {:?}", text);
+
+            if is_echo {
+                let text = format!("echo ({}): '{}'", peer_id, text);
+                if text.starts_with("echo") {
+                    debug!("Ignoring echo message");
+                    return Ok(());
                 }
-                _ => {}
+
+                match network_client
+                    .publish(message.topic, text.into_bytes())
+                    .await
+                {
+                    Ok(_) => debug!("Echoed message back"),
+                    Err(err) => error!(%err, "Failed to echo message back"),
+                };
             }
         }
+        network::types::NetworkEvent::Subscribed { topic, .. } => {
+            debug!("Subscribed to {:?}", topic);
+        }
         network::types::NetworkEvent::ListeningOn { address, .. } => {
             info!("Listening on: {}", address);
         }
@@ -185,128 +197,125 @@ async fn handle_line(
             let args = match args {
                 Some(args) => args,
                 None => {
-                    println!("Usage: dial <multiaddr>");
+                    println!("{LINE_START} Usage: dial <multiaddr>");
                     return Ok(());
                 }
             };
 
             let addr = match Multiaddr::from_str(args) {
                 Ok(addr) => addr,
                 Err(err) => {
-                    println!("Failed to parse MultiAddr: {:?}", err);
+                    println!("{LINE_START} Failed to parse MultiAddr: {:?}", err);
                     return Ok(());
                 }
             };
 
-            info!("Dialing {:?}", addr);
+            info!("{LINE_START} Dialing {:?}", addr);
 
             match network_client.dial(addr).await {
                 Ok(_) => {
-                    println!("Peer dialed");
+                    println!("{LINE_START} Peer dialed");
                 }
                 Err(err) => {
-                    println!("Failed to parse peer id: {:?}", err);
+                    println!("{LINE_START} Failed to dial peer: {:?}", err);
                 }
             };
         }
         "subscribe" => {
             let args = match args {
                 Some(args) => args,
                 None => {
-                    println!("Usage: subscribe <topic-name>");
+                    println!("{LINE_START} Usage: subscribe <topic-name>");
                     return Ok(());
                 }
             };
 
             let topic = gossipsub::IdentTopic::new(args.to_string());
             match network_client.subscribe(topic).await {
                 Ok(_) => {
-                    println!("Peer dialed");
+                    println!("{LINE_START} Peer dialed");
                 }
                 Err(err) => {
-                    println!("Failed to parse peer id: {:?}", err);
+                    println!("{LINE_START} Failed to parse peer id: {:?}", err);
                 }
             };
         }
         "unsubscribe" => {
             let args = match args {
                 Some(args) => args,
                 None => {
-                    println!("Usage: unsubscribe <topic-name>");
+                    println!("{LINE_START} Usage: unsubscribe <topic-name>");
                     return Ok(());
                 }
             };
 
             let topic = gossipsub::IdentTopic::new(args.to_string());
             match network_client.unsubscribe(topic).await {
                 Ok(_) => {
-                    println!("Peer dialed");
+                    println!("{LINE_START} Peer dialed");
                 }
                 Err(err) => {
-                    println!("Failed to parse peer id: {:?}", err);
+                    println!("{LINE_START} Failed to parse peer id: {:?}", err);
                 }
             };
         }
         "publish" => {
             let args = match args {
                 Some(args) => args,
                 None => {
-                    println!("Usage: message <topic-name> <message>");
+                    println!("{LINE_START} Usage: message <topic-name> <message>");
                     return Ok(());
                 }
             };
 
-            // Extracting topic name and message data from args
             let mut args_iter = args.split_whitespace();
             let topic_name = match args_iter.next() {
                 Some(topic) => topic,
                 None => {
-                    println!("Usage: message <topic-name> <message>");
+                    println!("{LINE_START} Usage: message <topic-name> <message>");
                     return Ok(());
                 }
             };
 
             let message_data = match args_iter.next() {
                 Some(data) => data,
                 None => {
-                    println!("Usage: message <topic-name> <message>");
+                    println!("{LINE_START} Usage: message <topic-name> <message>");
                     return Ok(());
                 }
             };
 
             let topic = gossipsub::IdentTopic::new(topic_name.to_string());
-
-            // Publishing the message
             match network_client
                 .publish(topic.hash(), message_data.as_bytes().to_vec())
                 .await
             {
                 Ok(_) => {
-                    println!("Message published successfully");
+                    println!("{LINE_START} Message published successfully");
                 }
                 Err(err) => {
-                    println!("Failed to publish message: {:?}", err);
+                    println!("{LINE_START} Failed to publish message: {:?}", err);
                 }
             };
         }
         "peers" => {
             let peer_info = network_client.peer_info().await;
-            info!("Peer info: {:?}", peer_info);
+            println!("{LINE_START} Peer info: {:?}", peer_info);
         }
         "mesh-peers" => {
             let args = match args {
                 Some(args) => args,
                 None => {
-                    println!("Usage: mesh-peers <topic-name>");
+                    println!("{LINE_START} Usage: mesh-peers <topic-name>");
                     return Ok(());
                 }
             };
 
             let topic = gossipsub::IdentTopic::new(args.to_string());
             let mesh_peer_info = network_client.mesh_peer_info(topic.hash()).await;
-            info!("Mesh peer info: {:?}", mesh_peer_info);
+            println!("{LINE_START} Mesh peer info: {:?}", mesh_peer_info);
         }
-        _ => info!("Unknown command"),
+        _ => println!("{LINE_START} Unknown command"),
     }
 
     Ok(())
diff --git a/examples/chat/src/network.rs b/examples/chat/src/network.rs
@@ -93,7 +93,7 @@ pub async fn run(
                 observed_addr,
             } => {
                 if peer_id == relay_peer_id {
-                    info!(address=%observed_addr, "Relay told us our observed address");
+                    info!("Relay told us our observed address: {}", observed_addr);
                     learned_observed_addr = true;
                 }
             }
@@ -105,21 +105,42 @@ pub async fn run(
         }
     }
 
-    // Create reservation on relay server and wait for it to be accepted.
+    // Create reservation on relay server and wait for it to be accepted ...
     client
         .listen_on(relay_address.with(multiaddr::Protocol::P2pCircuit))
         .await?;
 
     loop {
         match event_receiver.recv().await.unwrap() {
             types::NetworkEvent::RelayReservationAccepted => {
-                info!("Relay reservation accepted");
+                info!("Relay accepted our reservation");
                 break;
             }
             event => info!("unexpected: {event:?}"),
         };
     }
 
+    // ... and now wait until we are listening on the "relayed" interfaces
+    // Reference: https://github.com/libp2p/rust-libp2p/blob/60fd566a955a33c42a6ab6eefc1f0fedef9f8b83/examples/dcutr/src/main.rs#L118
+    loop {
+        tokio::select! {
+            Some(event) = event_receiver.recv() => {
+                match event {
+                    types::NetworkEvent::ListeningOn { address, .. } => {
+                        info!("Listening on: {}", address);
+                    }
+                    _ => {
+                        error!("Recieved unexpected network event: {:?}", event)
+                    }
+                }
+            }
+            _ = tokio::time::sleep(Duration::from_secs(1)) => {
+                // Likely listening on all interfaces now, thus continuing by breaking the loop.
+                break;
+            }
+        }
+    }
+
     Ok((client, event_receiver))
 }
 
diff --git a/examples/chat/src/network/events.rs b/examples/chat/src/network/events.rs
@@ -44,7 +44,7 @@ impl EventLoop {
             SwarmEvent::ConnectionEstablished {
                 peer_id, endpoint, ..
             } => {
-                info!(peer=%peer_id, ?endpoint, "Connection established");
+                debug!(peer=%peer_id, ?endpoint, "Connection established");
                 match endpoint {
                     libp2p::core::ConnectedPoint::Dialer { address, .. } => {
                         let dial_preference = match DialPreference::try_from(&address) {
@@ -116,7 +116,7 @@ impl EventLoop {
                 );
             }
             SwarmEvent::OutgoingConnectionError { peer_id, error, .. } => {
-                info!(%error, "Outgoing connection error");
+                debug!(%error, ?peer_id, "Outgoing connection error");
                 if let Some(peer_id) = peer_id {
                     if let Some(entry) = self.pending_dial.remove(&peer_id) {
                         let _ = entry.sender.send(Err(eyre::eyre!(error)));
@@ -129,7 +129,7 @@ impl EventLoop {
             SwarmEvent::Dialing {
                 peer_id: Some(peer_id),
                 ..
-            } => debug!("Dialing peer: {}", peer_id),
+            } => trace!("Dialing peer: {}", peer_id),
             SwarmEvent::ExpiredListenAddr { address, .. } => {
                 trace!("Expired listen address: {}", address)
             }
