diff --git a/core/src/network.rs b/core/src/network.rs index 52d0da80187..94b7df2d326 100644 --- a/core/src/network.rs +++ b/core/src/network.rs @@ -52,7 +52,6 @@ use fnv::{FnvHashMap}; use futures::{prelude::*, future}; use std::{ collections::hash_map, - convert::TryFrom as _, error, fmt, hash::Hash, @@ -518,7 +517,7 @@ where // A pending outgoing connection to a known peer failed. let mut attempt = dialing.remove(&peer_id).expect("by (1)"); - let num_remain = u32::try_from(attempt.next.len()).unwrap(); + let num_remain = attempt.next.len(); let failed_addr = attempt.current.clone(); let opts = diff --git a/core/src/network/event.rs b/core/src/network/event.rs index 233b35fd9d3..835b6e643cd 100644 --- a/core/src/network/event.rs +++ b/core/src/network/event.rs @@ -42,7 +42,7 @@ use crate::{ transport::{Transport, TransportError}, }; use futures::prelude::*; -use std::{error, fmt, hash::Hash, num::NonZeroU32}; +use std::{error, fmt, hash::Hash}; /// Event that can happen on the `Network`. pub enum NetworkEvent<'a, TTrans, TInEvent, TOutEvent, THandler, TConnInfo, TPeerId> @@ -88,7 +88,7 @@ where /// A new connection arrived on a listener. IncomingConnection(IncomingConnectionEvent<'a, TTrans, TInEvent, TOutEvent, THandler, TConnInfo, TPeerId>), - /// An error happened on a connection during its initial handshake. + /// A new connection was arriving on a listener, but an error happened when negotiating it. /// /// This can include, for example, an error during the handshake of the encryption layer, or /// the connection unexpectedly closed. @@ -105,9 +105,8 @@ where ConnectionEstablished { /// The newly established connection. connection: EstablishedConnection<'a, TInEvent, TConnInfo, TPeerId>, - /// The total number of established connections to the same peer, including the one that - /// has just been opened. - num_established: NonZeroU32, + /// The total number of established connections to the same peer. + num_established: usize, }, /// An established connection to a peer has encountered an error. @@ -119,13 +118,13 @@ where /// The error that occurred. error: ConnectionError<::Error>, /// The remaining number of established connections to the same peer. - num_established: u32, + num_established: usize, }, /// A dialing attempt to an address of a peer failed. DialError { /// The number of remaining dialing attempts. - attempts_remaining: u32, + attempts_remaining: usize, /// Id of the peer we were trying to dial. peer_id: TPeerId, diff --git a/core/tests/network_dial_error.rs b/core/tests/network_dial_error.rs index c01cebddfa2..e4603f34e25 100644 --- a/core/tests/network_dial_error.rs +++ b/core/tests/network_dial_error.rs @@ -258,7 +258,7 @@ fn multiple_addresses_err() { assert_eq!(attempts_remaining, 0); return Poll::Ready(Ok(())); } else { - assert_eq!(attempts_remaining, addresses.len() as u32); + assert_eq!(attempts_remaining, addresses.len()); } }, Poll::Ready(_) => unreachable!(), diff --git a/swarm/src/lib.rs b/swarm/src/lib.rs index b9a2e6dc94a..6e92aab793b 100644 --- a/swarm/src/lib.rs +++ b/swarm/src/lib.rs @@ -79,30 +79,24 @@ pub use protocols_handler::{ SubstreamProtocol }; -use protocols_handler::{ - NodeHandlerWrapperBuilder, - NodeHandlerWrapperError, -}; +use protocols_handler::NodeHandlerWrapperBuilder; use futures::{ prelude::*, executor::{ThreadPool, ThreadPoolBuilder}, stream::FusedStream, }; use libp2p_core::{ - ConnectedPoint, Executor, Transport, Multiaddr, Negotiated, PeerId, connection::{ - ConnectionError, ConnectionId, ConnectionInfo, EstablishedConnection, IntoConnectionHandler, ListenerId, - PendingConnectionError, Substream }, transport::{TransportError, boxed::Boxed as BoxTransport}, @@ -122,7 +116,6 @@ use registry::{Addresses, AddressIntoIter}; use smallvec::SmallVec; use std::{error, fmt, hash::Hash, io, ops::{Deref, DerefMut}, pin::Pin, task::{Context, Poll}}; use std::collections::HashSet; -use std::num::NonZeroU32; use upgrade::UpgradeInfoSend as _; /// Contains the state of the network, plus the way it should behave. @@ -142,110 +135,29 @@ pub type NegotiatedSubstream = Negotiated>; /// Event generated by the `Swarm`. #[derive(Debug)] -pub enum SwarmEvent { +pub enum SwarmEvent { /// Event generated by the `NetworkBehaviour`. Behaviour(TBvEv), - /// A connection to the given peer has been opened. - ConnectionEstablished { - /// Identity of the peer that we have connected to. - peer_id: PeerId, - /// Endpoint of the connection that has been opened. - endpoint: ConnectedPoint, - /// Number of established connections to this peer, including the one that has just been - /// opened. - num_established: NonZeroU32, - }, - /// A connection with the given peer has been closed. - ConnectionClosed { - /// Identity of the peer that we have connected to. - peer_id: PeerId, - /// Endpoint of the connection that has been closed. - endpoint: ConnectedPoint, - /// Number of other remaining connections to this same peer. - num_established: u32, - /// Reason for the disconnection. - cause: ConnectionError>, - }, - /// A new connection arrived on a listener and is in the process of protocol negotiation. - /// - /// A corresponding [`ConnectionEstablished`](SwarmEvent::ConnectionEstablished), - /// [`BannedPeer`](SwarmEvent::BannedPeer), or - /// [`IncomingConnectionError`](SwarmEvent::IncomingConnectionError) event will later be - /// generated for this connection. - IncomingConnection { - /// Local connection address. - /// This address has been earlier reported with a [`NewListenAddr`](SwarmEvent::NewListenAddr) - /// event. - local_addr: Multiaddr, - /// Address used to send back data to the remote. - send_back_addr: Multiaddr, - }, - /// An error happened on a connection during its initial handshake. - /// - /// This can include, for example, an error during the handshake of the encryption layer, or - /// the connection unexpectedly closed. - IncomingConnectionError { - /// Local connection address. - /// This address has been earlier reported with a [`NewListenAddr`](SwarmEvent::NewListenAddr) - /// event. - local_addr: Multiaddr, - /// Address used to send back data to the remote. - send_back_addr: Multiaddr, - /// The error that happened. - error: PendingConnectionError, - }, - /// We connected to a peer, but we immediately closed the connection because that peer is banned. - BannedPeer { - /// Identity of the banned peer. - peer_id: PeerId, - /// Endpoint of the connection that has been closed. - endpoint: ConnectedPoint, - }, - /// Starting to try to reach the given peer. - /// - /// We are trying to connect to this peer until a [`ConnectionEstablished`](SwarmEvent::ConnectionEstablished) - /// event is reported, or a [`UnreachableAddr`](SwarmEvent::UnreachableAddr) event is reported - /// with `attempts_remaining` equal to 0. - Dialing(PeerId), + /// We are now connected to the given peer. + Connected(PeerId), + /// We are now disconnected from the given peer. + Disconnected(PeerId), + /// One of our listeners has reported a new local listening address. + NewListenAddr(Multiaddr), + /// One of our listeners has reported the expiration of a listening address. + ExpiredListenAddr(Multiaddr), /// Tried to dial an address but it ended up being unreachaable. UnreachableAddr { - /// `PeerId` that we were trying to reach. - peer_id: PeerId, - /// Address that we failed to reach. - address: Multiaddr, - /// Error that has been encountered. - error: PendingConnectionError, - /// Number of remaining connection attempts that are being tried for this peer. - attempts_remaining: u32, - }, - /// Tried to dial an address but it ended up being unreachaable. - /// Contrary to `UnreachableAddr`, we don't know the identity of the peer that we were trying - /// to reach. - UnknownPeerUnreachableAddr { + /// `PeerId` that we were trying to reach. `None` if we don't know in advance which peer + /// we were trying to reach. + peer_id: Option, /// Address that we failed to reach. address: Multiaddr, /// Error that has been encountered. - error: PendingConnectionError, - }, - /// One of our listeners has reported a new local listening address. - NewListenAddr(Multiaddr), - /// One of our listeners has reported the expiration of a listening address. - ExpiredListenAddr(Multiaddr), - /// One of the listeners gracefully closed. - ListenerClosed { - /// The addresses that the listener was listening on. These addresses are now considered - /// expired, similar to if a [`ExpiredListenAddr`](SwarmEvent::ExpiredListenAddr) event - /// has been generated for each of them. - addresses: Vec, - /// Reason for the closure. Contains `Ok(())` if the stream produced `None`, or `Err` - /// if the stream produced an error. - reason: Result<(), io::Error>, - }, - /// One of the listeners reported a non-fatal error. - ListenerError { - /// The listener error. - error: io::Error, + error: Box, }, + /// Startng to try to reach the given peer. + StartConnect(PeerId), } /// Contains the state of the network, plus the way it should behave. @@ -318,15 +230,14 @@ where { } -impl +impl ExpandedSwarm where TBehaviour: NetworkBehaviour, TInEvent: Clone + Send + 'static, TOutEvent: Send + 'static, TConnInfo: ConnectionInfo + fmt::Debug + Clone + Send + 'static, THandler: IntoProtocolsHandler + Send + 'static, - THandler::Handler: ProtocolsHandler, - THandleErr: error::Error + Send + 'static, + THandler::Handler: ProtocolsHandler, { /// Builds a new `Swarm`. pub fn new(transport: TTransport, behaviour: TBehaviour, local_peer_id: PeerId) -> Self @@ -449,7 +360,7 @@ where TBehaviour: NetworkBehaviour, /// Returns the next event that happens in the `Swarm`. /// /// Includes events from the `NetworkBehaviour` but also events about the connections status. - pub async fn next_event(&mut self) -> SwarmEvent { + pub async fn next_event(&mut self) -> SwarmEvent { future::poll_fn(move |cx| ExpandedSwarm::poll_next_event(Pin::new(self), cx)).await } @@ -469,7 +380,7 @@ where TBehaviour: NetworkBehaviour, /// /// Polls the `Swarm` for the next event. fn poll_next_event(mut self: Pin<&mut Self>, cx: &mut Context) - -> Poll> + -> Poll> { // We use a `this` variable because the compiler can't mutably borrow multiple times // across a `Deref`. @@ -487,62 +398,38 @@ where TBehaviour: NetworkBehaviour, this.behaviour.inject_event(peer, connection, event); }, Poll::Ready(NetworkEvent::ConnectionEstablished { connection, num_established }) => { - let peer_id = connection.peer_id().clone(); - let endpoint = connection.endpoint().clone(); - if this.banned_peers.contains(&peer_id) { - this.network.peer(peer_id.clone()) + let peer = connection.peer_id().clone(); + if this.banned_peers.contains(&peer) { + this.network.peer(peer) .into_connected() .expect("the Network just notified us that we were connected; QED") .disconnect(); - return Poll::Ready(SwarmEvent::BannedPeer { - peer_id, - endpoint, - }); - } else if num_established.get() == 1 { - this.behaviour.inject_connected(peer_id.clone(), endpoint.clone()); - return Poll::Ready(SwarmEvent::ConnectionEstablished { - peer_id, - endpoint, - num_established, - }); + } else if num_established == 1 { + let endpoint = connection.endpoint().clone(); + this.behaviour.inject_connected(peer.clone(), endpoint); + return Poll::Ready(SwarmEvent::Connected(peer)); } else { // For now, secondary connections are not explicitly reported to // the behaviour. A behaviour only gets awareness of the // connections via the events emitted from the connection handlers. log::trace!("Secondary connection established: {:?}; Total (peer): {}.", connection.connected(), num_established); - return Poll::Ready(SwarmEvent::ConnectionEstablished { - peer_id, - endpoint, - num_established, - }); } }, Poll::Ready(NetworkEvent::ConnectionError { connected, error, num_established }) => { log::debug!("Connection {:?} closed by {:?}", connected, error); - let peer_id = connected.peer_id().clone(); - let endpoint = connected.endpoint; if num_established == 0 { - this.behaviour.inject_disconnected(&peer_id, endpoint.clone()); + let peer = connected.peer_id().clone(); + let endpoint = connected.endpoint; + this.behaviour.inject_disconnected(&peer, endpoint); + return Poll::Ready(SwarmEvent::Disconnected(peer)); } - return Poll::Ready(SwarmEvent::ConnectionClosed { - peer_id, - endpoint, - cause: error, - num_established, - }); }, Poll::Ready(NetworkEvent::IncomingConnection(incoming)) => { let handler = this.behaviour.new_handler(); - let local_addr = incoming.local_addr().clone(); - let send_back_addr = incoming.send_back_addr().clone(); if let Err(e) = incoming.accept(handler.into_node_handler_builder()) { log::warn!("Incoming connection rejected: {:?}", e); } - return Poll::Ready(SwarmEvent::IncomingConnection { - local_addr, - send_back_addr, - }); }, Poll::Ready(NetworkEvent::NewListenerAddress { listener_id, listen_addr }) => { log::debug!("Listener {:?}; New address: {:?}", listener_id, listen_addr); @@ -564,24 +451,11 @@ where TBehaviour: NetworkBehaviour, this.behaviour.inject_expired_listen_addr(addr); } this.behaviour.inject_listener_closed(listener_id); - return Poll::Ready(SwarmEvent::ListenerClosed { - addresses, - reason, - }); } - Poll::Ready(NetworkEvent::ListenerError { listener_id, error }) => { - this.behaviour.inject_listener_error(listener_id, &error); - return Poll::Ready(SwarmEvent::ListenerError { - error, - }); - }, - Poll::Ready(NetworkEvent::IncomingConnectionError { local_addr, send_back_addr, error }) => { + Poll::Ready(NetworkEvent::ListenerError { listener_id, error }) => + this.behaviour.inject_listener_error(listener_id, &error), + Poll::Ready(NetworkEvent::IncomingConnectionError { error, .. }) => { log::debug!("Incoming connection failed: {:?}", error); - return Poll::Ready(SwarmEvent::IncomingConnectionError { - local_addr, - send_back_addr, - error, - }); }, Poll::Ready(NetworkEvent::DialError { peer_id, multiaddr, error, attempts_remaining }) => { log::debug!( @@ -592,19 +466,19 @@ where TBehaviour: NetworkBehaviour, this.behaviour.inject_dial_failure(&peer_id); } return Poll::Ready(SwarmEvent::UnreachableAddr { - peer_id, + peer_id: Some(peer_id.clone()), address: multiaddr, - error, - attempts_remaining, + error: Box::new(error), }); }, Poll::Ready(NetworkEvent::UnknownPeerDialError { multiaddr, error, .. }) => { log::debug!("Connection attempt to address {:?} of unknown peer failed with {:?}", multiaddr, error); this.behaviour.inject_addr_reach_failure(None, &multiaddr, &error); - return Poll::Ready(SwarmEvent::UnknownPeerUnreachableAddr { + return Poll::Ready(SwarmEvent::UnreachableAddr { + peer_id: None, address: multiaddr, - error, + error: Box::new(error), }); }, } @@ -668,7 +542,7 @@ where TBehaviour: NetworkBehaviour, this.behaviour.inject_dial_failure(&peer_id); } else { ExpandedSwarm::dial(&mut *this, peer_id.clone()); - return Poll::Ready(SwarmEvent::Dialing(peer_id)) + return Poll::Ready(SwarmEvent::StartConnect(peer_id)) } }, Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }) => {