Merge #277

277: Add the means to connect by PeerId r=ljedrz a=ljedrz

I've encountered this while working on improved bitswap <> kademlia interop - we still can't easily connect using `PeerId`. Improve this by introducing an `enum ConnectionTarget` that is either a `Multiaddr` or a `PeerId` and adjusting the `SwarmApi.connect_registry` accordingly. There's also a new related test.

As a drive-by, adjust some `tracing` features that are among the defaults but we'd still enjoy having 😃.

Co-authored-by: ljedrz <ljedrz@gmail.com>

Author: bors[bot]

Cargo.lock

@@ -42,7 +42,7 @@ prost-build = { default-features = false, version = "0.6" }
 [dev-dependencies]
 hex-literal = { default-features = false, version = "0.3" }
 sha2 = { default-features = false, version = "0.9" }
-tracing-subscriber = { default-features = false, features = ["fmt", "tracing-log"], version = "0.2" }
+tracing-subscriber = { default-features = false, features = ["fmt", "tracing-log", "ansi", "env-filter"], version = "0.2" }
 
 [workspace]
 members = [ "bitswap", "http", "unixfs" ]

Cargo.toml

@@ -47,9 +47,8 @@ use self::dag::IpldDag;
 pub use self::error::Error;
 use self::ipns::Ipns;
 pub use self::p2p::pubsub::{PubsubMessage, SubscriptionStream};
-pub use self::p2p::Connection;
-pub use self::p2p::SwarmTypes;
 use self::p2p::{create_swarm, SwarmOptions, TSwarm};
+pub use self::p2p::{Connection, ConnectionTarget, SwarmTypes};
 pub use self::path::IpfsPath;
 pub use self::repo::RepoTypes;
 use self::repo::{create_repo, Repo, RepoEvent, RepoOptions};
