mirror of
https://github.com/fluencelabs/rust-libp2p
synced 2025-06-12 09:31:20 +00:00
Add Swarm::next_extended (#1374)
* Add Swarm::next_extended * Fix ipfs-kad example * Fix tests * Renames
This commit is contained in:
Pierre Krieger
GitHub
309
swarm/src/lib.rs
309
swarm/src/lib.rs
@ -93,7 +93,7 @@ use libp2p_core::{
|
||||
};
|
||||
use registry::{Addresses, AddressIntoIter};
|
||||
use smallvec::SmallVec;
|
||||
use std::{error, fmt, io, ops::{Deref, DerefMut}, pin::Pin, task::{Context, Poll}};
|
||||
use std::{error, fmt, ops::{Deref, DerefMut}, pin::Pin, task::{Context, Poll}};
|
||||
use std::collections::HashSet;
|
||||
|
||||
/// Contains the state of the network, plus the way it should behave.
|
||||
@ -107,6 +107,33 @@ pub type Swarm<TTransport, TBehaviour, TConnInfo = PeerId> = ExpandedSwarm<
|
||||
TConnInfo,
|
||||
>;
|
||||
|
||||
/// Event generated by the `Swarm`.
|
||||
#[derive(Debug)]
|
||||
pub enum SwarmEvent<TBvEv> {
|
||||
/// Event generated by the `NetworkBehaviour`.
|
||||
Behaviour(TBvEv),
|
||||
/// 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. `None` if we don't know in advance which peer
|
||||
/// we were trying to reach.
|
||||
peer_id: Option<PeerId>,
|
||||
/// Address that we failed to reach.
|
||||
address: Multiaddr,
|
||||
/// Error that has been encountered.
|
||||
error: Box<dyn error::Error + Send>,
|
||||
},
|
||||
/// Startng to try to reach the given peer.
|
||||
StartConnect(PeerId),
|
||||
}
|
||||
|
||||
/// Contains the state of the network, plus the way it should behave.
|
||||
pub struct ExpandedSwarm<TTransport, TBehaviour, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo = PeerId>
|
||||
where
|
||||
@ -305,6 +332,166 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
|
||||
pub fn unban_peer_id(me: &mut Self, peer_id: PeerId) {
|
||||
me.banned_peers.remove(&peer_id);
|
||||
}
|
||||
|
||||
/// 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<TBehaviour::OutEvent> {
|
||||
future::poll_fn(move |cx| ExpandedSwarm::poll_next_event(Pin::new(self), cx)).await
|
||||
}
|
||||
|
||||
/// Returns the next event produced by the [`NetworkBehaviour`].
|
||||
pub async fn next(&mut self) -> TBehaviour::OutEvent {
|
||||
future::poll_fn(move |cx| {
|
||||
loop {
|
||||
let event = futures::ready!(ExpandedSwarm::poll_next_event(Pin::new(self), cx));
|
||||
if let SwarmEvent::Behaviour(event) = event {
|
||||
return Poll::Ready(event);
|
||||
}
|
||||
}
|
||||
}).await
|
||||
}
|
||||
|
||||
/// Internal function used by everything event-related.
|
||||
///
|
||||
/// Polls the `Swarm` for the next event.
|
||||
fn poll_next_event(mut self: Pin<&mut Self>, cx: &mut Context)
|
||||
-> Poll<SwarmEvent<TBehaviour::OutEvent>>
|
||||
{
|
||||
// We use a `this` variable because the compiler can't mutably borrow multiple times
|
||||
// across a `Deref`.
|
||||
let this = &mut *self;
|
||||
|
||||
loop {
|
||||
let mut network_not_ready = false;
|
||||
|
||||
match this.network.poll(cx) {
|
||||
Poll::Pending => network_not_ready = true,
|
||||
Poll::Ready(NetworkEvent::NodeEvent { conn_info, event }) => {
|
||||
this.behaviour.inject_node_event(conn_info.peer_id().clone(), event);
|
||||
},
|
||||
Poll::Ready(NetworkEvent::Connected { conn_info, endpoint }) => {
|
||||
if this.banned_peers.contains(conn_info.peer_id()) {
|
||||
this.network.peer(conn_info.peer_id().clone())
|
||||
.into_connected()
|
||||
.expect("the Network just notified us that we were connected; QED")
|
||||
.close();
|
||||
} else {
|
||||
this.behaviour.inject_connected(conn_info.peer_id().clone(), endpoint);
|
||||
return Poll::Ready(SwarmEvent::Connected(conn_info.peer_id().clone()));
|
||||
}
|
||||
},
|
||||
Poll::Ready(NetworkEvent::NodeClosed { conn_info, endpoint, .. }) => {
|
||||
this.behaviour.inject_disconnected(conn_info.peer_id(), endpoint);
|
||||
return Poll::Ready(SwarmEvent::Disconnected(conn_info.peer_id().clone()));
|
||||
},
|
||||
Poll::Ready(NetworkEvent::Replaced { new_info, closed_endpoint, endpoint, .. }) => {
|
||||
this.behaviour.inject_replaced(new_info.peer_id().clone(), closed_endpoint, endpoint);
|
||||
},
|
||||
Poll::Ready(NetworkEvent::IncomingConnection(incoming)) => {
|
||||
let handler = this.behaviour.new_handler();
|
||||
incoming.accept(handler.into_node_handler_builder());
|
||||
},
|
||||
Poll::Ready(NetworkEvent::NewListenerAddress { listen_addr, .. }) => {
|
||||
if !this.listened_addrs.contains(&listen_addr) {
|
||||
this.listened_addrs.push(listen_addr.clone())
|
||||
}
|
||||
this.behaviour.inject_new_listen_addr(&listen_addr);
|
||||
return Poll::Ready(SwarmEvent::NewListenAddr(listen_addr));
|
||||
}
|
||||
Poll::Ready(NetworkEvent::ExpiredListenerAddress { listen_addr, .. }) => {
|
||||
this.listened_addrs.retain(|a| a != &listen_addr);
|
||||
this.behaviour.inject_expired_listen_addr(&listen_addr);
|
||||
return Poll::Ready(SwarmEvent::ExpiredListenAddr(listen_addr));
|
||||
}
|
||||
Poll::Ready(NetworkEvent::ListenerClosed { listener_id, .. }) =>
|
||||
this.behaviour.inject_listener_closed(listener_id),
|
||||
Poll::Ready(NetworkEvent::ListenerError { listener_id, error }) =>
|
||||
this.behaviour.inject_listener_error(listener_id, &error),
|
||||
Poll::Ready(NetworkEvent::IncomingConnectionError { .. }) => {},
|
||||
Poll::Ready(NetworkEvent::DialError { peer_id, multiaddr, error, new_state }) => {
|
||||
this.behaviour.inject_addr_reach_failure(Some(&peer_id), &multiaddr, &error);
|
||||
if let network::PeerState::NotConnected = new_state {
|
||||
this.behaviour.inject_dial_failure(&peer_id);
|
||||
}
|
||||
return Poll::Ready(SwarmEvent::UnreachableAddr {
|
||||
peer_id: Some(peer_id.clone()),
|
||||
address: multiaddr,
|
||||
error: Box::new(error),
|
||||
});
|
||||
},
|
||||
Poll::Ready(NetworkEvent::UnknownPeerDialError { multiaddr, error, .. }) => {
|
||||
this.behaviour.inject_addr_reach_failure(None, &multiaddr, &error);
|
||||
return Poll::Ready(SwarmEvent::UnreachableAddr {
|
||||
peer_id: None,
|
||||
address: multiaddr,
|
||||
error: Box::new(error),
|
||||
});
|
||||
},
|
||||
}
|
||||
|
||||
// Try to deliver pending event.
|
||||
if let Some((id, pending)) = this.send_event_to_complete.take() {
|
||||
if let Some(mut peer) = this.network.peer(id.clone()).into_connected() {
|
||||
match peer.poll_ready_event(cx) {
|
||||
Poll::Ready(()) => peer.start_send_event(pending),
|
||||
Poll::Pending => {
|
||||
this.send_event_to_complete = Some((id, pending));
|
||||
return Poll::Pending
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let behaviour_poll = {
|
||||
let mut parameters = SwarmPollParameters {
|
||||
local_peer_id: &mut this.network.local_peer_id(),
|
||||
supported_protocols: &this.supported_protocols,
|
||||
listened_addrs: &this.listened_addrs,
|
||||
external_addrs: &this.external_addrs
|
||||
};
|
||||
this.behaviour.poll(cx, &mut parameters)
|
||||
};
|
||||
|
||||
match behaviour_poll {
|
||||
Poll::Pending if network_not_ready => return Poll::Pending,
|
||||
Poll::Pending => (),
|
||||
Poll::Ready(NetworkBehaviourAction::GenerateEvent(event)) => {
|
||||
return Poll::Ready(SwarmEvent::Behaviour(event))
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::DialAddress { address }) => {
|
||||
let _ = ExpandedSwarm::dial_addr(&mut *this, address);
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }) => {
|
||||
if this.banned_peers.contains(&peer_id) {
|
||||
this.behaviour.inject_dial_failure(&peer_id);
|
||||
} else {
|
||||
ExpandedSwarm::dial(&mut *this, peer_id.clone());
|
||||
return Poll::Ready(SwarmEvent::StartConnect(peer_id))
|
||||
}
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }) => {
|
||||
if let Some(mut peer) = this.network.peer(peer_id.clone()).into_connected() {
|
||||
if let Poll::Ready(()) = peer.poll_ready_event(cx) {
|
||||
peer.start_send_event(event);
|
||||
} else {
|
||||
debug_assert!(this.send_event_to_complete.is_none());
|
||||
this.send_event_to_complete = Some((peer_id, event));
|
||||
return Poll::Pending;
|
||||
}
|
||||
}
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }) => {
|
||||
for addr in this.network.address_translation(&address) {
|
||||
if this.external_addrs.iter().all(|a| *a != addr) {
|
||||
this.behaviour.inject_new_external_addr(&addr);
|
||||
}
|
||||
this.external_addrs.add(addr);
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<TTransport, TBehaviour, TMuxer, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo> Stream for
|
||||
@ -340,125 +527,13 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
|
||||
<NodeHandlerWrapper<<THandler as IntoProtocolsHandler>::Handler> as NodeHandler>::OutboundOpenInfo: Send + 'static, // TODO: shouldn't be necessary
|
||||
TConnInfo: ConnectionInfo<PeerId = PeerId> + fmt::Debug + Clone + Send + 'static,
|
||||
{
|
||||
type Item = Result<TBehaviour::OutEvent, io::Error>;
|
||||
type Item = TBehaviour::OutEvent;
|
||||
|
||||
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
|
||||
// We use a `this` variable because the compiler can't mutably borrow multiple times
|
||||
// across a `Deref`.
|
||||
let this = &mut *self;
|
||||
|
||||
loop {
|
||||
let mut network_not_ready = false;
|
||||
|
||||
match this.network.poll(cx) {
|
||||
Poll::Pending => network_not_ready = true,
|
||||
Poll::Ready(NetworkEvent::NodeEvent { conn_info, event }) => {
|
||||
this.behaviour.inject_node_event(conn_info.peer_id().clone(), event);
|
||||
},
|
||||
Poll::Ready(NetworkEvent::Connected { conn_info, endpoint }) => {
|
||||
if this.banned_peers.contains(conn_info.peer_id()) {
|
||||
this.network.peer(conn_info.peer_id().clone())
|
||||
.into_connected()
|
||||
.expect("the Network just notified us that we were connected; QED")
|
||||
.close();
|
||||
} else {
|
||||
this.behaviour.inject_connected(conn_info.peer_id().clone(), endpoint);
|
||||
}
|
||||
},
|
||||
Poll::Ready(NetworkEvent::NodeClosed { conn_info, endpoint, .. }) => {
|
||||
this.behaviour.inject_disconnected(conn_info.peer_id(), endpoint);
|
||||
},
|
||||
Poll::Ready(NetworkEvent::Replaced { new_info, closed_endpoint, endpoint, .. }) => {
|
||||
this.behaviour.inject_replaced(new_info.peer_id().clone(), closed_endpoint, endpoint);
|
||||
},
|
||||
Poll::Ready(NetworkEvent::IncomingConnection(incoming)) => {
|
||||
let handler = this.behaviour.new_handler();
|
||||
incoming.accept(handler.into_node_handler_builder());
|
||||
},
|
||||
Poll::Ready(NetworkEvent::NewListenerAddress { listen_addr, .. }) => {
|
||||
if !this.listened_addrs.contains(&listen_addr) {
|
||||
this.listened_addrs.push(listen_addr.clone())
|
||||
}
|
||||
this.behaviour.inject_new_listen_addr(&listen_addr);
|
||||
}
|
||||
Poll::Ready(NetworkEvent::ExpiredListenerAddress { listen_addr, .. }) => {
|
||||
this.listened_addrs.retain(|a| a != &listen_addr);
|
||||
this.behaviour.inject_expired_listen_addr(&listen_addr);
|
||||
}
|
||||
Poll::Ready(NetworkEvent::ListenerClosed { listener_id, .. }) =>
|
||||
this.behaviour.inject_listener_closed(listener_id),
|
||||
Poll::Ready(NetworkEvent::ListenerError { listener_id, error }) =>
|
||||
this.behaviour.inject_listener_error(listener_id, &error),
|
||||
Poll::Ready(NetworkEvent::IncomingConnectionError { .. }) => {},
|
||||
Poll::Ready(NetworkEvent::DialError { peer_id, multiaddr, error, new_state }) => {
|
||||
this.behaviour.inject_addr_reach_failure(Some(&peer_id), &multiaddr, &error);
|
||||
if let network::PeerState::NotConnected = new_state {
|
||||
this.behaviour.inject_dial_failure(&peer_id);
|
||||
}
|
||||
},
|
||||
Poll::Ready(NetworkEvent::UnknownPeerDialError { multiaddr, error, .. }) => {
|
||||
this.behaviour.inject_addr_reach_failure(None, &multiaddr, &error);
|
||||
},
|
||||
}
|
||||
|
||||
// Try to deliver pending event.
|
||||
if let Some((id, pending)) = this.send_event_to_complete.take() {
|
||||
if let Some(mut peer) = this.network.peer(id.clone()).into_connected() {
|
||||
match peer.poll_ready_event(cx) {
|
||||
Poll::Ready(()) => peer.start_send_event(pending),
|
||||
Poll::Pending => {
|
||||
this.send_event_to_complete = Some((id, pending));
|
||||
return Poll::Pending
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let behaviour_poll = {
|
||||
let mut parameters = SwarmPollParameters {
|
||||
local_peer_id: &mut this.network.local_peer_id(),
|
||||
supported_protocols: &this.supported_protocols,
|
||||
listened_addrs: &this.listened_addrs,
|
||||
external_addrs: &this.external_addrs
|
||||
};
|
||||
this.behaviour.poll(cx, &mut parameters)
|
||||
};
|
||||
|
||||
match behaviour_poll {
|
||||
Poll::Pending if network_not_ready => return Poll::Pending,
|
||||
Poll::Pending => (),
|
||||
Poll::Ready(NetworkBehaviourAction::GenerateEvent(event)) => {
|
||||
return Poll::Ready(Some(Ok(event)))
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::DialAddress { address }) => {
|
||||
let _ = ExpandedSwarm::dial_addr(&mut *this, address);
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }) => {
|
||||
if this.banned_peers.contains(&peer_id) {
|
||||
this.behaviour.inject_dial_failure(&peer_id);
|
||||
} else {
|
||||
ExpandedSwarm::dial(&mut *this, peer_id);
|
||||
}
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }) => {
|
||||
if let Some(mut peer) = this.network.peer(peer_id.clone()).into_connected() {
|
||||
if let Poll::Ready(()) = peer.poll_ready_event(cx) {
|
||||
peer.start_send_event(event);
|
||||
} else {
|
||||
debug_assert!(this.send_event_to_complete.is_none());
|
||||
this.send_event_to_complete = Some((peer_id, event));
|
||||
return Poll::Pending;
|
||||
}
|
||||
}
|
||||
},
|
||||
Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }) => {
|
||||
for addr in this.network.address_translation(&address) {
|
||||
if this.external_addrs.iter().all(|a| *a != addr) {
|
||||
this.behaviour.inject_new_external_addr(&addr);
|
||||
}
|
||||
this.external_addrs.add(addr)
|
||||
}
|
||||
},
|
||||
let event = futures::ready!(ExpandedSwarm::poll_next_event(self.as_mut(), cx));
|
||||
if let SwarmEvent::Behaviour(event) = event {
|
||||
return Poll::Ready(Some(event));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user