Jt/82 webrtcinrustping

ping/rust/Cargo.toml

@@ -26,11 +26,11 @@ libp2pv0461 = { package = "libp2p", version = "0.46.1", default_features = false
 libp2pv0470 = { package = "libp2p", version = "0.47.0", default_features = false, features = ["websocket", "mplex", "yamux", "tcp-async-io", "ping", "noise", "dns-async-std"] }
 libp2pv0480 = { package = "libp2p", version = "0.48.0", default_features = false, features = ["websocket", "mplex", "yamux", "tcp-async-io", "ping", "noise", "dns-async-std", "rsa"] }
 libp2pv0490 = { package = "libp2p", version = "0.49.0", features = ["websocket", "mplex", "yamux", "tcp", "async-std", "ping", "noise", "dns", "rsa"] }
-libp2pv0500 = { package = "libp2p", version = "0.50.0", features = ["websocket", "mplex", "yamux", "tcp", "async-std", "ping", "noise", "dns", "rsa", "macros"] }
+libp2pv0500 = { package = "libp2p", version = "0.50.0", features = ["websocket", "mplex", "yamux", "tcp", "tokio",     "ping", "noise", "dns", "rsa", "macros", "webrtc"] }
 
 # Next release
 [dependencies."libp2pv0510"]
 package = "libp2p"
 git = "https://github.com/libp2p/rust-libp2p"
 rev = "be0b62a78fe9d72811b9eda742137cc8ddc4da35" # This should usually point to a commit on master.
-features = ["websocket", "mplex", "yamux", "tcp", "ping", "noise", "dns", "async-std", "rsa", "macros"]
+features = ["websocket", "mplex", "yamux", "tcp", "ping", "noise", "dns", "tokio", "rsa", "macros", "webrtc"]

ping/rust/manifest.toml

@@ -16,4 +16,5 @@ instances = { min = 2, max = 10000, default = 5 }
 
   [testcases.params]
   max_latency_ms    = { type = "int", desc = "maximum value for random local link latency", unit = "ms", default = 1000 }
-  iterations        = { type = "int", desc = "number of ping iterations we'll run against each peer", unit = "count", default = 5 }
+  iterations        = { type = "int", desc = "number of ping iterations we'll run against each peer", unit = "count", default = 5 }
+  transport         = { type = "enum", desc = "transport used", values = ["tcp", "webrtc"], default = "tcp" }

ping/rust/src/bin/testplan_0440.rs

@@ -21,7 +21,6 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
     run_ping(swarm).await?;
 
     Ok(())
@@ -31,18 +30,16 @@ struct OrphanRuleWorkaround(Swarm<ping::Behaviour>);
 
 #[async_trait]
 impl PingSwarm for OrphanRuleWorkaround {
-    async fn listen_on(&mut self, address: &str) -> Result<()> {
+    async fn listen_on(&mut self, address: &str) -> Result<String> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
             if let Some(SwarmEvent::NewListenAddr { listener_id, .. }) = self.0.next().await {
                 if listener_id == id {
-                    break;
+                    return Ok(address.to_string());
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0451.rs

@@ -21,28 +21,24 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
     run_ping(swarm).await?;
-
     Ok(())
 }
 
 struct OrphanRuleWorkaround(Swarm<ping::Behaviour>);
 
 #[async_trait]
 impl PingSwarm for OrphanRuleWorkaround {
-    async fn listen_on(&mut self, address: &str) -> Result<()> {
+    async fn listen_on(&mut self, address: &str) -> Result<String> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
             if let Some(SwarmEvent::NewListenAddr { listener_id, .. }) = self.0.next().await {
                 if listener_id == id {
-                    break;
+                    return Ok(address.to_string());
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0461.rs

@@ -21,7 +21,6 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
     run_ping(swarm).await?;
 
     Ok(())
@@ -31,18 +30,16 @@ struct OrphanRuleWorkaround(Swarm<ping::Behaviour>);
 
 #[async_trait]
 impl PingSwarm for OrphanRuleWorkaround {
-    async fn listen_on(&mut self, address: &str) -> Result<()> {
+    async fn listen_on(&mut self, address: &str) -> Result<String> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
             if let Some(SwarmEvent::NewListenAddr { listener_id, .. }) = self.0.next().await {
                 if listener_id == id {
-                    break;
+                    return Ok(address.to_string());
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0470.rs

@@ -21,7 +21,6 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
     run_ping(swarm).await?;
 
     Ok(())
@@ -31,18 +30,16 @@ struct OrphanRuleWorkaround(Swarm<ping::Behaviour>);
 
 #[async_trait]
 impl PingSwarm for OrphanRuleWorkaround {
-    async fn listen_on(&mut self, address: &str) -> Result<()> {
+    async fn listen_on(&mut self, address: &str) -> Result<String> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
             if let Some(SwarmEvent::NewListenAddr { listener_id, .. }) = self.0.next().await {
                 if listener_id == id {
-                    break;
+                    return Ok(address.to_string());
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0480.rs

@@ -21,7 +21,6 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
     run_ping(swarm).await?;
 
     Ok(())
@@ -31,18 +30,16 @@ struct OrphanRuleWorkaround(Swarm<ping::Behaviour>);
 
 #[async_trait]
 impl PingSwarm for OrphanRuleWorkaround {
-    async fn listen_on(&mut self, address: &str) -> Result<()> {
+    async fn listen_on(&mut self, address: &str) -> Result<String> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
             if let Some(SwarmEvent::NewListenAddr { listener_id, .. }) = self.0.next().await {
                 if listener_id == id {
-                    break;
+                    return Ok(address.to_string());
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0490.rs

@@ -23,7 +23,6 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
     run_ping(swarm).await?;
 
     Ok(())
@@ -33,18 +32,16 @@ struct OrphanRuleWorkaround(Swarm<ping::Behaviour>);
 
 #[async_trait]
 impl PingSwarm for OrphanRuleWorkaround {
-    async fn listen_on(&mut self, address: &str) -> Result<()> {
+    async fn listen_on(&mut self, address: &str) -> Result<String> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
             if let Some(SwarmEvent::NewListenAddr { listener_id, .. }) = self.0.next().await {
                 if listener_id == id {
-                    break;
+                    return Ok(address.to_string());
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0500.rs

@@ -1,19 +1,44 @@
 use anyhow::Result;
 use async_trait::async_trait;
 use futures::StreamExt;
-use libp2pv0500::swarm::{keep_alive, NetworkBehaviour, SwarmEvent};
-use libp2pv0500::*;
-use std::collections::HashSet;
-use std::time::Duration;
-use testplan::{run_ping, PingSwarm};
-
-#[async_std::main]
+use libp2pv0500::{
+    core::{
+        either::EitherOutput,
+        muxing::StreamMuxerBox,
+        upgrade::{SelectUpgrade, Version},
+    },
+    identity, mplex, noise, ping,
+    swarm::{keep_alive, NetworkBehaviour, SwarmEvent},
+    tcp, webrtc, yamux, Multiaddr, PeerId, Swarm, Transport,
+};
+use log::{debug, info};
+use rand::thread_rng;
+use std::{collections::HashSet, time::Duration};
+use testplan::*;
+
+#[tokio::main]
 async fn main() -> Result<()> {
     let local_key = identity::Keypair::generate_ed25519();
     let local_peer_id = PeerId::from(local_key.public());
-
-    let swarm = OrphanRuleWorkaround(Swarm::with_async_std_executor(
-        development_transport(local_key).await?,
+    let transport = tcp::tokio::Transport::default()
+        .upgrade(Version::V1)
+        .authenticate(noise::NoiseAuthenticated::xx(&local_key).unwrap())
+        .multiplex(SelectUpgrade::new(
+            yamux::YamuxConfig::default(),
+            mplex::MplexConfig::default(),
+        ))
+        .timeout(Duration::from_secs(20))
+        .or_transport(webrtc::tokio::Transport::new(
+            local_key,
+            webrtc::tokio::Certificate::generate(&mut thread_rng())?,
+        ))
+        .map(|either, _| match either {
+            EitherOutput::First((p, conn)) => (p, StreamMuxerBox::new(conn)),
+            EitherOutput::Second((p, conn)) => (p, StreamMuxerBox::new(conn)),
+        })
+        .boxed();
+    let swarm = OrphanRuleWorkaround(Swarm::with_tokio_executor(
+        transport,
         Behaviour {
             keep_alive: keep_alive::Behaviour,
             ping: ping::Behaviour::new(ping::Config::new().with_interval(Duration::from_secs(1))),
@@ -37,18 +62,19 @@ struct OrphanRuleWorkaround(Swarm<Behaviour>);
 
 #[async_trait]
 impl PingSwarm for OrphanRuleWorkaround {
-    async fn listen_on(&mut self, address: &str) -> Result<()> {
+    async fn listen_on(&mut self, address: &str) -> Result<String> {
         let id = self.0.listen_on(address.parse()?)?;
-
         loop {
-            if let Some(SwarmEvent::NewListenAddr { listener_id, .. }) = self.0.next().await {
+            if let Some(SwarmEvent::NewListenAddr {
+                listener_id,
+                address,
+            }) = self.0.next().await
+            {
                 if listener_id == id {
-                    break;
+                    return Ok(address.to_string());
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {
@@ -61,8 +87,19 @@ impl PingSwarm for OrphanRuleWorkaround {
         let mut connected = HashSet::with_capacity(number);
 
         while connected.len() < number {
-            if let Some(SwarmEvent::ConnectionEstablished { peer_id, .. }) = self.0.next().await {
-                connected.insert(peer_id);
+            match self.0.next().await {
+                Some(SwarmEvent::ConnectionEstablished {
+                    peer_id, endpoint, ..
+                }) => {
+                    info!(
+                        "Connection established! {}={}",
+                        &peer_id,
+                        &endpoint.get_remote_address()
+                    );
+                    connected.insert(peer_id);
+                }
+                Some(event) => debug!("Received event {:?}", &event), //This is useful, because it sometimes logs error messages
+                None => (),
             }
         }
     }