fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-12 17:41:22 +00:00
Code Issues Projects Releases Wiki Activity

Rename RawSwarm* to Network*. (#1194)

Browse Source 
* Rename RawSwarm* to Network*.

To complete the cut performed in [1].

The only remaining mention of a "swarm" in libp2p-core is in some tests
which actually depend on libp2p-swarm.

[1]: https://github.com/libp2p/rust-libp2p/pull/1188

* Post-merge corrections.
This commit is contained in:
Roman Borschel
2019-07-10 10:27:21 +02:00
committed by GitHub
parent ede114a13c
commit dfbf5b65c5
7 changed files with 229 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/nodes/handled_node.rs
View File

@ -32,7 +32,7 @@ mod tests;
pub trait NodeHandler {
/// Custom event that can be received from the outside.
type InEvent;
/// Custom event that can be produced by the handler and that will be returned by the swarm.
/// Custom event that can be produced by the handler and that will be returned by the `Network`.
type OutEvent;
/// Error that can happen during the processing of the node.
type Error;

6
core/src/nodes/mod.rs
View File

@ -23,7 +23,7 @@
//! Contains structs that are aiming at providing very precise control over what happens over the
//! network.
//!
//! The more complete and highest-level struct is the `RawSwarm`. The `RawSwarm` directly or
//! The more complete and highest-level struct is the `Network`. The `Network` directly or
//! indirectly uses all the other structs of this module.
pub mod collection;
@ -31,9 +31,9 @@ pub mod handled_node;
pub mod tasks;
pub mod listeners;
pub mod node;
pub mod raw_swarm;
pub mod network;
pub use collection::ConnectionInfo;
pub use node::Substream;
pub use handled_node::{NodeHandlerEvent, NodeHandlerEndpoint};
pub use raw_swarm::{Peer, RawSwarm, RawSwarmEvent};
pub use network::{Peer, Network, NetworkEvent};

102
core/src/nodes/raw_swarm.rs → core/src/nodes/network.rs
View File

@ -56,7 +56,7 @@ pub use crate::nodes::collection::StartTakeOver;
mod tests;
/// Implementation of `Stream` that handles the nodes.
pub struct RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo = PeerId, TPeerId = PeerId>
pub struct Network<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo = PeerId, TPeerId = PeerId>
where
TTrans: Transport,
{
@ -66,7 +66,7 @@ where
/// The nodes currently active.
active_nodes: CollectionStream<TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TConnInfo>, THandlerErr, (), (TConnInfo, ConnectedPoint), TPeerId>,
/// The reach attempts of the swarm.
/// The reach attempts of the network.
/// This needs to be a separate struct in order to handle multiple mutable borrows issues.
reach_attempts: ReachAttempts<TPeerId>,
@ -85,7 +85,7 @@ where
}
impl<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId> fmt::Debug for
RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>
Network<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>
where
TTrans: fmt::Debug + Transport,
TConnInfo: fmt::Debug,
@ -155,8 +155,8 @@ struct OutReachAttempt {
next_attempts: Vec<Multiaddr>,
}
/// Event that can happen on the `RawSwarm`.
pub enum RawSwarmEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo = PeerId, TPeerId = PeerId>
/// Event that can happen on the `Network`.
pub enum NetworkEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo = PeerId, TPeerId = PeerId>
where
TTrans: Transport,
{
@ -242,7 +242,7 @@ where
multiaddr: Multiaddr,
/// The error that happened.
error: RawSwarmReachError<TTrans::Error, TConnInfo>,
error: NetworkReachError<TTrans::Error, TConnInfo>,
},
/// Failed to reach a peer that we were trying to dial.
@ -267,7 +267,7 @@ where
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId> fmt::Debug for
RawSwarmEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>
NetworkEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>
where
TOutEvent: fmt::Debug,
TTrans: Transport,
@ -278,28 +278,28 @@ where
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
match *self {
RawSwarmEvent::NewListenerAddress { ref listen_addr } => {
NetworkEvent::NewListenerAddress { ref listen_addr } => {
f.debug_struct("NewListenerAddress")
.field("listen_addr", listen_addr)
.finish()
}
RawSwarmEvent::ExpiredListenerAddress { ref listen_addr } => {
NetworkEvent::ExpiredListenerAddress { ref listen_addr } => {
f.debug_struct("ExpiredListenerAddress")
.field("listen_addr", listen_addr)
.finish()
}
RawSwarmEvent::ListenerClosed { ref result, .. } => {
NetworkEvent::ListenerClosed { ref result, .. } => {
f.debug_struct("ListenerClosed")
.field("result", result)
.finish()
}
RawSwarmEvent::IncomingConnection(ref event) => {
NetworkEvent::IncomingConnection(ref event) => {
f.debug_struct("IncomingConnection")
.field("listen_addr", &event.listen_addr)
.field("send_back_addr", &event.send_back_addr)
.finish()
}
RawSwarmEvent::IncomingConnectionError {
NetworkEvent::IncomingConnectionError {
ref listen_addr,
ref send_back_addr,
ref error
@ -310,13 +310,13 @@ where
.field("error", error)
.finish()
}
RawSwarmEvent::Connected { ref conn_info, ref endpoint } => {
NetworkEvent::Connected { ref conn_info, ref endpoint } => {
f.debug_struct("Connected")
.field("conn_info", conn_info)
.field("endpoint", endpoint)
.finish()
}
RawSwarmEvent::Replaced { ref new_info, ref old_info, ref closed_endpoint, ref endpoint } => {
NetworkEvent::Replaced { ref new_info, ref old_info, ref closed_endpoint, ref endpoint } => {
f.debug_struct("Replaced")
.field("new_info", new_info)
.field("old_info", old_info)
@ -324,14 +324,14 @@ where
.field("endpoint", endpoint)
.finish()
}
RawSwarmEvent::NodeClosed { ref conn_info, ref endpoint, ref error } => {
NetworkEvent::NodeClosed { ref conn_info, ref endpoint, ref error } => {
f.debug_struct("NodeClosed")
.field("conn_info", conn_info)
.field("endpoint", endpoint)
.field("error", error)
.finish()
}
RawSwarmEvent::DialError { ref new_state, ref peer_id, ref multiaddr, ref error } => {
NetworkEvent::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)
@ -339,13 +339,13 @@ where
.field("error", error)
.finish()
}
RawSwarmEvent::UnknownPeerDialError { ref multiaddr, ref error, .. } => {
NetworkEvent::UnknownPeerDialError { ref multiaddr, ref error, .. } => {
f.debug_struct("UnknownPeerDialError")
.field("multiaddr", multiaddr)
.field("error", error)
.finish()
}
RawSwarmEvent::NodeEvent { ref conn_info, ref event } => {
NetworkEvent::NodeEvent { ref conn_info, ref event } => {
f.debug_struct("NodeEvent")
.field("conn_info", conn_info)
.field("event", event)
@ -418,7 +418,7 @@ pub enum PeerState {
/// Error that can happen when trying to reach a node.
#[derive(Debug)]
pub enum RawSwarmReachError<TTransErr, TConnInfo> {
pub enum NetworkReachError<TTransErr, TConnInfo> {
/// Error in the transport layer.
Transport(TransportError<TTransErr>),
@ -430,30 +430,30 @@ pub enum RawSwarmReachError<TTransErr, TConnInfo> {
}
}
impl<TTransErr, TConnInfo> fmt::Display for RawSwarmReachError<TTransErr, TConnInfo>
impl<TTransErr, TConnInfo> fmt::Display for NetworkReachError<TTransErr, TConnInfo>
where
TTransErr: fmt::Display,
TConnInfo: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
RawSwarmReachError::Transport(err) => write!(f, "{}", err),
RawSwarmReachError::PeerIdMismatch { obtained } => {
NetworkReachError::Transport(err) => write!(f, "{}", err),
NetworkReachError::PeerIdMismatch { obtained } => {
write!(f, "Peer ID mismatch, obtained: {:?}", obtained)
},
}
}
}
impl<TTransErr, TConnInfo> error::Error for RawSwarmReachError<TTransErr, TConnInfo>
impl<TTransErr, TConnInfo> error::Error for NetworkReachError<TTransErr, TConnInfo>
where
TTransErr: error::Error + 'static,
TConnInfo: fmt::Debug,
{
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
RawSwarmReachError::Transport(err) => Some(err),
RawSwarmReachError::PeerIdMismatch { .. } => None,
NetworkReachError::Transport(err) => Some(err),
NetworkReachError::PeerIdMismatch { .. } => None,
}
}
}
@ -544,9 +544,9 @@ where TTrans: Transport
listen_addr: Multiaddr,
/// Address used to send back data to the remote.
send_back_addr: Multiaddr,
/// Reference to the `active_nodes` field of the swarm.
/// Reference to the `active_nodes` field of the `Network`.
active_nodes: &'a mut CollectionStream<TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TConnInfo>, THandlerErr, (), (TConnInfo, ConnectedPoint), TPeerId>,
/// Reference to the `other_reach_attempts` field of the swarm.
/// Reference to the `other_reach_attempts` field of the `Network`.
other_reach_attempts: &'a mut Vec<(ReachAttemptId, ConnectedPoint)>,
}
@ -648,7 +648,7 @@ impl<'a> IncomingInfo<'a> {
}
impl<TTrans, TInEvent, TOutEvent, TMuxer, THandler, THandlerErr, TConnInfo, TPeerId>
RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>
Network<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>
where
TTrans: Transport + Clone,
TMuxer: StreamMuxer,
@ -662,7 +662,7 @@ where
/// Creates a new node events stream.
pub fn new(transport: TTrans, local_peer_id: TPeerId) -> Self {
// TODO: with_capacity?
RawSwarm {
Network {
listeners: ListenersStream::new(transport),
active_nodes: CollectionStream::new(),
reach_attempts: ReachAttempts {
@ -680,7 +680,7 @@ where
pub fn new_with_incoming_limit(transport: TTrans,
local_peer_id: TPeerId, incoming_limit: Option<u32>) -> Self
{
RawSwarm {
Network {
incoming_limit,
listeners: ListenersStream::new(transport),
active_nodes: CollectionStream::new(),
@ -942,7 +942,7 @@ where
}
/// Provides an API similar to `Stream`, except that it cannot error.
pub fn poll(&mut self) -> Async<RawSwarmEvent<'_, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>>
pub fn poll(&mut self) -> Async<NetworkEvent<'_, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>>
where
TTrans: Transport<Output = (TConnInfo, TMuxer)>,
TTrans::Error: Send + 'static,
@ -977,16 +977,16 @@ where
active_nodes: &mut self.active_nodes,
other_reach_attempts: &mut self.reach_attempts.other_reach_attempts,
};
return Async::Ready(RawSwarmEvent::IncomingConnection(event));
return Async::Ready(NetworkEvent::IncomingConnection(event));
}
Async::Ready(ListenersEvent::NewAddress { listen_addr }) => {
return Async::Ready(RawSwarmEvent::NewListenerAddress { listen_addr })
return Async::Ready(NetworkEvent::NewListenerAddress { listen_addr })
}
Async::Ready(ListenersEvent::AddressExpired { listen_addr }) => {
return Async::Ready(RawSwarmEvent::ExpiredListenerAddress { listen_addr })
return Async::Ready(NetworkEvent::ExpiredListenerAddress { listen_addr })
}
Async::Ready(ListenersEvent::Closed { listener, result }) => {
return Async::Ready(RawSwarmEvent::ListenerClosed { listener, result })
return Async::Ready(NetworkEvent::ListenerClosed { listener, result })
}
}
}
@ -1036,7 +1036,7 @@ where
closed message after it has been opened, and no two closed \
messages; QED");
action = Default::default();
out_event = RawSwarmEvent::NodeClosed {
out_event = NetworkEvent::NodeClosed {
conn_info: conn_info.0,
endpoint,
error,
@ -1044,7 +1044,7 @@ where
}
Async::Ready(CollectionEvent::NodeEvent { peer, event }) => {
action = Default::default();
out_event = RawSwarmEvent::NodeEvent { conn_info: peer.info().0.clone(), event };
out_event = NetworkEvent::NodeEvent { conn_info: peer.info().0.clone(), event };
}
}
@ -1077,7 +1077,7 @@ where
}
}
/// Internal struct indicating an action to perform of the swarm.
/// Internal struct indicating an action to perform on the network.
#[derive(Debug)]
#[must_use]
struct ActionItem<THandler, TPeerId> {
@ -1105,7 +1105,7 @@ impl<THandler, TPeerId> Default for ActionItem<THandler, TPeerId> {
fn handle_node_reached<'a, TTrans, TMuxer, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>(
reach_attempts: &mut ReachAttempts<TPeerId>,
event: CollectionReachEvent<'_, TInEvent, TOutEvent, THandler, InternalReachErr<TTrans::Error, TConnInfo>, THandlerErr, (), (TConnInfo, ConnectedPoint), TPeerId>,
) -> (ActionItem<THandler, TPeerId>, RawSwarmEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>)
) -> (ActionItem<THandler, TPeerId>, NetworkEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>)
where
TTrans: Transport<Output = (TConnInfo, TMuxer)> + Clone,
TMuxer: StreamMuxer + Send + Sync + 'static,
@ -1131,7 +1131,7 @@ where
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 {
return (Default::default(), NetworkEvent::IncomingConnectionError {
listen_addr,
send_back_addr,
error: IncomingError::DeniedLowerPriority,
@ -1172,14 +1172,14 @@ where
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 {
return (action, NetworkEvent::Replaced {
new_info: conn_info.0,
old_info: old_info.0,
endpoint: opened_endpoint,
closed_endpoint,
});
} else {
return (action, RawSwarmEvent::Connected {
return (action, NetworkEvent::Connected {
conn_info: conn_info.0,
endpoint: opened_endpoint
});
@ -1214,7 +1214,7 @@ where
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 {
return (Default::default(), NetworkEvent::Replaced {
new_info: conn_info.0,
old_info: old_info.0,
endpoint: opened_endpoint,
@ -1222,7 +1222,7 @@ where
});
} else {
return (Default::default(), RawSwarmEvent::Connected {
return (Default::default(), NetworkEvent::Connected {
conn_info: conn_info.0,
endpoint: opened_endpoint
});
@ -1260,7 +1260,7 @@ fn handle_reach_error<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TC
reach_id: ReachAttemptId,
error: InternalReachErr<TTrans::Error, TConnInfo>,
handler: THandler,
) -> (ActionItem<THandler, TPeerId>, RawSwarmEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>)
) -> (ActionItem<THandler, TPeerId>, NetworkEvent<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>)
where
TTrans: Transport,
TConnInfo: ConnectionInfo<PeerId = TPeerId> + Send + 'static,
@ -1303,9 +1303,9 @@ where
};
let error = match error {
InternalReachErr::Transport(err) => RawSwarmReachError::Transport(err),
InternalReachErr::Transport(err) => NetworkReachError::Transport(err),
InternalReachErr::PeerIdMismatch { obtained } => {
RawSwarmReachError::PeerIdMismatch { obtained }
NetworkReachError::PeerIdMismatch { obtained }
},
InternalReachErr::FoundLocalPeerId => {
unreachable!("We only generate FoundLocalPeerId within dial() or accept(); neither \
@ -1313,7 +1313,7 @@ where
},
};
return (action, RawSwarmEvent::DialError {
return (action, NetworkEvent::DialError {
new_state,
peer_id,
multiaddr: failed_addr,
@ -1338,7 +1338,7 @@ where
which doesn't add any entry in other_reach_attempts; QED")
},
};
return (Default::default(), RawSwarmEvent::UnknownPeerDialError {
return (Default::default(), NetworkEvent::UnknownPeerDialError {
multiaddr: address,
error,
handler,
@ -1353,7 +1353,7 @@ where
which doesn't add any entry in other_reach_attempts; QED")
},
};
return (Default::default(), RawSwarmEvent::IncomingConnectionError {
return (Default::default(), NetworkEvent::IncomingConnectionError {
listen_addr,
send_back_addr,
error
@ -1693,7 +1693,7 @@ where
TTrans: Transport,
{
peer_id: TPeerId,
nodes: &'a mut RawSwarm<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>,
nodes: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId>,
}
impl<'a, TTrans, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo, TPeerId> fmt::Debug for

206
core/src/nodes/raw_swarm/tests.rs → core/src/nodes/network/tests.rs
View File

@ -36,45 +36,45 @@ use tokio::runtime::{Builder, Runtime};
fn query_transport() {
let transport = DummyTransport::new();
let transport2 = transport.clone();
let raw_swarm = RawSwarm::<_, _, _, Handler, _>::new(transport, PeerId::random());
assert_eq!(raw_swarm.transport(), &transport2);
let network = Network::<_, _, _, Handler, _>::new(transport, PeerId::random());
assert_eq!(network.transport(), &transport2);
}
#[test]
fn local_node_peer() {
let peer_id = PeerId::random();
let mut raw_swarm = RawSwarm::<_, _, _, Handler, _>::new(DummyTransport::new(), peer_id.clone());
assert_matches!(raw_swarm.peer(peer_id), Peer::LocalNode);
let mut network = Network::<_, _, _, Handler, _>::new(DummyTransport::new(), peer_id.clone());
assert_matches!(network.peer(peer_id), Peer::LocalNode);
}
#[test]
fn successful_dial_reaches_a_node() {
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let mut network = Network::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let addr = "/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr");
let dial_res = swarm.dial(addr, Handler::default());
let dial_res = network.dial(addr, Handler::default());
assert!(dial_res.is_ok());
// Poll the swarm until we get a `NodeReached` then assert on the peer:
// Poll the network until we get a `NodeReached` then assert on the peer:
// it's there and it's connected.
let swarm = Arc::new(Mutex::new(swarm));
let network = Arc::new(Mutex::new(network));
let mut rt = Runtime::new().unwrap();
let mut peer_id : Option<PeerId> = None;
// Drive forward until we're Connected
while peer_id.is_none() {
let swarm_fut = swarm.clone();
let network_fut = network.clone();
peer_id = rt.block_on(future::poll_fn(move || -> Poll<Option<PeerId>, ()> {
let mut swarm = swarm_fut.lock();
let poll_res = swarm.poll();
let mut network = network_fut.lock();
let poll_res = network.poll();
match poll_res {
Async::Ready(RawSwarmEvent::Connected { conn_info, .. }) => Ok(Async::Ready(Some(conn_info))),
Async::Ready(NetworkEvent::Connected { conn_info, .. }) => Ok(Async::Ready(Some(conn_info))),
_ => Ok(Async::Ready(None))
}
})).expect("tokio works");
}
let mut swarm = swarm.lock();
let peer = swarm.peer(peer_id.unwrap());
let mut network = network.lock();
let peer = network.peer(peer_id.unwrap());
assert_matches!(peer, Peer::Connected(PeerConnected{..}));
}
@ -94,46 +94,46 @@ fn num_incoming_negotiated() {
];
transport.set_initial_listener_state(ListenerState::Events(events));
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(transport, PeerId::random());
swarm.listen_on("/memory/0".parse().unwrap()).unwrap();
let mut network = Network::<_, _, _, Handler, _>::new(transport, PeerId::random());
network.listen_on("/memory/0".parse().unwrap()).unwrap();
// no incoming yet
assert_eq!(swarm.incoming_negotiated().count(), 0);
assert_eq!(network.incoming_negotiated().count(), 0);
let mut rt = Runtime::new().unwrap();
let swarm = Arc::new(Mutex::new(swarm));
let swarm_fut = swarm.clone();
let network = Arc::new(Mutex::new(network));
let network_fut = network.clone();
let fut = future::poll_fn(move || -> Poll<_, ()> {
let mut swarm_fut = swarm_fut.lock();
assert_matches!(swarm_fut.poll(), Async::Ready(RawSwarmEvent::NewListenerAddress {..}));
assert_matches!(swarm_fut.poll(), Async::Ready(RawSwarmEvent::IncomingConnection(incoming)) => {
let mut network_fut = network_fut.lock();
assert_matches!(network_fut.poll(), Async::Ready(NetworkEvent::NewListenerAddress {..}));
assert_matches!(network_fut.poll(), Async::Ready(NetworkEvent::IncomingConnection(incoming)) => {
incoming.accept(Handler::default());
});
Ok(Async::Ready(()))
});
rt.block_on(fut).expect("tokio works");
let swarm = swarm.lock();
let network = network.lock();
// Now there's an incoming connection
assert_eq!(swarm.incoming_negotiated().count(), 1);
assert_eq!(network.incoming_negotiated().count(), 1);
}
#[test]
fn broadcasted_events_reach_active_nodes() {
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let mut network = Network::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let mut muxer = DummyMuxer::new();
muxer.set_inbound_connection_state(DummyConnectionState::Pending);
muxer.set_outbound_connection_state(DummyConnectionState::Opened);
let addr = "/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr");
let mut handler = Handler::default();
handler.next_states = vec![HandlerState::Ready(NodeHandlerEvent::Custom(OutEvent::Custom("from handler 1") )),];
let dial_result = swarm.dial(addr, handler);
let dial_result = network.dial(addr, handler);
assert!(dial_result.is_ok());
let swarm = Arc::new(Mutex::new(swarm));
let network = Arc::new(Mutex::new(network));
let mut rt = Runtime::new().unwrap();
let swarm2 = swarm.clone();
let network2 = network.clone();
rt.block_on(future::poll_fn(move || {
if swarm2.lock().start_broadcast(&InEvent::NextState).is_not_ready() {
if network2.lock().start_broadcast(&InEvent::NextState).is_not_ready() {
Ok::<_, ()>(Async::NotReady)
} else {
Ok(Async::Ready(()))
@ -141,15 +141,15 @@ fn broadcasted_events_reach_active_nodes() {
})).unwrap();
let mut peer_id : Option<PeerId> = None;
while peer_id.is_none() {
let swarm_fut = swarm.clone();
let network_fut = network.clone();
peer_id = rt.block_on(future::poll_fn(move || -> Poll<Option<PeerId>, ()> {
let mut swarm = swarm_fut.lock();
if swarm.complete_broadcast().is_not_ready() {
let mut network = network_fut.lock();
if network.complete_broadcast().is_not_ready() {
return Ok(Async::NotReady)
}
let poll_res = swarm.poll();
let poll_res = network.poll();
match poll_res {
Async::Ready(RawSwarmEvent::Connected { conn_info, .. }) => Ok(Async::Ready(Some(conn_info))),
Async::Ready(NetworkEvent::Connected { conn_info, .. }) => Ok(Async::Ready(Some(conn_info))),
_ => Ok(Async::Ready(None))
}
})).expect("tokio works");
@ -157,12 +157,12 @@ fn broadcasted_events_reach_active_nodes() {
let mut keep_polling = true;
while keep_polling {
let swarm_fut = swarm.clone();
let network_fut = network.clone();
keep_polling = rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
let mut swarm = swarm_fut.lock();
match swarm.poll() {
let mut network = network_fut.lock();
match network.poll() {
Async::Ready(event) => {
assert_matches!(event, RawSwarmEvent::NodeEvent { conn_info: _, event: inner_event } => {
assert_matches!(event, NetworkEvent::NodeEvent { conn_info: _, event: inner_event } => {
// The event we sent reached the node and triggered sending the out event we told it to return
assert_matches!(inner_event, OutEvent::Custom("from handler 1"));
});
@ -176,9 +176,9 @@ fn broadcasted_events_reach_active_nodes() {
#[test]
fn querying_for_pending_peer() {
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let mut network = Network::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let peer_id = PeerId::random();
let peer = swarm.peer(peer_id.clone());
let peer = network.peer(peer_id.clone());
assert_matches!(peer, Peer::NotConnected(PeerNotConnected{ .. }));
let addr = "/memory/0".parse().expect("bad multiaddr");
let pending_peer = peer.into_not_connected().unwrap().connect(addr, Handler::default());
@ -187,9 +187,9 @@ fn querying_for_pending_peer() {
#[test]
fn querying_for_unknown_peer() {
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let mut network = Network::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let peer_id = PeerId::random();
let peer = swarm.peer(peer_id.clone());
let peer = network.peer(peer_id.clone());
assert_matches!(peer, Peer::NotConnected( PeerNotConnected { nodes: _, peer_id: node_peer_id }) => {
assert_eq!(node_peer_id, peer_id);
});
@ -197,31 +197,31 @@ fn querying_for_unknown_peer() {
#[test]
fn querying_for_connected_peer() {
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
let mut network = Network::<_, _, _, Handler, _>::new(DummyTransport::new(), PeerId::random());
// Dial a node
let addr = "/ip4/127.0.0.1/tcp/1234".parse().expect("bad multiaddr");
swarm.dial(addr, Handler::default()).expect("dialing works");
network.dial(addr, Handler::default()).expect("dialing works");
let swarm = Arc::new(Mutex::new(swarm));
let network = Arc::new(Mutex::new(network));
let mut rt = Runtime::new().unwrap();
// Drive it forward until we connect; extract the new PeerId.
let mut peer_id : Option<PeerId> = None;
while peer_id.is_none() {
let swarm_fut = swarm.clone();
let network_fut = network.clone();
peer_id = rt.block_on(future::poll_fn(move || -> Poll<Option<PeerId>, ()> {
let mut swarm = swarm_fut.lock();
let poll_res = swarm.poll();
let mut network = network_fut.lock();
let poll_res = network.poll();
match poll_res {
Async::Ready(RawSwarmEvent::Connected { conn_info, .. }) => Ok(Async::Ready(Some(conn_info))),
Async::Ready(NetworkEvent::Connected { conn_info, .. }) => Ok(Async::Ready(Some(conn_info))),
_ => Ok(Async::Ready(None))
}
})).expect("tokio works");
}
// We're connected.
let mut swarm = swarm.lock();
let peer = swarm.peer(peer_id.unwrap());
let mut network = network.lock();
let peer = network.peer(peer_id.unwrap());
assert_matches!(peer, Peer::Connected( PeerConnected { .. } ));
}
@ -231,16 +231,16 @@ fn poll_with_closed_listener() {
// Set up listener to be closed
transport.set_initial_listener_state(ListenerState::Ok(Async::Ready(None)));
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(transport, PeerId::random());
swarm.listen_on("/memory/0".parse().unwrap()).unwrap();
let mut network = Network::<_, _, _, Handler, _>::new(transport, PeerId::random());
network.listen_on("/memory/0".parse().unwrap()).unwrap();
let mut rt = Runtime::new().unwrap();
let swarm = Arc::new(Mutex::new(swarm));
let network = Arc::new(Mutex::new(network));
let swarm_fut = swarm.clone();
let network_fut = network.clone();
let fut = future::poll_fn(move || -> Poll<_, ()> {
let mut swarm = swarm_fut.lock();
assert_matches!(swarm.poll(), Async::Ready(RawSwarmEvent::ListenerClosed { .. } ));
let mut network = network_fut.lock();
assert_matches!(network.poll(), Async::Ready(NetworkEvent::ListenerClosed { .. } ));
Ok(Async::Ready(()))
});
rt.block_on(fut).expect("tokio works");
@ -250,24 +250,24 @@ fn poll_with_closed_listener() {
fn unknown_peer_that_is_unreachable_yields_unknown_peer_dial_error() {
let mut transport = DummyTransport::new();
transport.make_dial_fail();
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new(transport, PeerId::random());
let mut network = Network::<_, _, _, Handler, _>::new(transport, PeerId::random());
let addr = "/memory/0".parse::<Multiaddr>().expect("bad multiaddr");
let handler = Handler::default();
let dial_result = swarm.dial(addr, handler);
let dial_result = network.dial(addr, handler);
assert!(dial_result.is_ok());
let swarm = Arc::new(Mutex::new(swarm));
let network = Arc::new(Mutex::new(network));
let mut rt = Runtime::new().unwrap();
// Drive it forward until we hear back from the node.
let mut keep_polling = true;
while keep_polling {
let swarm_fut = swarm.clone();
let network_fut = network.clone();
keep_polling = rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
let mut swarm = swarm_fut.lock();
match swarm.poll() {
let mut network = network_fut.lock();
match network.poll() {
Async::NotReady => Ok(Async::Ready(true)),
Async::Ready(event) => {
assert_matches!(event, RawSwarmEvent::UnknownPeerDialError { .. } );
assert_matches!(event, NetworkEvent::UnknownPeerDialError { .. } );
Ok(Async::Ready(false))
},
}
@ -281,12 +281,12 @@ fn known_peer_that_is_unreachable_yields_dial_error() {
let peer_id = PeerId::random();
transport.set_next_peer_id(&peer_id);
transport.make_dial_fail();
let swarm = Arc::new(Mutex::new(RawSwarm::<_, _, _, Handler, _>::new(transport, PeerId::random())));
let network = Arc::new(Mutex::new(Network::<_, _, _, Handler, _>::new(transport, PeerId::random())));
{
let swarm1 = swarm.clone();
let mut swarm1 = swarm1.lock();
let peer = swarm1.peer(peer_id.clone());
let network1 = network.clone();
let mut network1 = network1.lock();
let peer = network1.peer(peer_id.clone());
assert_matches!(peer, Peer::NotConnected(PeerNotConnected{ .. }));
let addr = "/memory/0".parse::<Multiaddr>().expect("bad multiaddr");
let pending_peer = peer.into_not_connected().unwrap().connect(addr, Handler::default());
@ -296,16 +296,16 @@ fn known_peer_that_is_unreachable_yields_dial_error() {
// Drive it forward until we hear back from the node.
let mut keep_polling = true;
while keep_polling {
let swarm_fut = swarm.clone();
let network_fut = network.clone();
let peer_id = peer_id.clone();
keep_polling = rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
let mut swarm = swarm_fut.lock();
match swarm.poll() {
let mut network = network_fut.lock();
match network.poll() {
Async::NotReady => Ok(Async::Ready(true)),
Async::Ready(event) => {
let failed_peer_id = assert_matches!(
event,
RawSwarmEvent::DialError { new_state: _, peer_id: failed_peer_id, .. } => failed_peer_id
NetworkEvent::DialError { new_state: _, peer_id: failed_peer_id, .. } => failed_peer_id
);
assert_eq!(peer_id, failed_peer_id);
Ok(Async::Ready(false))
@ -320,13 +320,13 @@ fn yields_node_error_when_there_is_an_error_after_successful_connect() {
let mut transport = DummyTransport::new();
let peer_id = PeerId::random();
transport.set_next_peer_id(&peer_id);
let swarm = Arc::new(Mutex::new(RawSwarm::<_, _, _, Handler, _>::new(transport, PeerId::random())));
let network = Arc::new(Mutex::new(Network::<_, _, _, Handler, _>::new(transport, PeerId::random())));
{
// Set up an outgoing connection with a PeerId we know
let swarm1 = swarm.clone();
let mut swarm1 = swarm1.lock();
let peer = swarm1.peer(peer_id.clone());
let network1 = network.clone();
let mut network1 = network1.lock();
let peer = network1.peer(peer_id.clone());
let addr = "/unix/reachable".parse().expect("bad multiaddr");
let mut handler = Handler::default();
// Force an error
@ -340,25 +340,25 @@ fn yields_node_error_when_there_is_an_error_after_successful_connect() {
// Drive it forward until we connect to the node.
let mut keep_polling = true;
while keep_polling {
let swarm_fut = swarm.clone();
let swarm2 = swarm.clone();
let network_fut = network.clone();
let network2 = network.clone();
rt.block_on(future::poll_fn(move || {
if swarm2.lock().start_broadcast(&InEvent::NextState).is_not_ready() {
if network2.lock().start_broadcast(&InEvent::NextState).is_not_ready() {
Ok::<_, ()>(Async::NotReady)
} else {
Ok(Async::Ready(()))
}
})).unwrap();
keep_polling = rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
let mut swarm = swarm_fut.lock();
let mut network = network_fut.lock();
// Push the Handler into an error state on the next poll
if swarm.complete_broadcast().is_not_ready() {
if network.complete_broadcast().is_not_ready() {
return Ok(Async::NotReady)
}
match swarm.poll() {
match network.poll() {
Async::NotReady => Ok(Async::Ready(true)),
Async::Ready(event) => {
assert_matches!(event, RawSwarmEvent::Connected { .. });
assert_matches!(event, NetworkEvent::Connected { .. });
// We're connected, we can move on
Ok(Async::Ready(false))
},
@ -368,11 +368,11 @@ fn yields_node_error_when_there_is_an_error_after_successful_connect() {
// Poll again. It is going to be a NodeClosed because of how the
// handler's next state was set up.
let swarm_fut = swarm.clone();
let network_fut = network.clone();
let expected_peer_id = peer_id.clone();
rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
let mut swarm = swarm_fut.lock();
assert_matches!(swarm.poll(), Async::Ready(RawSwarmEvent::NodeClosed { conn_info, .. }) => {
let mut network = network_fut.lock();
assert_matches!(network.poll(), Async::Ready(NetworkEvent::NodeClosed { conn_info, .. }) => {
assert_eq!(conn_info, expected_peer_id);
});
Ok(Async::Ready(()))
@ -405,32 +405,32 @@ fn limit_incoming_connections() {
).take(10));
transport.set_initial_listener_state(ListenerState::Events(events));
let mut swarm = RawSwarm::<_, _, _, Handler, _>::new_with_incoming_limit(transport, PeerId::random(), Some(limit));
assert_eq!(swarm.incoming_limit(), Some(limit));
swarm.listen_on("/memory/0".parse().unwrap()).unwrap();
assert_eq!(swarm.incoming_negotiated().count(), 0);
let mut network = Network::<_, _, _, Handler, _>::new_with_incoming_limit(transport, PeerId::random(), Some(limit));
assert_eq!(network.incoming_limit(), Some(limit));
network.listen_on("/memory/0".parse().unwrap()).unwrap();
assert_eq!(network.incoming_negotiated().count(), 0);
let swarm = Arc::new(Mutex::new(swarm));
let network = Arc::new(Mutex::new(network));
let mut rt = Runtime::new().unwrap();
for i in 1..10 {
let swarm_fut = swarm.clone();
let network_fut = network.clone();
let fut = future::poll_fn(move || -> Poll<_, ()> {
let mut swarm_fut = swarm_fut.lock();
let mut network_fut = network_fut.lock();
if i <= limit {
assert_matches!(swarm_fut.poll(), Async::Ready(RawSwarmEvent::NewListenerAddress {..}));
assert_matches!(swarm_fut.poll(),
Async::Ready(RawSwarmEvent::IncomingConnection(incoming)) => {
assert_matches!(network_fut.poll(), Async::Ready(NetworkEvent::NewListenerAddress {..}));
assert_matches!(network_fut.poll(),
Async::Ready(NetworkEvent::IncomingConnection(incoming)) => {
incoming.accept(Handler::default());
});
} else {
match swarm_fut.poll() {
match network_fut.poll() {
Async::NotReady => (),
Async::Ready(x) => {
match x {
RawSwarmEvent::NewListenerAddress {..} => {}
RawSwarmEvent::ExpiredListenerAddress {..} => {}
RawSwarmEvent::IncomingConnection(_) => {}
RawSwarmEvent::Connected {..} => {}
NetworkEvent::NewListenerAddress {..} => {}
NetworkEvent::ExpiredListenerAddress {..} => {}
NetworkEvent::IncomingConnection(_) => {}
NetworkEvent::Connected {..} => {}
e => panic!("Not expected event: {:?}", e)
}
},
@ -439,7 +439,7 @@ fn limit_incoming_connections() {
Ok(Async::Ready(()))
});
rt.block_on(fut).expect("tokio works");
let swarm = swarm.lock();
assert!(swarm.incoming_negotiated().count() <= (limit as usize));
let network = network.lock();
assert!(network.incoming_negotiated().count() <= (limit as usize));
}
}