Upgrade libp2p 0.19.0

Cargo.toml

@@ -20,7 +20,7 @@ dirs = "2.0.2"
 domain = { git = "https://github.com/nlnetlabs/domain", rev="084964", features = ["resolv"] }
 futures = { version = "0.3.4", features = ["compat", "io-compat"] }
 libipld = { git = "https://github.com/ipfs-rust/rust-ipld", rev = "b2286c53c13f3eeec2a3766387f2926838e8e4c9", features = ["dag-pb", "dag-json"] }
-libp2p = "0.16.2"
+libp2p = "0.18.1"
 log = "0.4.8"
 multibase = "0.8.0"
 multihash = "0.10.1"
@@ -45,7 +45,7 @@ members = [ "bitswap", "http", "examples" ]
 ctr = { git = "https://github.com/koivunej/stream-ciphers.git", branch = "ctr128-64to128" }
 
 # these are needed for the floodsub local originated messages to be seen by subscribers
-libp2p = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }
-libp2p-core = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }
-libp2p-swarm = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }
-libp2p-floodsub = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }
+# libp2p = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }
+# libp2p-core = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }
+# libp2p-swarm = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }
+# libp2p-floodsub = { git = "https://github.com/koivunej/rust-libp2p.git", branch = "v0.16.2-with-floodsub-opts" }

bitswap/Cargo.toml

@@ -14,8 +14,8 @@ async-trait = "0.1.29"
 fnv = "1.0.6"
 futures = "0.3.4"
 libipld = "0.1.0"
-libp2p-core = "0.16.0"
-libp2p-swarm = "0.16.1"
+libp2p-core = "0.18.0"
+libp2p-swarm = "0.18.1"
 log = "0.4.8"
 multihash = "0.10.1"
 prost = "0.6.1"

bitswap/src/behaviour.rs

@@ -13,9 +13,11 @@ use fnv::FnvHashSet;
 use futures::task::Context;
 use futures::task::Poll;
 use libipld::cid::Cid;
-use libp2p_core::{ConnectedPoint, Multiaddr, PeerId};
+use libp2p_core::{connection::ConnectionId, Multiaddr, PeerId};
 use libp2p_swarm::protocols_handler::{IntoProtocolsHandler, OneShotHandler, ProtocolsHandler};
-use libp2p_swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
+use libp2p_swarm::{
+    DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
+};
 use std::collections::{HashMap, VecDeque};
 
 /// Network behaviour that handles sending and receiving IPFS blocks.
@@ -94,10 +96,11 @@ impl<TStrategy> Bitswap<TStrategy> {
         debug!("bitswap: connect");
         if self.target_peers.insert(peer_id.clone()) {
             debug!("  queuing dial_peer to {}", peer_id.to_base58());
-            self.events
-                .push_back(NetworkBehaviourAction::DialPeer { peer_id });
+            self.events.push_back(NetworkBehaviourAction::DialPeer {
+                peer_id,
+                condition: DialPeerCondition::Disconnected,
+            });
         }
-        debug!("");
     }
 
     /// Sends a block to the peer.
@@ -111,10 +114,12 @@ impl<TStrategy> Bitswap<TStrategy> {
             .expect("Peer not in ledger?!");
         let message = ledger.send_block(block);
         debug!("  queuing block for {}", peer_id.to_base58());
-        self.events.push_back(NetworkBehaviourAction::SendEvent {
-            peer_id,
-            event: message,
-        });
+        self.events
+            .push_back(NetworkBehaviourAction::NotifyHandler {
+                peer_id,
+                event: message,
+                handler: NotifyHandler::Any,
+            });
         debug!("");
     }
 
@@ -127,10 +132,12 @@ impl<TStrategy> Bitswap<TStrategy> {
                 message.want_block(cid, *priority);
             }
             debug!("  queuing wanted blocks");