@@ -244,7 +243,10 @@ type FutureSubscription<T, E> = SubscriptionFuture<Result<T, E>>;
 #[derive(Debug)]
 enum IpfsEvent {
     /// Connect
-    Connect(Multiaddr, OneshotSender<FutureSubscription<(), String>>),
+    Connect(
+        ConnectionTarget,
+        OneshotSender<FutureSubscription<(), String>>,
+    ),
     /// Addresses
     Addresses(Channel<Vec<(PeerId, Vec<Multiaddr>)>>),
     /// Local addresses
@@ -467,13 +469,13 @@ impl<Types: IpfsTypes> Ipfs<Types> {
         self.ipns().cancel(key).instrument(self.span.clone()).await
     }
 
-    pub async fn connect(&self, addr: Multiaddr) -> Result<(), Error> {
+    pub async fn connect<T: Into<ConnectionTarget>>(&self, target: T) -> Result<(), Error> {
         self.span
             .in_scope(|| async {
                 let (tx, rx) = oneshot_channel();
                 self.to_task
                     .clone()
-                    .send(IpfsEvent::Connect(addr, tx))
+                    .send(IpfsEvent::Connect(target.into(), tx))
                     .await?;
                 let subscription = rx.await?;
                 subscription.await?.map_err(|e| format_err!("{}", e))
@@ -892,8 +894,8 @@ impl<TRepoTypes: RepoTypes> Future for IpfsFuture<TRepoTypes> {
                 };
 
                 match inner {
-                    IpfsEvent::Connect(addr, ret) => {
-                        ret.send(self.swarm.connect(addr)).ok();
+                    IpfsEvent::Connect(target, ret) => {
+                        ret.send(self.swarm.connect(target)).ok();
                     }
                     IpfsEvent::Addresses(ret) => {
                         let addrs = self.swarm.addrs();

src/lib.rs

@@ -1,5 +1,5 @@
 use super::pubsub::Pubsub;
-use super::swarm::{Connection, Disconnector, SwarmApi};
+use super::swarm::{Connection, ConnectionTarget, Disconnector, SwarmApi};
 use crate::p2p::{SwarmOptions, SwarmTypes};
 use crate::repo::BlockPut;
 use crate::subscription::{SubscriptionFuture, SubscriptionRegistry};
@@ -411,8 +411,8 @@ impl<Types: IpfsTypes> Behaviour<Types> {
         self.swarm.connections()
     }
 
-    pub fn connect(&mut self, addr: Multiaddr) -> SubscriptionFuture<Result<(), String>> {
-        self.swarm.connect(addr)
+    pub fn connect(&mut self, target: ConnectionTarget) -> SubscriptionFuture<Result<(), String>> {
+        self.swarm.connect(target)
     }
 
     pub fn disconnect(&mut self, addr: Multiaddr) -> Option<Disconnector> {

src/p2p/behaviour.rs

@@ -13,7 +13,7 @@ pub(crate) mod pubsub;
 mod swarm;
 mod transport;
 
-pub use swarm::Connection;
+pub use swarm::{Connection, ConnectionTarget};
 
 pub type TSwarm<T> = Swarm<behaviour::Behaviour<T>>;
 

src/p2p/mod.rs

@@ -4,7 +4,7 @@ use libp2p::core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
 use libp2p::swarm::protocols_handler::{
     DummyProtocolsHandler, IntoProtocolsHandler, ProtocolsHandler,
 };
-use libp2p::swarm::{self, NetworkBehaviour, PollParameters, Swarm};
+use libp2p::swarm::{self, DialPeerCondition, NetworkBehaviour, PollParameters, Swarm};
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::time::Duration;
 
@@ -19,6 +19,24 @@ pub struct Connection {
     pub rtt: Option<Duration>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum ConnectionTarget {
+    Addr(Multiaddr),
+    PeerId(PeerId),
+}
+
+impl From<Multiaddr> for ConnectionTarget {
+    fn from(addr: Multiaddr) -> Self {
+        Self::Addr(addr)
+    }
+}
+
+impl From<PeerId> for ConnectionTarget {
+    fn from(peer_id: PeerId) -> Self {
+        Self::PeerId(peer_id)
+    }
+}
+
 /// Disconnected will use banning to disconnect a node. Disconnecting a single peer connection is
 /// not supported at the moment.
 pub struct Disconnector {
@@ -87,13 +105,21 @@ impl SwarmApi {
         self.roundtrip_times.insert(peer_id.clone(), rtt);
     }
 
-    pub fn connect(&mut self, address: Multiaddr) -> SubscriptionFuture<Result<(), String>> {
-        trace!("starting to connect to {}", address);
-        self.events.push_back(NetworkBehaviourAction::DialAddress {
-            address: address.clone(),
+    pub fn connect(&mut self, target: ConnectionTarget) -> SubscriptionFuture<Result<(), String>> {
+        trace!("Connecting to {:?}", target);
+
+        self.events.push_back(match target {
+            ConnectionTarget::Addr(ref addr) => NetworkBehaviourAction::DialAddress {
+                address: addr.clone(),
+            },
+            ConnectionTarget::PeerId(ref id) => NetworkBehaviourAction::DialPeer {
+                peer_id: id.clone(),
+                condition: DialPeerCondition::Disconnected,
+            },
         });
+
         self.connect_registry
-            .create_subscription(address.into(), None)
+            .create_subscription(target.into(), None)
     }
 
     pub fn disconnect(&mut self, address: Multiaddr) -> Option<Disconnector> {
@@ -157,6 +183,8 @@ impl NetworkBehaviour for SwarmApi {
         self.connections.insert(addr.clone(), peer_id.clone());
         self.connect_registry
             .finish_subscription(addr.clone().into(), Ok(()));
+        self.connect_registry
+            .finish_subscription(peer_id.clone().into(), Ok(()));
     }
 
     fn inject_connected(&mut self, _peer_id: &PeerId) {
@@ -262,7 +290,7 @@ mod tests {
         };
 
         let peer2 = async move {
-            let future = swarm2.connect(rx.next().await.unwrap());
+            let future = swarm2.connect(rx.next().await.unwrap().into());
 
             let poll_swarm = async move {
                 loop {

src/p2p/swarm.rs

@@ -3,7 +3,7 @@
 //! that contains them. `SubscriptionFuture` is the `Future` bound to pending `Subscription`s and
 //! sharing the same unique numeric identifier, the `SubscriptionId`.
 
-use crate::RepoEvent;
+use crate::{p2p::ConnectionTarget, RepoEvent};
 use async_std::future::Future;
 use async_std::task::{self, Context, Poll, Waker};
 use core::fmt::Debug;
@@ -12,7 +12,7 @@ use core::pin::Pin;
 use futures::channel::mpsc::Sender;
 use futures::lock::Mutex;
 use libipld::Cid;
-use libp2p::{kad::QueryId, Multiaddr};
+use libp2p::{kad::QueryId, Multiaddr, PeerId};
 use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::fmt;
@@ -29,8 +29,8 @@ static GLOBAL_REQ_COUNT: AtomicU64 = AtomicU64::new(0);
 /// The type of a request for subscription.
 #[derive(Debug, PartialEq, Eq, Hash, Clone)]
 pub enum RequestKind {
-    /// A request to connect to the given `Multiaddr`.
-    Connect(Multiaddr),
+    /// A request to connect to the given `Multiaddr` or `PeerId`.
+    Connect(ConnectionTarget),
     /// A request to obtain a `Block` with a specific `Cid`.
     GetBlock(Cid),
     /// A DHT request to Kademlia.
@@ -41,7 +41,19 @@ pub enum RequestKind {
 
 impl From<Multiaddr> for RequestKind {
     fn from(addr: Multiaddr) -> Self {
-        Self::Connect(addr)
+        Self::Connect(ConnectionTarget::Addr(addr))
+    }
+}
+
+impl From<PeerId> for RequestKind {
+    fn from(peer_id: PeerId) -> Self {
+        Self::Connect(ConnectionTarget::PeerId(peer_id))
+    }
+}
+
+impl From<ConnectionTarget> for RequestKind {
+    fn from(target: ConnectionTarget) -> Self {
+        Self::Connect(target)
     }
 }
 
@@ -60,7 +72,7 @@ impl From<QueryId> for RequestKind {
 impl fmt::Display for RequestKind {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            Self::Connect(addr) => write!(fmt, "Connect to {}", addr),
+            Self::Connect(tgt) => write!(fmt, "Connect to {:?}", tgt),
             Self::GetBlock(cid) => write!(fmt, "Obtain block {}", cid),
             Self::KadQuery(id) => write!(fmt, "Kad request {:?}", id),
             #[cfg(test)]

src/subscription.rs

@@ -1,31 +1,40 @@
 use ipfs::Node;
+use libp2p::PeerId;
 
 // Make sure two instances of ipfs can be connected.
 #[async_std::test]
-async fn connect_two_nodes() {
+async fn connect_two_nodes_by_addr() {
     let node_a = Node::new("a").await;
     let node_b = Node::new("b").await;
 
     let (_, b_addrs) = node_b.identity().await.unwrap();
     assert!(!b_addrs.is_empty());
 
-    let mut connected = None;
+    let mut connected = false;
 
     for addr in b_addrs {
-        println!("trying {}", addr);
-        match node_a.connect(addr.clone()).await {
-            Ok(_) => {
-                connected = Some(addr);
-                break;
-            }
-            Err(e) => {
-                println!("Failed connecting to {}: {}", addr, e);
-            }
+        if node_a.connect(addr.clone()).await.is_ok() {
+            connected = true;
+            break;
         }
     }
 
-    let connected = connected.expect("Failed to connect to anything");
-    println!("connected to {}", connected);
+    assert!(connected);
+}
+
+// Make sure two instances of ipfs can be connected by `PeerId`.
+#[async_std::test]
+async fn connect_two_nodes_by_peer_id() {
+    let node_a = Node::new("a").await;
+    let node_b = Node::new("b").await;
+
+    let (b_key, mut b_addrs) = node_b.identity().await.unwrap();
+    let b_id = PeerId::from_public_key(b_key);
+
+    while let Some(addr) = b_addrs.pop() {
+        node_a.add_peer(b_id.clone(), addr).await.unwrap();
+    }
+    node_a.connect(b_id).await.unwrap();
 }
 
 // More complicated one to the above; first node will have two listening addresses and the second