use crate::muxing::StreamMuxer;
use crate::{
Endpoint, Multiaddr, PeerId,
nodes::{
collection::{
CollectionEvent,
CollectionNodeAccept,
CollectionReachEvent,
CollectionStream,
ReachAttemptId
},
handled_node::{
HandledNodeError,
NodeHandler
},
handled_node_tasks::IntoNodeHandler,
node::Substream
},
nodes::listeners::{ListenersEvent, ListenersStream},
transport::Transport,
transport::TransportError,
};
use fnv::FnvHashMap;
use futures::{prelude::*, future};
use std::{
collections::hash_map::{Entry, OccupiedEntry},
error,
fmt,
hash::Hash,
num::NonZeroUsize,
};
mod tests;
pub struct RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId = PeerId>
where
TTrans: Transport,
{
listeners: ListenersStream<TTrans>,
active_nodes: CollectionStream<TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TPeerId>, THandlerErr, (), TPeerId>,
reach_attempts: ReachAttempts<TPeerId>,
incoming_limit: Option<u32>,
}
impl<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId> fmt::Debug for
RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport + fmt::Debug,
TPeerId: fmt::Debug + Eq + Hash,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
f.debug_struct("ReachAttempts")
.field("listeners", &self.listeners)
.field("active_nodes", &self.active_nodes)
.field("reach_attempts", &self.reach_attempts)
.field("incoming_limit", &self.incoming_limit)
.finish()
}
}
struct ReachAttempts<TPeerId> {
local_peer_id: TPeerId,
out_reach_attempts: FnvHashMap<TPeerId, OutReachAttempt>,
other_reach_attempts: Vec<(ReachAttemptId, ConnectedPoint)>,
connected_points: FnvHashMap<TPeerId, ConnectedPoint>,
}
impl<TPeerId> fmt::Debug for ReachAttempts<TPeerId>
where
TPeerId: fmt::Debug + Eq + Hash,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
f.debug_struct("ReachAttempts")
.field("local_peer_id", &self.local_peer_id)
.field("out_reach_attempts", &self.out_reach_attempts)
.field("other_reach_attempts", &self.other_reach_attempts)
.field("connected_points", &self.connected_points)
.finish()
}
}
#[derive(Debug, Clone)]
struct OutReachAttempt {
id: ReachAttemptId,
cur_attempted: Multiaddr,
next_attempts: Vec<Multiaddr>,
}
pub enum RawSwarmEvent<'a, TTrans: 'a, TInEvent: 'a, TOutEvent: 'a, THandler: 'a, THandlerErr: 'a, TPeerId: 'a = PeerId>
where
TTrans: Transport,
{
ListenerClosed {
listen_addr: Multiaddr,
listener: TTrans::Listener,
result: Result<(), <TTrans::Listener as Stream>::Error>,
},
IncomingConnection(IncomingConnectionEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>),
IncomingConnectionError {
listen_addr: Multiaddr,
send_back_addr: Multiaddr,
error: IncomingError<TTrans::Error>,
},
Connected {
peer_id: TPeerId,
endpoint: ConnectedPoint,
},
Replaced {
peer_id: TPeerId,
closed_endpoint: ConnectedPoint,
endpoint: ConnectedPoint,
},
NodeClosed {
peer_id: TPeerId,
endpoint: ConnectedPoint,
error: HandledNodeError<THandlerErr>,
},
DialError {
new_state: PeerState,
peer_id: TPeerId,
multiaddr: Multiaddr,
error: RawSwarmReachError<TTrans::Error, TPeerId>,
},
UnknownPeerDialError {
multiaddr: Multiaddr,
error: UnknownPeerDialErr<TTrans::Error>,
handler: THandler,
},
NodeEvent {
peer_id: TPeerId,
event: TOutEvent,
},
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId> fmt::Debug for
RawSwarmEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TOutEvent: fmt::Debug,
TTrans: Transport,
TTrans::Error: fmt::Debug,
THandlerErr: fmt::Debug,
TPeerId: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
match *self {
RawSwarmEvent::ListenerClosed { ref listen_addr, ref result, .. } => {
f.debug_struct("ListenerClosed")
.field("listen_addr", listen_addr)
.field("result", result)
.finish()
}
RawSwarmEvent::IncomingConnection( IncomingConnectionEvent { ref listen_addr, ref send_back_addr, .. } ) => {
f.debug_struct("IncomingConnection")
.field("listen_addr", listen_addr)
.field("send_back_addr", send_back_addr)
.finish()
}
RawSwarmEvent::IncomingConnectionError { ref listen_addr, ref send_back_addr, ref error} => {
f.debug_struct("IncomingConnectionError")
.field("listen_addr", listen_addr)
.field("send_back_addr", send_back_addr)
.field("error", error)
.finish()
}
RawSwarmEvent::Connected { ref peer_id, ref endpoint } => {
f.debug_struct("Connected")
.field("peer_id", peer_id)
.field("endpoint", endpoint)
.finish()
}
RawSwarmEvent::Replaced { ref peer_id, ref closed_endpoint, ref endpoint } => {
f.debug_struct("Replaced")
.field("peer_id", peer_id)
.field("closed_endpoint", closed_endpoint)
.field("endpoint", endpoint)
.finish()
}
RawSwarmEvent::NodeClosed { ref peer_id, ref endpoint, ref error } => {
f.debug_struct("NodeClosed")
.field("peer_id", peer_id)
.field("endpoint", endpoint)
.field("error", error)
.finish()
}
RawSwarmEvent::DialError { ref new_state, ref peer_id, ref multiaddr, ref error } => {
f.debug_struct("DialError")
.field("new_state", new_state)
.field("peer_id", peer_id)
.field("multiaddr", multiaddr)
.field("error", error)
.finish()
}
RawSwarmEvent::UnknownPeerDialError { ref multiaddr, ref error, .. } => {
f.debug_struct("UnknownPeerDialError")
.field("multiaddr", multiaddr)
.field("error", error)
.finish()
}
RawSwarmEvent::NodeEvent { ref peer_id, ref event } => {
f.debug_struct("NodeEvent")
.field("peer_id", peer_id)
.field("event", event)
.finish()
}
}
}
}
#[derive(Debug)]
enum InternalReachErr<TTransErr, TPeerId> {
Transport(TransportError<TTransErr>),
PeerIdMismatch {
obtained: TPeerId,
},
FoundLocalPeerId,
}
impl<TTransErr, TPeerId> fmt::Display for InternalReachErr<TTransErr, TPeerId>
where
TTransErr: fmt::Display,
TPeerId: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
InternalReachErr::Transport(err) => write!(f, "{}", err),
InternalReachErr::PeerIdMismatch { obtained } => {
write!(f, "Peer ID mismatch, obtained: {:?}", obtained)
},
InternalReachErr::FoundLocalPeerId => {
write!(f, "Remote has the same PeerId as us")
}
}
}
}
impl<TTransErr, TPeerId> error::Error for InternalReachErr<TTransErr, TPeerId>
where
TTransErr: error::Error + 'static,
TPeerId: fmt::Debug,
{
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
InternalReachErr::Transport(err) => Some(err),
InternalReachErr::PeerIdMismatch { .. } => None,
InternalReachErr::FoundLocalPeerId => None,
}
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum PeerState {
Connected,
Dialing {
num_pending_addresses: NonZeroUsize,
},
NotConnected,
}
#[derive(Debug)]
pub enum RawSwarmReachError<TTransErr, TPeerId> {
Transport(TransportError<TTransErr>),
PeerIdMismatch {
obtained: TPeerId,
}
}
impl<TTransErr, TPeerId> fmt::Display for RawSwarmReachError<TTransErr, TPeerId>
where
TTransErr: fmt::Display,
TPeerId: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
RawSwarmReachError::Transport(err) => write!(f, "{}", err),
RawSwarmReachError::PeerIdMismatch { obtained } => {
write!(f, "Peer ID mismatch, obtained: {:?}", obtained)
},
}
}
}
impl<TTransErr, TPeerId> error::Error for RawSwarmReachError<TTransErr, TPeerId>
where
TTransErr: error::Error + 'static,
TPeerId: fmt::Debug,
{
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
RawSwarmReachError::Transport(err) => Some(err),
RawSwarmReachError::PeerIdMismatch { .. } => None,
}
}
}
#[derive(Debug)]
pub enum UnknownPeerDialErr<TTransErr> {
Transport(TransportError<TTransErr>),
FoundLocalPeerId,
}
impl<TTransErr> fmt::Display for UnknownPeerDialErr<TTransErr>
where TTransErr: fmt::Display
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
UnknownPeerDialErr::Transport(err) => write!(f, "{}", err),
UnknownPeerDialErr::FoundLocalPeerId => {
write!(f, "Unknown peer has same PeerId as us")
},
}
}
}
impl<TTransErr> error::Error for UnknownPeerDialErr<TTransErr>
where TTransErr: error::Error + 'static
{
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
UnknownPeerDialErr::Transport(err) => Some(err),
UnknownPeerDialErr::FoundLocalPeerId => None,
}
}
}
#[derive(Debug)]
pub enum IncomingError<TTransErr> {
Transport(TransportError<TTransErr>),
DeniedLowerPriority,
FoundLocalPeerId,
}
impl<TTransErr> fmt::Display for IncomingError<TTransErr>
where TTransErr: fmt::Display
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
IncomingError::Transport(err) => write!(f, "{}", err),
IncomingError::DeniedLowerPriority => {
write!(f, "Denied because of lower priority")
},
IncomingError::FoundLocalPeerId => {
write!(f, "Incoming connection has same PeerId as us")
},
}
}
}
impl<TTransErr> error::Error for IncomingError<TTransErr>
where TTransErr: error::Error + 'static
{
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
IncomingError::Transport(err) => Some(err),
IncomingError::DeniedLowerPriority => None,
IncomingError::FoundLocalPeerId => None,
}
}
}
pub struct IncomingConnectionEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where TTrans: Transport
{
upgrade: TTrans::ListenerUpgrade,
local_peer_id: TPeerId,
listen_addr: Multiaddr,
send_back_addr: Multiaddr,
active_nodes: &'a mut CollectionStream<TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TPeerId>, THandlerErr, (), TPeerId>,
other_reach_attempts: &'a mut Vec<(ReachAttemptId, ConnectedPoint)>,
}
impl<'a, TTrans, TInEvent, TOutEvent, TMuxer, THandler, THandlerErr, TPeerId>
IncomingConnectionEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport<Output = (TPeerId, TMuxer)>,
TTrans::Error: Send + 'static,
TTrans::ListenerUpgrade: Send + 'static,
THandler: IntoNodeHandler<TPeerId> + Send + 'static,
THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
<THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static,
THandlerErr: error::Error + Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
TPeerId: fmt::Debug + Eq + Hash + Clone + Send + 'static,
{
#[inline]
pub fn accept(self, handler: THandler) {
self.accept_with_builder(|_| handler)
}
pub fn accept_with_builder<TBuilder>(self, builder: TBuilder)
where TBuilder: FnOnce(IncomingInfo<'_>) -> THandler
{
let connected_point = self.to_connected_point();
let handler = builder(self.info());
let local_peer_id = self.local_peer_id;
let upgrade = self.upgrade
.map_err(|err| InternalReachErr::Transport(TransportError::Other(err)))
.and_then(move |(peer_id, muxer)| {
if peer_id == local_peer_id {
Err(InternalReachErr::FoundLocalPeerId)
} else {
Ok((peer_id, muxer))
}
});
let id = self.active_nodes.add_reach_attempt(upgrade, handler);
self.other_reach_attempts.push((
id,
connected_point,
));
}
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
IncomingConnectionEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where TTrans: Transport
{
#[inline]
pub fn info(&self) -> IncomingInfo<'_> {
IncomingInfo {
listen_addr: &self.listen_addr,
send_back_addr: &self.send_back_addr,
}
}
#[inline]
pub fn listen_addr(&self) -> &Multiaddr {
&self.listen_addr
}
#[inline]
pub fn send_back_addr(&self) -> &Multiaddr {
&self.send_back_addr
}
#[inline]
pub fn to_connected_point(&self) -> ConnectedPoint {
self.info().to_connected_point()
}
}
#[derive(Debug, Clone)]
pub enum ConnectedPoint {
Dialer {
address: Multiaddr,
},
Listener {
listen_addr: Multiaddr,
send_back_addr: Multiaddr,
},
}
impl<'a> From<&'a ConnectedPoint> for Endpoint {
#[inline]
fn from(endpoint: &'a ConnectedPoint) -> Endpoint {
endpoint.to_endpoint()
}
}
impl From<ConnectedPoint> for Endpoint {
#[inline]
fn from(endpoint: ConnectedPoint) -> Endpoint {
endpoint.to_endpoint()
}
}
impl ConnectedPoint {
#[inline]
pub fn to_endpoint(&self) -> Endpoint {
match *self {
ConnectedPoint::Dialer { .. } => Endpoint::Dialer,
ConnectedPoint::Listener { .. } => Endpoint::Listener,
}
}
#[inline]
pub fn is_dialer(&self) -> bool {
match *self {
ConnectedPoint::Dialer { .. } => true,
ConnectedPoint::Listener { .. } => false,
}
}
#[inline]
pub fn is_listener(&self) -> bool {
match *self {
ConnectedPoint::Dialer { .. } => false,
ConnectedPoint::Listener { .. } => true,
}
}
}
#[derive(Debug, Copy, Clone)]
pub struct IncomingInfo<'a> {
pub listen_addr: &'a Multiaddr,
pub send_back_addr: &'a Multiaddr,
}
impl<'a> IncomingInfo<'a> {
#[inline]
pub fn to_connected_point(&self) -> ConnectedPoint {
ConnectedPoint::Listener {
listen_addr: self.listen_addr.clone(),
send_back_addr: self.send_back_addr.clone(),
}
}
}
impl<TTrans, TInEvent, TOutEvent, TMuxer, THandler, THandlerErr, TPeerId>
RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport + Clone,
TMuxer: StreamMuxer,
THandler: IntoNodeHandler<TPeerId> + Send + 'static,
THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
<THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static,
THandlerErr: error::Error + Send + 'static,
TPeerId: fmt::Debug + Eq + Hash + Clone + AsRef<[u8]> + Send + 'static,
{
#[inline]
pub fn new(transport: TTrans, local_peer_id: TPeerId) -> Self {
RawSwarm {
listeners: ListenersStream::new(transport),
active_nodes: CollectionStream::new(),
reach_attempts: ReachAttempts {
local_peer_id,
out_reach_attempts: Default::default(),
other_reach_attempts: Vec::new(),
connected_points: Default::default(),
},
incoming_limit: None,
}
}
#[inline]
pub fn new_with_incoming_limit(transport: TTrans,
local_peer_id: TPeerId, incoming_limit: Option<u32>) -> Self
{
RawSwarm {
incoming_limit,
listeners: ListenersStream::new(transport),
active_nodes: CollectionStream::new(),
reach_attempts: ReachAttempts {
local_peer_id,
out_reach_attempts: Default::default(),
other_reach_attempts: Vec::new(),
connected_points: Default::default(),
},
}
}
#[inline]
pub fn transport(&self) -> &TTrans {
self.listeners.transport()
}
#[inline]
pub fn listen_on(&mut self, addr: Multiaddr) -> Result<Multiaddr, TransportError<TTrans::Error>> {
self.listeners.listen_on(addr)
}
#[inline]
pub fn listeners(&self) -> impl Iterator<Item = &Multiaddr> {
self.listeners.listeners()
}
#[inline]
pub fn incoming_limit(&self) -> Option<u32> {
self.incoming_limit
}
#[inline]
pub fn nat_traversal<'a>(
&'a self,
observed_addr: &'a Multiaddr,
) -> impl Iterator<Item = Multiaddr> + 'a
where TMuxer: 'a,
THandler: 'a,
{
self.listeners()
.flat_map(move |server| self.transport().nat_traversal(server, observed_addr))
}
#[inline]
pub fn local_peer_id(&self) -> &TPeerId {
&self.reach_attempts.local_peer_id
}
pub fn dial(&mut self, addr: Multiaddr, handler: THandler) -> Result<(), TransportError<TTrans::Error>>
where
TTrans: Transport<Output = (TPeerId, TMuxer)>,
TTrans::Error: Send + 'static,
TTrans::Dial: Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
{
let local_peer_id = self.reach_attempts.local_peer_id.clone();
let future = self.transport().clone().dial(addr.clone())?
.map_err(|err| InternalReachErr::Transport(TransportError::Other(err)))
.and_then(move |(peer_id, muxer)| {
if peer_id == local_peer_id {
Err(InternalReachErr::FoundLocalPeerId)
} else {
Ok((peer_id, muxer))
}
});
let connected_point = ConnectedPoint::Dialer { address: addr };
let reach_id = self.active_nodes.add_reach_attempt(future, handler);
self.reach_attempts.other_reach_attempts.push((reach_id, connected_point));
Ok(())
}
#[deprecated(note = "Use incoming_negotiated().count() instead")]
#[inline]
pub fn num_incoming_negotiated(&self) -> usize {
self.reach_attempts.other_reach_attempts
.iter()
.filter(|&(_, endpoint)| endpoint.is_listener())
.count()
}
#[inline]
pub fn incoming_negotiated(&self) -> impl Iterator<Item = IncomingInfo<'_>> {
self.reach_attempts
.other_reach_attempts
.iter()
.filter_map(|&(_, ref endpoint)| {
match endpoint {
ConnectedPoint::Listener { listen_addr, send_back_addr } => {
Some(IncomingInfo {
listen_addr,
send_back_addr,
})
},
ConnectedPoint::Dialer { .. } => None,
}
})
}
#[inline]
pub fn broadcast_event(&mut self, event: &TInEvent)
where TInEvent: Clone,
{
self.active_nodes.broadcast_event(event)
}
#[inline]
pub fn peer(&mut self, peer_id: TPeerId) -> Peer<'_, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId> {
if peer_id == self.reach_attempts.local_peer_id {
return Peer::LocalNode;
}
if self.active_nodes.peer_mut(&peer_id).is_some() {
return Peer::Connected(PeerConnected {
active_nodes: &mut self.active_nodes,
peer_id,
connected_points: &mut self.reach_attempts.connected_points,
out_reach_attempts: &mut self.reach_attempts.out_reach_attempts,
});
}
debug_assert!(!self.reach_attempts.connected_points.contains_key(&peer_id));
if self.reach_attempts.out_reach_attempts.get_mut(&peer_id).is_some() {
return Peer::PendingConnect(PeerPendingConnect {
attempt: match self.reach_attempts.out_reach_attempts.entry(peer_id.clone()) {
Entry::Occupied(e) => e,
Entry::Vacant(_) => panic!("we checked for Some just above"),
},
active_nodes: &mut self.active_nodes,
});
}
Peer::NotConnected(PeerNotConnected {
nodes: self,
peer_id,
})
}
fn start_dial_out(&mut self, peer_id: TPeerId, handler: THandler, first: Multiaddr, rest: Vec<Multiaddr>)
where
TTrans: Transport<Output = (TPeerId, TMuxer)>,
TTrans::Dial: Send + 'static,
TTrans::Error: Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
{
let reach_id = match self.transport().clone().dial(first.clone()) {
Ok(fut) => {
let expected_peer_id = peer_id.clone();
let fut = fut
.map_err(|err| InternalReachErr::Transport(TransportError::Other(err)))
.and_then(move |(actual_peer_id, muxer)| {
if actual_peer_id == expected_peer_id {
Ok((actual_peer_id, muxer))
} else {
Err(InternalReachErr::PeerIdMismatch { obtained: actual_peer_id })
}
});
self.active_nodes.add_reach_attempt(fut, handler)
},
Err(err) => {
let fut = future::err(InternalReachErr::Transport(err));
self.active_nodes.add_reach_attempt(fut, handler)
},
};
let former = self.reach_attempts.out_reach_attempts.insert(
peer_id,
OutReachAttempt {
id: reach_id,
cur_attempted: first,
next_attempts: rest,
},
);
debug_assert!(former.is_none());
}
pub fn poll(&mut self) -> Async<RawSwarmEvent<'_, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>>
where
TTrans: Transport<Output = (TPeerId, TMuxer)>,
TTrans::Error: Send + 'static,
TTrans::Dial: Send + 'static,
TTrans::ListenerUpgrade: Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
THandler: IntoNodeHandler<TPeerId> + Send + 'static,
THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
<THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static,
THandlerErr: error::Error + Send + 'static,
{
match self.incoming_limit {
Some(x) if self.incoming_negotiated().count() >= (x as usize)
=> (),
_ => {
match self.listeners.poll() {
Async::NotReady => (),
Async::Ready(ListenersEvent::Incoming {
upgrade, listen_addr, send_back_addr }) =>
{
let event = IncomingConnectionEvent {
upgrade,
local_peer_id:
self.reach_attempts.local_peer_id.clone(),
listen_addr,
send_back_addr,
active_nodes: &mut self.active_nodes,
other_reach_attempts: &mut self.reach_attempts.other_reach_attempts,
};
return Async::Ready(RawSwarmEvent::IncomingConnection(event));
},
Async::Ready(ListenersEvent::Closed {
listen_addr, listener, result }) =>
{
return Async::Ready(RawSwarmEvent::ListenerClosed {
listen_addr,
listener,
result,
});
}
}
}
}
let (action, out_event);
match self.active_nodes.poll() {
Async::NotReady => return Async::NotReady,
Async::Ready(CollectionEvent::NodeReached(reach_event)) => {
let (a, e) = handle_node_reached(&mut self.reach_attempts, reach_event);
action = a;
out_event = e;
}
Async::Ready(CollectionEvent::ReachError { id, error, handler }) => {
let (a, e) = handle_reach_error(&mut self.reach_attempts, id, error, handler);
action = a;
out_event = e;
}
Async::Ready(CollectionEvent::NodeClosed {
peer_id,
error,
..
}) => {
let endpoint = self.reach_attempts.connected_points.remove(&peer_id)
.expect("We insert into connected_points whenever a connection is \
opened and remove only when a connection is closed; the \
underlying API is guaranteed to always deliver a connection \
closed message after it has been opened, and no two closed \
messages; QED");
action = Default::default();
out_event = RawSwarmEvent::NodeClosed {
peer_id,
endpoint,
error,
};
}
Async::Ready(CollectionEvent::NodeEvent { peer, event }) => {
action = Default::default();
out_event = RawSwarmEvent::NodeEvent { peer_id: peer.id().clone(), event };
}
}
if let Some((peer_id, handler, first, rest)) = action.start_dial_out {
self.start_dial_out(peer_id, handler, first, rest);
}
if let Some((peer_id, interrupt)) = action.take_over {
let interrupted = self.active_nodes
.interrupt(interrupt)
.expect("take_over is guaranteed to be gathered from `out_reach_attempts`;
we insert in out_reach_attempts only when we call \
active_nodes.add_reach_attempt, and we remove only when we call \
interrupt or when a reach attempt succeeds or errors; therefore the \
out_reach_attempts should always be in sync with the actual \
attempts; QED");
self.active_nodes.peer_mut(&peer_id).unwrap().take_over(interrupted);
}
Async::Ready(out_event)
}
}
#[derive(Debug)]
#[must_use]
struct ActionItem<THandler, TPeerId> {
start_dial_out: Option<(TPeerId, THandler, Multiaddr, Vec<Multiaddr>)>,
take_over: Option<(TPeerId, ReachAttemptId)>,
}
impl<THandler, TPeerId> Default for ActionItem<THandler, TPeerId> {
fn default() -> Self {
ActionItem {
start_dial_out: None,
take_over: None,
}
}
}
fn handle_node_reached<'a, TTrans, TMuxer, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>(
reach_attempts: &mut ReachAttempts<TPeerId>,
event: CollectionReachEvent<'_, TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TPeerId>, THandlerErr, (), TPeerId>,
) -> (ActionItem<THandler, TPeerId>, RawSwarmEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>)
where
TTrans: Transport<Output = (TPeerId, TMuxer)> + Clone,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
TPeerId: fmt::Debug + Eq + Hash + Clone + AsRef<[u8]> + Send + 'static,
{
if let Some(in_pos) = reach_attempts
.other_reach_attempts
.iter()
.position(|i| i.0 == event.reach_attempt_id())
{
let (_, opened_endpoint) = reach_attempts.other_reach_attempts.swap_remove(in_pos);
let has_dial_prio = has_dial_prio(&reach_attempts.local_peer_id, event.peer_id());
if event.would_replace() && has_dial_prio {
if let Some(ConnectedPoint::Dialer { .. }) = reach_attempts.connected_points.get(event.peer_id()) {
if let ConnectedPoint::Listener { listen_addr, send_back_addr } = opened_endpoint {
return (Default::default(), RawSwarmEvent::IncomingConnectionError {
listen_addr,
send_back_addr,
error: IncomingError::DeniedLowerPriority,
});
}
}
}
let closed_endpoint = reach_attempts.connected_points.insert(event.peer_id().clone(), opened_endpoint.clone());
let action = if has_dial_prio {
if let Some(attempt) = reach_attempts.out_reach_attempts.get_mut(&event.peer_id()) {
debug_assert_ne!(attempt.id, event.reach_attempt_id());
attempt.next_attempts.clear();
}
ActionItem::default()
} else {
if let Some(attempt) = reach_attempts.out_reach_attempts.remove(&event.peer_id()) {
debug_assert_ne!(attempt.id, event.reach_attempt_id());
ActionItem {
take_over: Some((event.peer_id().clone(), attempt.id)),
.. Default::default()
}
} else {
ActionItem::default()
}
};
let (outcome, peer_id) = event.accept(());
if let CollectionNodeAccept::ReplacedExisting(()) = outcome {
let closed_endpoint = closed_endpoint
.expect("We insert into connected_points whenever a connection is opened and \
remove only when a connection is closed; the underlying API is \
guaranteed to always deliver a connection closed message after it has \
been opened, and no two closed messages; QED");
return (action, RawSwarmEvent::Replaced {
peer_id,
endpoint: opened_endpoint,
closed_endpoint,
});
} else {
return (action, RawSwarmEvent::Connected { peer_id, endpoint: opened_endpoint });
}
}
let is_outgoing_and_ok = if let Some(attempt) = reach_attempts.out_reach_attempts.get(event.peer_id()) {
attempt.id == event.reach_attempt_id()
} else {
false
};
if is_outgoing_and_ok {
let attempt = reach_attempts.out_reach_attempts.remove(event.peer_id())
.expect("is_outgoing_and_ok is true only if reach_attempts.out_reach_attempts.get(event.peer_id()) \
returned Some");
let opened_endpoint = ConnectedPoint::Dialer {
address: attempt.cur_attempted,
};
let closed_endpoint = reach_attempts.connected_points
.insert(event.peer_id().clone(), opened_endpoint.clone());
let (outcome, peer_id) = event.accept(());
if let CollectionNodeAccept::ReplacedExisting(()) = outcome {
let closed_endpoint = closed_endpoint
.expect("We insert into connected_points whenever a connection is opened and \
remove only when a connection is closed; the underlying API is guaranteed \
to always deliver a connection closed message after it has been opened, \
and no two closed messages; QED");
return (Default::default(), RawSwarmEvent::Replaced {
peer_id,
endpoint: opened_endpoint,
closed_endpoint,
});
} else {
return (Default::default(), RawSwarmEvent::Connected { peer_id, endpoint: opened_endpoint });
}
}
panic!("The API of collection guarantees that the id sent back in NodeReached (which is where \
we call handle_node_reached) is one that was passed to add_reach_attempt. Whenever we \
call add_reach_attempt, we also insert at the same time an entry either in \
out_reach_attempts or in other_reach_attempts. It is therefore guaranteed that we \
find back this ID in either of these two sets");
}
#[inline]
fn has_dial_prio<TPeerId>(local: &TPeerId, other: &TPeerId) -> bool
where
TPeerId: AsRef<[u8]>,
{
local.as_ref() < other.as_ref()
}
fn handle_reach_error<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>(
reach_attempts: &mut ReachAttempts<TPeerId>,
reach_id: ReachAttemptId,
error: InternalReachErr<TTrans::Error, TPeerId>,
handler: THandler,
) -> (ActionItem<THandler, TPeerId>, RawSwarmEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>)
where
TTrans: Transport,
TPeerId: Eq + Hash + Clone,
{
let out_reach_peer_id = reach_attempts
.out_reach_attempts
.iter()
.find(|(_, a)| a.id == reach_id)
.map(|(p, _)| p.clone());
if let Some(peer_id) = out_reach_peer_id {
let attempt = reach_attempts.out_reach_attempts.remove(&peer_id)
.expect("out_reach_peer_id is a key that is grabbed from out_reach_attempts");
let num_remain = attempt.next_attempts.len();
let failed_addr = attempt.cur_attempted.clone();
let new_state = if reach_attempts.connected_points.contains_key(&peer_id) {
PeerState::Connected
} else if num_remain == 0 {
PeerState::NotConnected
} else {
PeerState::Dialing {
num_pending_addresses: NonZeroUsize::new(num_remain)
.expect("We check that num_remain is not 0 right above; QED"),
}
};
let action = if !attempt.next_attempts.is_empty() {
let mut attempt = attempt;
let next_attempt = attempt.next_attempts.remove(0);
ActionItem {
start_dial_out: Some((peer_id.clone(), handler, next_attempt, attempt.next_attempts)),
.. Default::default()
}
} else {
Default::default()
};
let error = match error {
InternalReachErr::Transport(err) => RawSwarmReachError::Transport(err),
InternalReachErr::PeerIdMismatch { obtained } => {
RawSwarmReachError::PeerIdMismatch { obtained }
},
InternalReachErr::FoundLocalPeerId => {
unreachable!("We only generate FoundLocalPeerId within dial() or accept(); neither \
of these methods add an entry to out_reach_attempts; QED")
},
};
return (action, RawSwarmEvent::DialError {
new_state,
peer_id,
multiaddr: failed_addr,
error,
});
}
if let Some(in_pos) = reach_attempts
.other_reach_attempts
.iter()
.position(|i| i.0 == reach_id)
{
let (_, endpoint) = reach_attempts.other_reach_attempts.swap_remove(in_pos);
match endpoint {
ConnectedPoint::Dialer { address } => {
let error = match error {
InternalReachErr::Transport(err) => UnknownPeerDialErr::Transport(err),
InternalReachErr::FoundLocalPeerId => UnknownPeerDialErr::FoundLocalPeerId,
InternalReachErr::PeerIdMismatch { .. } => {
unreachable!("We only generate PeerIdMismatch within start_dial_out(),
which doesn't add any entry in other_reach_attempts; QED")
},
};
return (Default::default(), RawSwarmEvent::UnknownPeerDialError {
multiaddr: address,
error,
handler,
});
}
ConnectedPoint::Listener { listen_addr, send_back_addr } => {
let error = match error {
InternalReachErr::Transport(err) => IncomingError::Transport(err),
InternalReachErr::FoundLocalPeerId => IncomingError::FoundLocalPeerId,
InternalReachErr::PeerIdMismatch { .. } => {
unreachable!("We only generate PeerIdMismatch within start_dial_out(),
which doesn't add any entry in other_reach_attempts; QED")
},
};
return (Default::default(), RawSwarmEvent::IncomingConnectionError {
listen_addr,
send_back_addr,
error
});
}
}
}
panic!("The API of collection guarantees that the id sent back in ReachError events \
(which is where we call handle_reach_error) is one that was passed to \
add_reach_attempt. Whenever we call add_reach_attempt, we also insert \
at the same time an entry either in out_reach_attempts or in \
other_reach_attempts. It is therefore guaranteed that we find back this ID in \
either of these two sets");
}
pub enum Peer<'a, TTrans: 'a, TInEvent: 'a, TOutEvent: 'a, THandler: 'a, THandlerErr: 'a, TPeerId: 'a>
where
TTrans: Transport,
{
Connected(PeerConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>),
PendingConnect(PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>),
NotConnected(PeerNotConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>),
LocalNode,
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId> fmt::Debug for
Peer<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport,
TPeerId: Eq + Hash + fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
match *self {
Peer::Connected( PeerConnected { ref peer_id, ref connected_points, .. }) => {
f.debug_struct("Connected")
.field("peer_id", peer_id)
.field("connected_points", connected_points)
.finish()
}
Peer::PendingConnect( PeerPendingConnect { ref attempt, .. } ) => {
f.debug_struct("PendingConnect")
.field("attempt", attempt)
.finish()
}
Peer::NotConnected(PeerNotConnected { ref peer_id, .. }) => {
f.debug_struct("NotConnected")
.field("peer_id", peer_id)
.finish()
}
Peer::LocalNode => {
f.debug_struct("LocalNode")
.finish()
}
}
}
}
impl<'a, TTrans, TMuxer, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
Peer<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport<Output = (TPeerId, TMuxer)> + Clone,
TTrans::Error: Send + 'static,
TTrans::Dial: Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
THandler: IntoNodeHandler<TPeerId> + Send + 'static,
THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
<THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static,
THandlerErr: error::Error + Send + 'static,
TPeerId: fmt::Debug + Eq + Hash + Clone + AsRef<[u8]> + Send + 'static,
{
#[inline]
pub fn into_connected(self) -> Option<PeerConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>> {
match self {
Peer::Connected(peer) => Some(peer),
_ => None,
}
}
#[inline]
pub fn into_pending_connect(self) -> Option<PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>> {
match self {
Peer::PendingConnect(peer) => Some(peer),
_ => None,
}
}
#[inline]
pub fn into_not_connected(self) -> Option<PeerNotConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>> {
match self {
Peer::NotConnected(peer) => Some(peer),
_ => None,
}
}
#[inline]
pub fn or_connect(self, addr: Multiaddr, handler: THandler)
-> Result<PeerPotentialConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>, Self>
{
self.or_connect_with(move |_| addr, handler)
}
#[inline]
pub fn or_connect_with<TFn>(self, addr: TFn, handler: THandler)
-> Result<PeerPotentialConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>, Self>
where
TFn: FnOnce(&TPeerId) -> Multiaddr,
{
match self {
Peer::Connected(peer) => Ok(PeerPotentialConnect::Connected(peer)),
Peer::PendingConnect(peer) => Ok(PeerPotentialConnect::PendingConnect(peer)),
Peer::NotConnected(peer) => {
let addr = addr(&peer.peer_id);
Ok(PeerPotentialConnect::PendingConnect(peer.connect(addr, handler)))
},
Peer::LocalNode => Err(Peer::LocalNode),
}
}
}
pub enum PeerPotentialConnect<'a, TTrans: 'a, TInEvent: 'a, TOutEvent: 'a, THandler: 'a, THandlerErr: 'a, TPeerId: 'a>
where
TTrans: Transport
{
Connected(PeerConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>),
PendingConnect(PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>),
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
PeerPotentialConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport,
TPeerId: Eq + Hash + Clone,
{
#[inline]
pub fn close(self) {
match self {
PeerPotentialConnect::Connected(peer) => peer.close(),
PeerPotentialConnect::PendingConnect(peer) => peer.interrupt(),
}
}
#[inline]
pub fn into_connected(self) -> Option<PeerConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>> {
match self {
PeerPotentialConnect::Connected(peer) => Some(peer),
_ => None,
}
}
#[inline]
pub fn into_pending_connect(self) -> Option<PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>> {
match self {
PeerPotentialConnect::PendingConnect(peer) => Some(peer),
_ => None,
}
}
}
pub struct PeerConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where TTrans: Transport,
{
active_nodes: &'a mut CollectionStream<TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TPeerId>, THandlerErr, (), TPeerId>,
connected_points: &'a mut FnvHashMap<TPeerId, ConnectedPoint>,
out_reach_attempts: &'a mut FnvHashMap<TPeerId, OutReachAttempt>,
peer_id: TPeerId,
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId> PeerConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport,
TPeerId: Eq + Hash + Clone,
{
pub fn close(self) {
if let Some(reach_attempt) = self.out_reach_attempts.remove(&self.peer_id) {
self.active_nodes
.interrupt(reach_attempt.id)
.expect("Elements in out_reach_attempts are in sync with active_nodes; QED");
}
self.connected_points.remove(&self.peer_id);
self.active_nodes.peer_mut(&self.peer_id)
.expect("A PeerConnected is always created with a PeerId in active_nodes; QED")
.close();
}
#[inline]
pub fn endpoint(&self) -> &ConnectedPoint {
self.connected_points.get(&self.peer_id)
.expect("We insert into connected_points whenever a connection is opened and remove \
only when a connection is closed; the underlying API is guaranteed to always \
deliver a connection closed message after it has been opened, and no two \
closed messages; QED")
}
#[inline]
pub fn send_event(&mut self, event: TInEvent) {
self.active_nodes.peer_mut(&self.peer_id)
.expect("A PeerConnected is always created with a PeerId in active_nodes; QED")
.send_event(event)
}
}
#[derive(Debug)]
pub struct PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport
{
attempt: OccupiedEntry<'a, TPeerId, OutReachAttempt>,
active_nodes: &'a mut CollectionStream<TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TPeerId>, THandlerErr, (), TPeerId>,
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport,
TPeerId: Eq + Hash + Clone,
{
#[inline]
pub fn interrupt(self) {
let attempt = self.attempt.remove();
if self.active_nodes.interrupt(attempt.id).is_err() {
panic!("We retreived this attempt.id from out_reach_attempts. We insert in \
out_reach_attempts only at the same time as we call add_reach_attempt. \
Whenever we receive a NodeReached, NodeReplaced or ReachError event, which \
invalidate the attempt.id, we also remove the corresponding entry in \
out_reach_attempts.");
}
}
#[inline]
pub fn attempted_multiaddr(&self) -> &Multiaddr {
&self.attempt.get().cur_attempted
}
#[inline]
pub fn pending_multiaddrs(&self) -> impl Iterator<Item = &Multiaddr> {
self.attempt.get().next_attempts.iter()
}
pub fn append_multiaddr_attempts(&mut self, addrs: impl IntoIterator<Item = Multiaddr>) {
for addr in addrs {
self.append_multiaddr_attempt(addr);
}
}
pub fn append_multiaddr_attempt(&mut self, addr: Multiaddr) {
if self.attempt.get().next_attempts.iter().any(|a| a == &addr) {
return;
}
self.attempt.get_mut().next_attempts.push(addr);
}
}
pub struct PeerNotConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport,
{
peer_id: TPeerId,
nodes: &'a mut RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>,
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId> fmt::Debug for
PeerNotConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport,
TPeerId: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
f.debug_struct("PeerNotConnected")
.field("peer_id", &self.peer_id)
.finish()
}
}
impl<'a, TTrans, TInEvent, TOutEvent, TMuxer, THandler, THandlerErr, TPeerId>
PeerNotConnected<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TTrans: Transport<Output = (TPeerId, TMuxer)> + Clone,
TTrans::Error: Send + 'static,
TTrans::Dial: Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
THandler: IntoNodeHandler<TPeerId> + Send + 'static,
THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
<THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static,
THandlerErr: error::Error + Send + 'static,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
{
#[inline]
pub fn connect(self, addr: Multiaddr, handler: THandler)
-> PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TPeerId: fmt::Debug + Eq + Hash + Clone + AsRef<[u8]> + Send + 'static,
{
self.connect_inner(handler, addr, Vec::new())
}
#[inline]
pub fn connect_iter<TIter>(self, addrs: TIter, handler: THandler)
-> Result<PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>, Self>
where
TIter: IntoIterator<Item = Multiaddr>,
TPeerId: fmt::Debug + Eq + Hash + Clone + AsRef<[u8]> + Send + 'static,
{
let mut addrs = addrs.into_iter();
let first = match addrs.next() {
Some(f) => f,
None => return Err(self)
};
let rest = addrs.collect();
Ok(self.connect_inner(handler, first, rest))
}
fn connect_inner(self, handler: THandler, first: Multiaddr, rest: Vec<Multiaddr>)
-> PeerPendingConnect<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TPeerId>
where
TPeerId: fmt::Debug + Eq + Hash + Clone + AsRef<[u8]> + Send + 'static,
{
self.nodes.start_dial_out(self.peer_id.clone(), handler, first, rest);
PeerPendingConnect {
attempt: match self.nodes.reach_attempts.out_reach_attempts.entry(self.peer_id) {
Entry::Occupied(e) => e,
Entry::Vacant(_) => {
panic!("We called out_reach_attempts.insert with this peer id just above")
},
},
active_nodes: &mut self.nodes.active_nodes,
}
}
}