-            self.events.push_back(NetworkBehaviourAction::SendEvent {
-                peer_id,
-                event: message,
-            });
+            self.events
+                .push_back(NetworkBehaviourAction::NotifyHandler {
+                    peer_id,
+                    event: message,
+                    handler: NotifyHandler::Any,
+                });
         }
     }
 
@@ -142,10 +149,12 @@ impl<TStrategy> Bitswap<TStrategy> {
         for (peer_id, ledger) in self.connected_peers.iter_mut() {
             let message = ledger.want_block(&cid, priority);
             debug!("  queuing want for {}", peer_id.to_base58());
-            self.events.push_back(NetworkBehaviourAction::SendEvent {
-                peer_id: peer_id.to_owned(),
-                event: message,
-            });
+            self.events
+                .push_back(NetworkBehaviourAction::NotifyHandler {
+                    peer_id: peer_id.to_owned(),
+                    event: message,
+                    handler: NotifyHandler::Any,
+                });
         }
         self.wanted_blocks.insert(cid, priority);
         debug!("");
@@ -163,7 +172,11 @@ impl<TStrategy> Bitswap<TStrategy> {
                 let peer_id = peer_id.to_owned();
                 debug!("  queuing cancel for {}", peer_id.to_base58());
                 self.events
-                    .push_back(NetworkBehaviourAction::SendEvent { peer_id, event });
+                    .push_back(NetworkBehaviourAction::NotifyHandler {
+                        peer_id,
+                        event,
+                        handler: NotifyHandler::Any,
+                    });
             }
         }
         self.wanted_blocks.remove(cid);
@@ -185,25 +198,19 @@ impl<TStrategy: Strategy> NetworkBehaviour for Bitswap<TStrategy> {
         Vec::new()
     }
 
