feat(swarm): Make executor for connection tasks explicit

Cargo.toml

@@ -75,8 +75,8 @@ secp256k1 = ["libp2p-core/secp256k1"]
 rsa = ["libp2p-core/rsa"]
 ecdsa = ["libp2p-core/ecdsa"]
 serde = ["libp2p-core/serde", "libp2p-kad?/serde", "libp2p-gossipsub?/serde"]
-tokio = ["libp2p-mdns?/tokio", "libp2p-tcp?/tokio", "libp2p-dns?/tokio"]
-async-std = ["libp2p-mdns?/async-io", "libp2p-tcp?/async-io", "libp2p-dns?/async-std"]
+tokio = ["libp2p-mdns?/tokio", "libp2p-tcp?/tokio", "libp2p-dns?/tokio", "libp2p-swarm/tokio", "libp2p-core/tokio"]
+async-std = ["libp2p-mdns?/async-io", "libp2p-tcp?/async-io", "libp2p-dns?/async-std", "libp2p-swarm/async-std", "libp2p-core/async-std"]
 
 [dependencies]
 bytes = "1"

core/Cargo.toml

@@ -38,6 +38,9 @@ unsigned-varint = "0.7"
 void = "1"
 zeroize = "1"
 serde = { version = "1", optional = true, features = ["derive"] }
+# TODO make optional before commit
+tokio = { version = "1.15", features = ["rt"], optional = true }
+async-std = { version = "1.6.2", optional = true }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 ring = { version = "0.16.9", features = ["alloc", "std"], default-features = false, optional = true}
@@ -56,10 +59,12 @@ serde_json = "1.0"
 prost-build = "0.11"
 
 [features]
-secp256k1 = [ "libsecp256k1" ]
-ecdsa = [ "p256" ]
-rsa = [ "dep:ring" ]
+secp256k1 = ["libsecp256k1"]
+ecdsa = ["p256"]
+rsa = ["dep:ring"]
 serde = ["multihash/serde-codec", "dep:serde"]
+tokio = ["dep:tokio"]
+async-std = ["dep:async-std"]
 
 [[bench]]
 name = "peer_id"

core/src/lib.rs

@@ -51,6 +51,7 @@ mod peer_record_proto {
     include!(concat!(env!("OUT_DIR"), "/peer_record_proto.rs"));
 }
 
+use futures::executor::ThreadPool;
 /// Multi-address re-export.
 pub use multiaddr;
 pub type Negotiated<T> = multistream_select::Negotiated<T>;
@@ -102,3 +103,42 @@ impl<F: Fn(Pin<Box<dyn Future<Output = ()> + Send>>)> Executor for F {
         self(f)
     }
 }
+
+impl Executor for ThreadPool {
+    fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
+        self.spawn_ok(future)
+    }
+}
+
+#[cfg(feature = "tokio")]
+#[derive(Clone, Copy, Debug, Default)]
+pub enum TokioExecutor<'a> {
+    #[default]
+    Empty,
+    Given(&'a tokio::runtime::Runtime),
+}
+
+#[cfg(feature = "tokio")]
+impl<'a> Executor for TokioExecutor<'a> {
+    fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
+        match self {
+            Self::Given(runtime) => {
+                let _ = runtime.spawn(future);
+            }
+            Self::Empty => {
+                let _ = tokio::spawn(future);
+            }
+        }
+    }
+}
+
+#[cfg(feature = "async-std")]
+#[derive(Default, Debug, Clone, Copy, Hash, PartialEq, Eq, PartialOrd, Ord)]
+pub struct AsyncStdExecutor;
+
+#[cfg(feature = "async-std")]
+impl Executor for AsyncStdExecutor {
+    fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
+        let _ = async_std::task::spawn(future);
+    }
+}

examples/chat-tokio.rs

@@ -39,7 +39,7 @@ use libp2p::{
         TokioMdns,
     },
     mplex, noise,
-    swarm::{SwarmBuilder, SwarmEvent},
+    swarm::SwarmEvent,
     tcp, Multiaddr, NetworkBehaviour, PeerId, Transport,
 };
 use std::error::Error;
@@ -97,23 +97,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
     }
 
     // Create a Swarm to manage peers and events.
-    let mut swarm = {
-        let mdns = TokioMdns::new(Default::default())?;
-        let mut behaviour = MyBehaviour {
-            floodsub: Floodsub::new(peer_id),
-            mdns,
-        };
-
-        behaviour.floodsub.subscribe(floodsub_topic.clone());
-
-        SwarmBuilder::new(transport, behaviour, peer_id)
-            // We want the connection background tasks to be spawned
-            // onto the tokio runtime.
-            .executor(Box::new(|fut| {
-                tokio::spawn(fut);
-            }))
-            .build()
+    let mdns = TokioMdns::new(Default::default())?;
+    let behaviour = MyBehaviour {
+        floodsub: Floodsub::new(peer_id),
+        mdns,
     };
+    let mut swarm = libp2p_swarm::TokioSwarm::new(transport, behaviour, peer_id);
 
     // Reach out to another node if specified
     if let Some(to_dial) = std::env::args().nth(1) {

examples/chat.rs

@@ -115,7 +115,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         };
 
         behaviour.floodsub.subscribe(floodsub_topic.clone());
-        Swarm::new(transport, behaviour, local_peer_id)
+        Swarm::<_, libp2p_core::AsyncStdExecutor>::new(transport, behaviour, local_peer_id)
     };
 
     // Reach out to another node if specified

swarm/Cargo.toml

@@ -30,6 +30,10 @@ env_logger = "0.9"
 libp2p = { path = "..", features = ["full"] }
 quickcheck = { package = "quickcheck-ext", path = "../misc/quickcheck-ext" }
 
+[features]
+tokio = ["libp2p-core/tokio"]
+async-std = ["libp2p-core/async-std"]
+
 # Passing arguments to the docsrs builder in order to properly document cfg's. 
 # More information: https://docs.rs/about/builds#cross-compiling
 [package.metadata.docs.rs]