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,8 +21,8 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
-    run_ping(swarm).await?;
+    let client = testground::client::Client::new_and_init().await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -31,7 +31,7 @@ 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<Option<String>> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
@@ -42,7 +42,7 @@ impl PingSwarm for OrphanRuleWorkaround {
             }
         }
 
-        Ok(())
+        Ok(None)
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0451.rs

@@ -21,8 +21,8 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
-    run_ping(swarm).await?;
+    let client = testground::client::Client::new_and_init().await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -31,7 +31,7 @@ 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<Option<String>> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
@@ -42,7 +42,7 @@ impl PingSwarm for OrphanRuleWorkaround {
             }
         }
 
-        Ok(())
+        Ok(None)
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0461.rs

@@ -21,8 +21,8 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
-    run_ping(swarm).await?;
+    let client = testground::client::Client::new_and_init().await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -31,7 +31,7 @@ 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<Option<String>> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
@@ -42,7 +42,7 @@ impl PingSwarm for OrphanRuleWorkaround {
             }
         }
 
-        Ok(())
+        Ok(None)
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0470.rs

@@ -21,8 +21,8 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
-    run_ping(swarm).await?;
+    let client = testground::client::Client::new_and_init().await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -31,7 +31,7 @@ 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<Option<String>> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
@@ -42,7 +42,7 @@ impl PingSwarm for OrphanRuleWorkaround {
             }
         }
 
-        Ok(())
+        Ok(None)
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0480.rs

@@ -21,8 +21,8 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
-    run_ping(swarm).await?;
+    let client = testground::client::Client::new_and_init().await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -31,7 +31,7 @@ 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<Option<String>> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
@@ -42,7 +42,7 @@ impl PingSwarm for OrphanRuleWorkaround {
             }
         }
 
-        Ok(())
+        Ok(None)
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0490.rs

@@ -23,8 +23,8 @@ async fn main() -> Result<()> {
         ),
         local_peer_id,
     ));
-
-    run_ping(swarm).await?;
+    let client = testground::client::Client::new_and_init().await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -33,7 +33,7 @@ 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<Option<String>> {
         let id = self.0.listen_on(address.parse()?)?;
 
         loop {
@@ -44,7 +44,7 @@ impl PingSwarm for OrphanRuleWorkaround {
             }
         }
 
-        Ok(())
+        Ok(None)
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {

ping/rust/src/bin/testplan_0500.rs

@@ -1,27 +1,41 @@
 use anyhow::Result;
 use async_trait::async_trait;
 use futures::StreamExt;
+use libp2pv0500::core::multiaddr::*;
+use libp2pv0500::core::muxing::*;
 use libp2pv0500::swarm::{keep_alive, NetworkBehaviour, SwarmEvent};
 use libp2pv0500::*;
-use std::collections::HashSet;
-use std::time::Duration;
-use testplan::{run_ping, PingSwarm};
+use libp2pv0500::{tokio_development_transport, webrtc};
+use log::info;
+use rand::thread_rng;
+use std::{collections::HashSet, time::Duration};
+use testplan::*;
 
 #[async_std::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 client = testground::client::Client::new_and_init().await?;
+    let transport = match transport_param(&client).as_str() {
+        "tcp" => tokio_development_transport(local_key)?,
+        "webrtc" => webrtc::tokio::Transport::new(
+            local_key,
+            webrtc::tokio::Certificate::generate(&mut thread_rng())?,
+        )
+        .map(|(peer_id, conn), _| (peer_id, StreamMuxerBox::new(conn)))
+        .boxed(),
+        unhandled => unimplemented!("Transport unhandled in test: '{}'", unhandled),
+    };
+    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))),
         },
         local_peer_id,
     ));
 
-    run_ping(swarm).await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -37,18 +51,23 @@ 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<Option<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;
+                    info!(
+                        "NewListenAddr event: listener_id={:?}, address={:?}",
+                        &listener_id, &address
+                    );
+                    return Ok(Some(address.to_string()));
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {
@@ -61,8 +80,15 @@ 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);
+                    connected.insert(peer_id);
+                }
+                Some(event) => info!("Received event {:?}", &event),
+                None => (),
             }
         }
     }

ping/rust/src/bin/testplan_0510.rs

@@ -1,27 +1,43 @@
 use anyhow::Result;
 use async_trait::async_trait;
 use futures::StreamExt;
-use libp2pv0510::swarm::{keep_alive, NetworkBehaviour, SwarmEvent};
-use libp2pv0510::*;
+use libp2pv0510::{
+    core::muxing::StreamMuxerBox,
+    identity, ping,
+    swarm::{keep_alive, NetworkBehaviour, SwarmEvent},
+    tokio_development_transport, webrtc, Multiaddr, PeerId, Swarm, Transport,
+};
+use log::info;
+use rand::thread_rng;
 use std::collections::HashSet;
 use std::time::Duration;
-use testplan::{run_ping, PingSwarm};
+use testplan::{run_ping, transport_param, PingSwarm};
 
 #[async_std::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 client = testground::client::Client::new_and_init().await?;
+    let transport = match transport_param(&client).as_str() {
+        "tcp" => tokio_development_transport(local_key)?,
+        "webrtc" => webrtc::tokio::Transport::new(
+            local_key,
+            webrtc::tokio::Certificate::generate(&mut thread_rng())?,
+        )
+        .map(|(peer_id, conn), _| (peer_id, StreamMuxerBox::new(conn)))
+        .boxed(),
+        unhandled => unimplemented!("Transport unhandled in test: '{}'", unhandled),
+    };
+    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))),
         },
         local_peer_id,
     ));
 
-    run_ping(swarm).await?;
+    run_ping(swarm, client).await?;
 
     Ok(())
 }
@@ -37,18 +53,20 @@ 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<Option<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(Some(address.to_string()));
                 }
             }
         }
-
-        Ok(())
     }
 
     fn dial(&mut self, address: &str) -> Result<()> {
@@ -61,8 +79,15 @@ 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);
+                    connected.insert(peer_id);
+                }
+                Some(event) => info!("Received event {:?}", &event), //This is useful, because it sometimes logs error messages
+                None => (),
             }
         }
     }