-    fn inject_connected(&mut self, peer_id: PeerId, cp: ConnectedPoint) {
-        debug!("bitswap: inject_connected");
-        debug!("  peer_id: {}", peer_id.to_base58());
-        debug!("  connected_point: {:?}", cp);
+    fn inject_connected(&mut self, peer_id: &PeerId) {
+        debug!("bitswap: inject_connected {}", peer_id);
         let ledger = Ledger::new();
         self.connected_peers.insert(peer_id.clone(), ledger);
-        self.send_want_list(peer_id);
-        debug!("");
+        self.send_want_list(peer_id.clone());
     }
 
-    fn inject_disconnected(&mut self, peer_id: &PeerId, cp: ConnectedPoint) {
-        debug!("bitswap: inject_disconnected {:?}", cp);
-        debug!("  peer_id: {}", peer_id.to_base58());
-        debug!("  connected_point: {:?}", cp);
-        debug!("");
+    fn inject_disconnected(&mut self, peer_id: &PeerId) {
+        debug!("bitswap: inject_disconnected {:?}", peer_id);
         //self.connected_peers.remove(peer_id);
     }
 
-    fn inject_node_event(&mut self, source: PeerId, event: InnerMessage) {
+    fn inject_event(&mut self, source: PeerId, _connection: ConnectionId, event: InnerMessage) {
         debug!("bitswap: inject_node_event");
         debug!("{:?}", event);
         let message = match event {
@@ -232,7 +239,7 @@ impl<TStrategy: Strategy> NetworkBehaviour for Bitswap<TStrategy> {
                 .process_want(source.clone(), cid.to_owned(), *priority);
         }
         // TODO: Remove cancelled `Want` events from the queue.
-        // TODO: Remove cancelled blocks from `SendEvent`.
+        // TODO: Remove cancelled blocks from `NotifyHandler`.
         debug!("");
     }
 
@@ -242,19 +249,33 @@ impl<TStrategy: Strategy> NetworkBehaviour for Bitswap<TStrategy> {
     {
         // TODO concat messages to same destination to reduce traffic.
         if let Some(event) = self.events.pop_front() {
-            if let NetworkBehaviourAction::SendEvent { peer_id, event } = event {
+            if let NetworkBehaviourAction::NotifyHandler {
+                peer_id,
+                event,
+                handler,
+            } = event
+            {
                 match self.connected_peers.get_mut(&peer_id) {
                     None => {
-                        debug!("  requeueing send event to {}", peer_id.to_base58());
+                        debug!("  requeueing send event to {}", peer_id);
+                        // FIXME: I wonder if this should be
                         self.events
-                            .push_back(NetworkBehaviourAction::SendEvent { peer_id, event })
+                            .push_back(NetworkBehaviourAction::NotifyHandler {
+                                peer_id,
+                                event,
+                                handler,
+                            })
                     }
                     Some(ref mut ledger) => {
                         // FIXME: this is a bit early to update stats as the block hasn't been sent
                         // to anywhere at this point.
                         ledger.update_outgoing_stats(&event);
-                        debug!("  send_message to {}", peer_id.to_base58());
-                        return Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event });
+                        debug!("  send_message to {}", peer_id);
+                        return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
+                            peer_id,
+                            event,
+                            handler,
+                        });
                     }
                 }
             } else {

src/lib.rs

@@ -17,10 +17,11 @@ use futures::stream::Fuse;
 pub use libipld::cid::Cid;
 use libipld::cid::Codec;
 pub use libipld::ipld::Ipld;
-pub use libp2p::core::{ConnectedPoint, Multiaddr, PeerId, PublicKey};
+pub use libp2p::core::{connection::ListenerId, ConnectedPoint, Multiaddr, PeerId, PublicKey};
 pub use libp2p::identity::Keypair;
 
 use std::borrow::Borrow;
+use std::collections::HashMap;
 use std::fmt;
 use std::future::Future;
 use std::marker::PhantomData;
@@ -245,6 +246,8 @@ enum IpfsEvent {
     PubsubSubscribed(OneshotSender<Vec<String>>),
     WantList(Option<PeerId>, OneshotSender<Vec<(Cid, bitswap::Priority)>>),
     BitswapStats(OneshotSender<BitswapStats>),
+    AddListeningAddress(Multiaddr, Channel<()>),
+    RemoveListeningAddress(Multiaddr, Channel<()>),
     Exit,
 }
 
@@ -298,6 +301,7 @@ impl<Types: IpfsTypes> UninitializedIpfs<Types> {
             repo_events: repo_events.fuse(),
             from_facade: receiver.fuse(),
             swarm,
+            listening_addresses: HashMap::new(),
         };
 
         let UninitializedIpfs {
@@ -538,6 +542,33 @@ impl<Types: IpfsTypes> Ipfs<Types> {
         Ok(rx.await?)
     }
 
+    /// Add a given multiaddr as a listening address. Will fail if the address is unsupported, or
+    /// if it is already being listened on. Currently will invoke `Swarm::listen_on` internally,
+    /// keep the ListenerId for later `remove_listening_address` use in a HashMap.
+    pub async fn add_listening_address(&self, addr: Multiaddr) -> Result<(), Error> {
+        let (tx, rx) = oneshot_channel();
+
+        self.to_task
+            .clone()
+            .send(IpfsEvent::AddListeningAddress(addr, tx))
+            .await?;
+
+        rx.await?
+    }
+
+    /// Stop listening on a previously added listening address. Fails if the address is not being
+    /// listened to.
+    pub async fn remove_listening_address(&self, addr: Multiaddr) -> Result<(), Error> {
+        let (tx, rx) = oneshot_channel();
+
+        self.to_task
+            .clone()
+            .send(IpfsEvent::RemoveListeningAddress(addr, tx))
+            .await?;
+
+        rx.await?
+    }
+
     /// Exit daemon.
     pub async fn exit_daemon(mut self) {
         // FIXME: this is a stopgap measure needed while repo is part of the struct Ipfs instead of
@@ -556,14 +587,15 @@ struct IpfsFuture<Types: SwarmTypes> {
     swarm: TSwarm<Types>,
     repo_events: Fuse<Receiver<RepoEvent>>,
     from_facade: Fuse<Receiver<IpfsEvent>>,
+    listening_addresses: HashMap<Multiaddr, ListenerId>,
 }
 
 impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
     type Output = ();
 
     fn poll(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
         use futures::Stream;
-        use libp2p::{swarm::SwarmEvent, Swarm};
+        use libp2p::Swarm;
 
         // begin by polling the swarm so that initially it'll first have chance to bind listeners
         // and such. TODO: this no longer needs to be a swarm event but perhaps we should
@@ -573,7 +605,7 @@ impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
 
         loop {
             loop {
-                let inner = {
+                let _inner = {
                     let next = self.swarm.next_event();
                     futures::pin_mut!(next);
                     match next.poll(ctx) {
@@ -583,19 +615,7 @@ impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
                     }
                 };
                 done = false;
-                match inner {
-                    SwarmEvent::Behaviour(()) => {}
-                    SwarmEvent::Connected(_peer_id) => {}
-                    SwarmEvent::Disconnected(_peer_id) => {}
-                    SwarmEvent::NewListenAddr(_addr) => {}
-                    SwarmEvent::ExpiredListenAddr(_addr) => {}
-                    SwarmEvent::UnreachableAddr {
-                        peer_id: _peer_id,
-                        address: _address,
-                        error: _error,
-                    } => {}
-                    SwarmEvent::StartConnect(_peer_id) => {}
-                }
+                // the _inner can be useful for debugging
             }
 
             // temporary pinning of the receivers should be safe as we are pinning through the
@@ -676,6 +696,31 @@ impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
                         let wantlist = self.swarm.bitswap().local_wantlist();
                         let _ = ret.send((stats, peers, wantlist).into());
                     }
+                    IpfsEvent::AddListeningAddress(addr, ret) => {
+                        let added = match Swarm::listen_on(&mut self.swarm, addr.clone()) {
+                            Ok(id) => {
+                                self.listening_addresses.insert(addr, id);
+                                Ok(())
+                            }
+                            Err(e) => Err(Error::from(e)),
+                        };
+
+                        let _ = ret.send(added);
+                    }
+                    IpfsEvent::RemoveListeningAddress(addr, ret) => {
+                        let removed = if let Some(id) = self.listening_addresses.remove(&addr) {
+                            Swarm::remove_listener(&mut self.swarm, id).map_err(|_: ()| {
+                                format_err!(
+                                    "Failed to remove previously added listening address: {}",
+                                    addr
+                                )
+                            })
+                        } else {
+                            Err(format_err!("Address was not listened to before: {}", addr))
+                        };
+
+                        let _ = ret.send(removed);
+                    }
                     IpfsEvent::Exit => {
                         // FIXME: we could do a proper teardown
                         return Poll::Ready(());

src/p2p/behaviour.rs

@@ -113,13 +113,13 @@ impl<TSwarmTypes: SwarmTypes> NetworkBehaviourEventProcess<PingEvent> for Behavi
                 peer,
                 result: Result::Ok(PingSuccess::Pong),
             } => {
-                log::trace!("ping: pong from {}", peer.to_base58());
+                log::trace!("ping: pong from {}", peer);
             }
             PingEvent {
                 peer,
                 result: Result::Err(PingFailure::Timeout),
             } => {
-                log::trace!("ping: timeout to {}", peer.to_base58());
+                log::trace!("ping: timeout to {}", peer);
                 self.remove_peer(&peer);
             }
             PingEvent {
@@ -203,7 +203,7 @@ impl<TSwarmTypes: SwarmTypes> Behaviour<TSwarmTypes> {
     }
 
     pub fn connections(&self) -> Vec<Connection> {
-        self.swarm.connections().cloned().collect()
+        self.swarm.connections()
     }
 
     pub fn connect(&mut self, addr: Multiaddr) -> SubscriptionFuture<Result<(), String>> {