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

swarm/: Enable advanced dialing requests (#2317)

Browse Source 
Enable advanced dialing requests both on `Swarm` and via
`NetworkBehaviourAction`. Users can now trigger a dial with a specific
set of addresses, optionally extended via
`NetworkBehaviour::addresses_of_peer`. In addition the whole process is
now modelled in a type safe way via the builder pattern.

Example of a `NetworkBehaviour` requesting a dial to a specific peer
with a set of addresses additionally extended through
`NetworkBehaviour::addresses_of_peer`:

```rust
NetworkBehaviourAction::Dial {
    opts: DialOpts::peer_id(peer_id)
              .condition(PeerCondition::Always)
              .addresses(addresses)
              .extend_addresses_through_behaviour()
              .build(),
    handler,
}
```

Example of a user requesting a dial to an unknown peer with a single
address via `Swarm`:

```rust
swarm1.dial(
    DialOpts::unknown_peer_id()
        .address(addr2.clone())
        .build()
)
```
This commit is contained in:
Max Inden
2021-11-15 14:17:23 +01:00
committed by GitHub
parent 144dc12fb6
commit 220f84a97f
53 changed files with 680 additions and 414 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
Cargo.toml
View File

@ -72,21 +72,21 @@ getrandom = "0.2.3" # Explicit dependency to be used in `wasm-bindgen` feature
instant = "0.1.11" # Explicit dependency to be used in `wasm-bindgen` feature
lazy_static = "1.2"
libp2p-core = { version = "0.30.0", path = "core", default-features = false }
libp2p-floodsub = { version = "0.31.0", path = "protocols/floodsub", optional = true }
libp2p-gossipsub = { version = "0.33.0", path = "./protocols/gossipsub", optional = true }
libp2p-identify = { version = "0.31.0", path = "protocols/identify", optional = true }
libp2p-floodsub = { version = "0.32.0", path = "protocols/floodsub", optional = true }
libp2p-gossipsub = { version = "0.34.0", path = "./protocols/gossipsub", optional = true }
libp2p-identify = { version = "0.32.0", path = "protocols/identify", optional = true }
libp2p-kad = { version = "0.33.0", path = "protocols/kad", optional = true }
libp2p-metrics = { version = "0.1.0", path = "misc/metrics", optional = true }
libp2p-metrics = { version = "0.2.0", path = "misc/metrics", optional = true }
libp2p-mplex = { version = "0.30.0", path = "muxers/mplex", optional = true }
libp2p-noise = { version = "0.33.0", path = "transports/noise", optional = true }
libp2p-ping = { version = "0.31.0", path = "protocols/ping", optional = true }
libp2p-ping = { version = "0.32.0", path = "protocols/ping", optional = true }
libp2p-plaintext = { version = "0.30.0", path = "transports/plaintext", optional = true }
libp2p-pnet = { version = "0.22.0", path = "transports/pnet", optional = true }
libp2p-relay = { version = "0.4.0", path = "protocols/relay", optional = true }
libp2p-rendezvous = { version = "0.1.0", path = "protocols/rendezvous", optional = true }
libp2p-request-response = { version = "0.13.0", path = "protocols/request-response", optional = true }
libp2p-swarm = { version = "0.31.0", path = "swarm" }
libp2p-swarm-derive = { version = "0.25.0", path = "swarm-derive" }
libp2p-relay = { version = "0.5.0", path = "protocols/relay", optional = true }
libp2p-rendezvous = { version = "0.2.0", path = "protocols/rendezvous", optional = true }
libp2p-request-response = { version = "0.14.0", path = "protocols/request-response", optional = true }
libp2p-swarm = { version = "0.32.0", path = "swarm" }
libp2p-swarm-derive = { version = "0.26.0", path = "swarm-derive" }
libp2p-uds = { version = "0.30.0", path = "transports/uds", optional = true }
libp2p-wasm-ext = { version = "0.30.0", path = "transports/wasm-ext", default-features = false, optional = true }
libp2p-yamux = { version = "0.34.0", path = "muxers/yamux", optional = true }
@ -99,7 +99,7 @@ smallvec = "1.6.1"
[target.'cfg(not(any(target_os = "emscripten", target_os = "wasi", target_os = "unknown")))'.dependencies]
libp2p-deflate = { version = "0.30.0", path = "transports/deflate", optional = true }
libp2p-dns = { version = "0.30.0", path = "transports/dns", optional = true, default-features = false }
libp2p-mdns = { version = "0.32.0", path = "protocols/mdns", optional = true }
libp2p-mdns = { version = "0.33.0", path = "protocols/mdns", optional = true }
libp2p-tcp = { version = "0.30.0", path = "transports/tcp", default-features = false, optional = true }
libp2p-websocket = { version = "0.32.0", path = "transports/websocket", optional = true }

4
examples/chat-tokio.rs
View File

@ -44,7 +44,7 @@ use libp2p::{
mdns::{Mdns, MdnsEvent},
mplex,
noise,
swarm::{NetworkBehaviourEventProcess, SwarmBuilder, SwarmEvent},
swarm::{dial_opts::DialOpts, NetworkBehaviourEventProcess, SwarmBuilder, SwarmEvent},
// `TokioTcpConfig` is available through the `tcp-tokio` feature.
tcp::TokioTcpConfig,
Multiaddr,
@ -148,7 +148,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
// Reach out to another node if specified
if let Some(to_dial) = std::env::args().nth(1) {
let addr: Multiaddr = to_dial.parse()?;
swarm.dial_addr(addr)?;
swarm.dial(addr)?;
println!("Dialed {:?}", to_dial)
}

2
examples/chat.rs
View File

@ -128,7 +128,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
// Reach out to another node if specified
if let Some(to_dial) = std::env::args().nth(1) {
let addr: Multiaddr = to_dial.parse()?;
swarm.dial_addr(addr)?;
swarm.dial(addr)?;
println!("Dialed {:?}", to_dial)
}

2
examples/file-sharing.rs
View File

@ -525,7 +525,7 @@ mod network {
.add_address(&peer_id, peer_addr.clone());
match self
.swarm
.dial_addr(peer_addr.with(Protocol::P2p(peer_id.into())))
.dial(peer_addr.with(Protocol::P2p(peer_id.into())))
{
Ok(()) => {
self.pending_dial.insert(peer_id, sender);

15
examples/gossipsub-chat.rs
View File

@ -53,7 +53,7 @@ use libp2p::gossipsub::MessageId;
use libp2p::gossipsub::{
GossipsubEvent, GossipsubMessage, IdentTopic as Topic, MessageAuthenticity, ValidationMode,
};
use libp2p::{gossipsub, identity, swarm::SwarmEvent, PeerId};
use libp2p::{gossipsub, identity, swarm::SwarmEvent, Multiaddr, PeerId};
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::time::Duration;
@ -122,14 +122,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
// Reach out to another node if specified
if let Some(to_dial) = std::env::args().nth(1) {
let dialing = to_dial.clone();
match to_dial.parse() {
Ok(to_dial) => match swarm.dial_addr(to_dial) {
Ok(_) => println!("Dialed {:?}", dialing),
Err(e) => println!("Dial {:?} failed: {:?}", dialing, e),
},
Err(err) => println!("Failed to parse address to dial: {:?}", err),
}
let address: Multiaddr = to_dial.parse().expect("User to provide valid address.");
match swarm.dial(address.clone()) {
Ok(_) => println!("Dialed {:?}", address),
Err(e) => println!("Dial {:?} failed: {:?}", address, e),
};
}
// Read full lines from stdin

2
examples/ipfs-private.rs
View File

@ -265,7 +265,7 @@ fn main() -> Result<(), Box<dyn Error>> {
// Reach out to other nodes if specified
for to_dial in std::env::args().skip(1) {
let addr: Multiaddr = parse_legacy_multiaddr(&to_dial)?;
swarm.dial_addr(addr)?;
swarm.dial(addr)?;
println!("Dialed {:?}", to_dial)
}

6
examples/ping.rs
View File

@ -43,7 +43,7 @@
use futures::executor::block_on;
use futures::prelude::*;
use libp2p::swarm::{Swarm, SwarmEvent};
use libp2p::{identity, ping, PeerId};
use libp2p::{identity, ping, Multiaddr, PeerId};
use std::error::Error;
use std::task::Poll;
@ -70,8 +70,8 @@ fn main() -> Result<(), Box<dyn Error>> {
// Dial the peer identified by the multi-address given as the second
// command-line argument, if any.
if let Some(addr) = std::env::args().nth(1) {
let remote = addr.parse()?;
swarm.dial_addr(remote)?;
let remote: Multiaddr = addr.parse()?;
swarm.dial(remote)?;
println!("Dialed {}", addr)
}

4
misc/metrics/CHANGELOG.md
View File

@ -1,3 +1,7 @@
## Version 0.2.0 [unreleased]
- Update dependencies.
## Version 0.1.0 [2021-11-01]
- Add initial version.

10
misc/metrics/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-metrics"
edition = "2018"
description = "Metrics for libp2p"
version = "0.1.0"
version = "0.2.0"
authors = ["Max Inden <mail@max-inden.de>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -15,11 +15,11 @@ kad = ["libp2p-kad"]
ping = ["libp2p-ping"]
[dependencies]
libp2p-core= { version = "0.30.0", path = "../../core" }
libp2p-identify = { version = "0.31.0", path = "../../protocols/identify", optional = true }
libp2p-core = { version = "0.30.0", path = "../../core" }
libp2p-identify = { version = "0.32.0", path = "../../protocols/identify", optional = true }
libp2p-kad = { version = "0.33.0", path = "../../protocols/kad", optional = true }
libp2p-ping = { version = "0.31.0", path = "../../protocols/ping", optional = true }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-ping = { version = "0.32.0", path = "../../protocols/ping", optional = true }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
open-metrics-client = "0.12.0"
[dev-dependencies]

5
misc/metrics/examples/metrics.rs
View File

@ -50,6 +50,7 @@
use futures::executor::block_on;
use futures::stream::StreamExt;
use libp2p::core::Multiaddr;
use libp2p::metrics::{Metrics, Recorder};
use libp2p::ping::{Ping, PingConfig};
use libp2p::swarm::SwarmEvent;
@ -74,8 +75,8 @@ fn main() -> Result<(), Box<dyn Error>> {
);
swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
if let Some(addr) = std::env::args().nth(1) {
let remote = addr.parse()?;
swarm.dial_addr(remote)?;
let remote: Multiaddr = addr.parse()?;
swarm.dial(remote)?;
tide::log::info!("Dialed {}", addr)
}

4
protocols/floodsub/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.32.0 [unreleased]
- Update dependencies.
# 0.31.0 [2021-11-01]
- Make default features of `libp2p-core` optional.

4
protocols/floodsub/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-floodsub"
edition = "2018"
description = "Floodsub protocol for libp2p"
version = "0.31.0"
version = "0.32.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -14,7 +14,7 @@ cuckoofilter = "0.5.0"
fnv = "1.0"
futures = "0.3.1"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
log = "0.4"
prost = "0.9"
rand = "0.7"

18
protocols/floodsub/src/layer.rs
View File

@ -28,8 +28,8 @@ use cuckoofilter::{CuckooError, CuckooFilter};
use fnv::FnvHashSet;
use libp2p_core::{connection::ConnectionId, PeerId};
use libp2p_swarm::{
DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, OneShotHandler,
PollParameters,
dial_opts::{self, DialOpts},
NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, OneShotHandler, PollParameters,
};
use log::warn;
use smallvec::SmallVec;
@ -107,9 +107,10 @@ impl Floodsub {
if self.target_peers.insert(peer_id) {
let handler = self.new_handler();
self.events.push_back(NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
self.events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(peer_id)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
}
@ -310,9 +311,10 @@ impl NetworkBehaviour for Floodsub {
// try to reconnect.
if self.target_peers.contains(id) {
let handler = self.new_handler();
self.events.push_back(NetworkBehaviourAction::DialPeer {
peer_id: *id,
condition: DialPeerCondition::Disconnected,
self.events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(*id)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
}

4
protocols/gossipsub/CHANGELOG.md
View File

@ -1,9 +1,11 @@
# 0.33.1 [unreleased]
# 0.34.0 [unreleased]
- Fix bug in internal peer's topics tracking (see [PR 2325]).
- Use `instant` and `futures-timer` instead of `wasm-timer` (see [PR 2245]).
- Update dependencies.
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
[PR 2325]: https://github.com/libp2p/rust-libp2p/pull/2325

4
protocols/gossipsub/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-gossipsub"
edition = "2018"
description = "Gossipsub protocol for libp2p"
version = "0.33.1"
version = "0.34.0"
authors = ["Age Manning <Age@AgeManning.com>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -10,7 +10,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
categories = ["network-programming", "asynchronous"]
[dependencies]
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
bytes = "1.0"
byteorder = "1.3.4"

18
protocols/gossipsub/src/behaviour.rs
View File

@ -41,8 +41,8 @@ use libp2p_core::{
multiaddr::Protocol::Ip6, ConnectedPoint, Multiaddr, PeerId,
};
use libp2p_swarm::{
DialPeerCondition, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction,
NotifyHandler, PollParameters,
dial_opts::{self, DialOpts},
IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
};
use crate::config::{GossipsubConfig, ValidationMode};
@ -1046,9 +1046,10 @@ where
// Connect to peer
debug!("Connecting to explicit peer {:?}", peer_id);
let handler = self.new_handler();
self.events.push_back(NetworkBehaviourAction::DialPeer {
peer_id: *peer_id,
condition: DialPeerCondition::Disconnected,
self.events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(*peer_id)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
}
@ -1502,9 +1503,10 @@ where
// dial peer
let handler = self.new_handler();
self.events.push_back(NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
self.events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(peer_id)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
}

34
protocols/gossipsub/src/behaviour/tests.rs
View File

@ -1342,11 +1342,9 @@ mod tests {
.events
.iter()
.filter(|e| match e {
NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
handler: _,
} => peer_id == &peer,
NetworkBehaviourAction::Dial { opts, handler: _ } => {
opts.get_peer_id() == Some(peer)
}
_ => false,
})
.collect();
@ -1388,11 +1386,8 @@ mod tests {
gs.events
.iter()
.filter(|e| match e {
NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
handler: _,
} => peer_id == peer,
NetworkBehaviourAction::Dial { opts, handler: _ } =>
opts.get_peer_id() == Some(*peer),
_ => false,
})
.count(),
@ -1407,11 +1402,8 @@ mod tests {
gs.events
.iter()
.filter(|e| match e {
NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
handler: _,
} => peer_id == peer,
NetworkBehaviourAction::Dial { opts, handler: _ } =>
opts.get_peer_id() == Some(*peer),
_ => false,
})
.count()
@ -1821,11 +1813,7 @@ mod tests {
.events
.iter()
.filter_map(|e| match e {
NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
handler: _,
} => Some(peer_id.clone()),
NetworkBehaviourAction::Dial { opts, handler: _ } => opts.get_peer_id(),
_ => None,
})
.collect();
@ -2444,7 +2432,7 @@ mod tests {
gs.events
.iter()
.filter(|e| match e {
NetworkBehaviourAction::DialPeer { .. } => true,
NetworkBehaviourAction::Dial { .. } => true,
_ => false,
})
.count(),
@ -3047,7 +3035,7 @@ mod tests {
gs.events
.iter()
.filter(|e| match e {
NetworkBehaviourAction::DialPeer { .. } => true,
NetworkBehaviourAction::Dial { .. } => true,
_ => false,
})
.count(),
@ -3072,7 +3060,7 @@ mod tests {
gs.events
.iter()
.filter(|e| match e {
NetworkBehaviourAction::DialPeer { .. } => true,
NetworkBehaviourAction::Dial { .. } => true,
_ => false,
})
.count()

4
protocols/gossipsub/tests/smoke.rs
View File

@ -37,7 +37,7 @@ use libp2p_gossipsub::{
ValidationMode,
};
use libp2p_plaintext::PlainText2Config;
use libp2p_swarm::{Swarm, SwarmEvent};
use libp2p_swarm::{dial_opts::DialOpts, Swarm, SwarmEvent};
use libp2p_yamux as yamux;
struct Graph {
@ -98,7 +98,7 @@ impl Graph {
p2p_suffix_connected.unwrap()
);
Swarm::dial_addr(&mut next.1, connected_addr_no_p2p).unwrap();
next.1.dial(connected_addr_no_p2p).unwrap();
connected_nodes.push(next);
}

4
protocols/identify/CHANGELOG.md
View File

@ -1,7 +1,9 @@
# 0.31.1 [unreleased]
# 0.32.0 [unreleased]
- Use `futures-timer` instead of `wasm-timer` (see [PR 2245]).
- Update dependencies.
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
# 0.31.0 [2021-11-01]

4
protocols/identify/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-identify"
edition = "2018"
description = "Nodes identifcation protocol for libp2p"
version = "0.31.1"
version = "0.32.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -13,7 +13,7 @@ categories = ["network-programming", "asynchronous"]
futures = "0.3.1"
futures-timer = "3.0.2"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
log = "0.4.1"
lru = "0.6"
prost = "0.9"

35
protocols/identify/src/identify.rs
View File

@ -27,8 +27,9 @@ use libp2p_core::{
ConnectedPoint, Multiaddr, PeerId, PublicKey,
};
use libp2p_swarm::{
AddressScore, DialError, DialPeerCondition, IntoProtocolsHandler, NegotiatedSubstream,
NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters, ProtocolsHandler,
dial_opts::{self, DialOpts},
AddressScore, DialError, IntoProtocolsHandler, NegotiatedSubstream, NetworkBehaviour,
NetworkBehaviourAction, NotifyHandler, PollParameters, ProtocolsHandler,
ProtocolsHandlerUpgrErr,
};
use lru::LruCache;
@ -198,9 +199,10 @@ impl Identify {
if self.pending_push.insert(p) {
if !self.connected.contains_key(&p) {
let handler = self.new_handler();
self.events.push_back(NetworkBehaviourAction::DialPeer {
peer_id: p,
condition: DialPeerCondition::Disconnected,
self.events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(p)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
}
@ -561,7 +563,7 @@ mod tests {
}
}
});
swarm2.dial_addr(listen_addr).unwrap();
swarm2.dial(listen_addr).unwrap();
// nb. Either swarm may receive the `Identified` event first, upon which
// it will permit the connection to be closed, as defined by
@ -641,7 +643,7 @@ mod tests {
}
});
Swarm::dial_addr(&mut swarm2, listen_addr).unwrap();
swarm2.dial(listen_addr).unwrap();
async_std::task::block_on(async move {
loop {
@ -728,9 +730,9 @@ mod tests {
}
});
swarm2.dial_addr(listen_addr).unwrap();
swarm2.dial(listen_addr).unwrap();
// wait until we identified
// Wait until we identified.
async_std::task::block_on(async {
loop {
if let SwarmEvent::Behaviour(IdentifyEvent::Received { .. }) =
@ -743,8 +745,19 @@ mod tests {
swarm2.disconnect_peer_id(swarm1_peer_id).unwrap();
// we should still be able to dial now!
swarm2.dial(&swarm1_peer_id).unwrap();
// Wait for connection to close.
async_std::task::block_on(async {
loop {
if let SwarmEvent::ConnectionClosed { peer_id, .. } =
swarm2.select_next_some().await
{
break peer_id;
}
}
});
// We should still be able to dial now!
swarm2.dial(swarm1_peer_id).unwrap();
let connected_peer = async_std::task::block_on(async {
loop {

2
protocols/kad/CHANGELOG.md
View File

@ -8,6 +8,8 @@
- Populate the `key` field when converting `KadRequestMsg::PutValue` to `proto::Message` (see [PR 2309]).
- Update dependencies.
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
[PR 2297]: https://github.com/libp2p/rust-libp2p/pull/2297
[PR 2309]: https://github.com/libp2p/rust-libp2p/pull/2309

2
protocols/kad/Cargo.toml
View File

@ -18,7 +18,7 @@ asynchronous-codec = "0.6"
futures = "0.3.1"
log = "0.4"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
prost = "0.9"
rand = "0.7.2"
sha2 = "0.9.1"

32
protocols/kad/src/behaviour.rs
View File

@ -44,8 +44,8 @@ use libp2p_core::{
ConnectedPoint, Multiaddr, PeerId,
};
use libp2p_swarm::{
DialError, DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
PollParameters,
dial_opts::{self, DialOpts},
DialError, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
};
use log::{debug, info, warn};
use smallvec::SmallVec;
@ -564,10 +564,10 @@ where
}
kbucket::InsertResult::Pending { disconnected } => {
let handler = self.new_handler();
self.queued_events
.push_back(NetworkBehaviourAction::DialPeer {
peer_id: disconnected.into_preimage(),
condition: DialPeerCondition::Disconnected,
self.queued_events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(disconnected.into_preimage())
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
RoutingUpdate::Pending
@ -1152,10 +1152,10 @@ where
// Only try dialing peer if not currently connected.
if !self.connected_peers.contains(disconnected.preimage()) {
let handler = self.new_handler();
self.queued_events
.push_back(NetworkBehaviourAction::DialPeer {
peer_id: disconnected.into_preimage(),
condition: DialPeerCondition::Disconnected,
self.queued_events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(disconnected.into_preimage())
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
})
}
@ -1934,12 +1934,12 @@ where
}
}
DialError::DialPeerConditionFalse(
DialPeerCondition::Disconnected | DialPeerCondition::NotDialing,
dial_opts::PeerCondition::Disconnected | dial_opts::PeerCondition::NotDialing,
) => {
// We might (still) be connected, or about to be connected, thus do not report the
// failure to the queries.
}
DialError::DialPeerConditionFalse(DialPeerCondition::Always) => {
DialError::DialPeerConditionFalse(dial_opts::PeerCondition::Always) => {
unreachable!("DialPeerCondition::Always can not trigger DialPeerConditionFalse.");
}
}
@ -2321,10 +2321,10 @@ where
} else if &peer_id != self.kbuckets.local_key().preimage() {
query.inner.pending_rpcs.push((peer_id, event));
let handler = self.new_handler();
self.queued_events
.push_back(NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
self.queued_events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(peer_id)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
}

4
protocols/mdns/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.33.0 [unreleased]
- Update dependencies.
# 0.32.0 [2021-11-01]
- Make default features of `libp2p-core` optional.

4
protocols/mdns/Cargo.toml
View File

@ -1,7 +1,7 @@
[package]
name = "libp2p-mdns"
edition = "2018"
version = "0.32.0"
version = "0.33.0"
description = "Implementation of the libp2p mDNS discovery method"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
@ -17,7 +17,7 @@ futures = "0.3.13"
if-watch = "0.2.0"
lazy_static = "1.4.0"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
log = "0.4.14"
rand = "0.8.3"
smallvec = "1.6.1"

4
protocols/ping/CHANGELOG.md
View File

@ -1,7 +1,9 @@
# 0.31.1 [unreleased]
# 0.32.0 [unreleased]
- Use `instant` and `futures-timer` instead of `wasm-timer` (see [PR 2245]).
- Update dependencies.
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
# 0.31.0 [2021-11-01]

4
protocols/ping/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-ping"
edition = "2018"
description = "Ping protocol for libp2p"
version = "0.31.1"
version = "0.32.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -14,7 +14,7 @@ futures = "0.3.1"
futures-timer = "3.0.2"
instant = "0.1.11"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
log = "0.4.1"
rand = "0.7.2"
void = "1.0"

8
protocols/ping/tests/ping.rs
View File

@ -30,7 +30,7 @@ use libp2p_core::{
use libp2p_mplex as mplex;
use libp2p_noise as noise;
use libp2p_ping as ping;
use libp2p_swarm::{DummyBehaviour, KeepAlive, Swarm, SwarmEvent};
use libp2p_swarm::{dial_opts::DialOpts, DummyBehaviour, KeepAlive, Swarm, SwarmEvent};
use libp2p_tcp::TcpConfig;
use libp2p_yamux as yamux;
use quickcheck::*;
@ -82,7 +82,7 @@ fn ping_pong() {
let pid2 = peer2_id.clone();
let peer2 = async move {
swarm2.dial_addr(rx.next().await.unwrap()).unwrap();
swarm2.dial(rx.next().await.unwrap()).unwrap();
loop {
match swarm2.select_next_some().await {
@ -156,7 +156,7 @@ fn max_failures() {
};
let peer2 = async move {
swarm2.dial_addr(rx.next().await.unwrap()).unwrap();
swarm2.dial(rx.next().await.unwrap()).unwrap();
let mut count2: u8 = 0;
@ -216,7 +216,7 @@ fn unsupported_doesnt_fail() {
});
let result = async_std::task::block_on(async move {
swarm2.dial_addr(rx.next().await.unwrap()).unwrap();
swarm2.dial(rx.next().await.unwrap()).unwrap();
loop {
match swarm2.select_next_some().await {

4
protocols/relay/CHANGELOG.md
View File

@ -1,7 +1,9 @@
# 0.4.1 [unreleased]
# 0.5.0 [unreleased]
- Use `instant` instead of `wasm-timer` (see [PR 2245]).
- Update dependencies.
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
# 0.4.0 [2021-11-01]

4
protocols/relay/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-relay"
edition = "2018"
description = "Communications relaying for libp2p"
version = "0.4.1"
version = "0.5.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -16,7 +16,7 @@ futures = "0.3.1"
futures-timer = "3"
instant = "0.1.11"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
log = "0.4"
pin-project = "1"
prost = "0.9"

6
protocols/relay/examples/relay.rs
View File

@ -65,7 +65,7 @@ use libp2p::relay::{Relay, RelayConfig};
use libp2p::swarm::SwarmEvent;
use libp2p::tcp::TcpConfig;
use libp2p::Transport;
use libp2p::{core::upgrade, identity::ed25519, ping};
use libp2p::{core::upgrade, identity::ed25519, ping, Multiaddr};
use libp2p::{identity, NetworkBehaviour, PeerId, Swarm};
use std::error::Error;
use std::task::{Context, Poll};
@ -129,8 +129,8 @@ fn main() -> Result<(), Box<dyn Error>> {
println!("starting client listener via relay on {}", &relay_address);
}
Mode::ClientDial => {
let client_listen_address = get_client_listen_address(&opt);
swarm.dial_addr(client_listen_address.parse()?)?;
let client_listen_address: Multiaddr = get_client_listen_address(&opt).parse()?;
swarm.dial(client_listen_address.clone())?;
println!("starting as client dialer on {}", client_listen_address);
}
}

28
protocols/relay/src/behaviour.rs
View File

@ -29,8 +29,9 @@ use libp2p_core::connection::{ConnectedPoint, ConnectionId, ListenerId};
use libp2p_core::multiaddr::Multiaddr;
use libp2p_core::PeerId;
use libp2p_swarm::{
DialError, DialPeerCondition, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction,
NotifyHandler, PollParameters,
dial_opts::{self, DialOpts},
DialError, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
PollParameters,
};
use std::collections::{hash_map::Entry, HashMap, HashSet, VecDeque};
use std::task::{Context, Poll};
@ -310,7 +311,7 @@ impl NetworkBehaviour for Relay {
error: &DialError,
) {
if let DialError::DialPeerConditionFalse(
DialPeerCondition::Disconnected | DialPeerCondition::NotDialing,
dial_opts::PeerCondition::Disconnected | dial_opts::PeerCondition::NotDialing,
) = error
{
// Return early. The dial, that this dial was canceled for, might still succeed.
@ -483,9 +484,10 @@ impl NetworkBehaviour for Relay {
);
let handler = self.new_handler();
self.outbox_to_swarm
.push_back(NetworkBehaviourAction::DialPeer {
peer_id: dest_id,
condition: DialPeerCondition::NotDialing,
.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(dest_id)
.condition(dial_opts::PeerCondition::NotDialing)
.build(),
handler,
});
} else {
@ -676,9 +678,10 @@ impl NetworkBehaviour for Relay {
dst_peer_id,
send_back,
});
return Poll::Ready(NetworkBehaviourAction::DialPeer {
peer_id: relay_peer_id,
condition: DialPeerCondition::Disconnected,
return Poll::Ready(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(relay_peer_id)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler: self.new_handler(),
});
}
@ -743,9 +746,10 @@ impl NetworkBehaviour for Relay {
to_listener,
},
);
return Poll::Ready(NetworkBehaviourAction::DialPeer {
peer_id: relay_peer_id,
condition: DialPeerCondition::Disconnected,
return Poll::Ready(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(relay_peer_id)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler: self.new_handler(),
});
}

4
protocols/relay/src/lib.rs
View File

@ -26,7 +26,7 @@
//! ```rust
//! # use libp2p_core::transport::memory::MemoryTransport;
//! # use libp2p_relay::{RelayConfig, new_transport_and_behaviour};
//! # use libp2p_swarm::Swarm;
//! # use libp2p_swarm::{Swarm, dial_opts::DialOpts};
//! # use libp2p_core::{identity, Multiaddr, multiaddr::Protocol, PeerId, upgrade, Transport};
//! # use libp2p_yamux::YamuxConfig;
//! # use libp2p_plaintext::PlainText2Config;
@ -62,7 +62,7 @@
//! swarm.listen_on(relay_addr).unwrap();
//!
//! // Dial node (5678) via relay node (1234).
//! swarm.dial_addr(dst_addr).unwrap();
//! swarm.dial(dst_addr).unwrap();
//! ```
//!
//! ## Terminology

32
protocols/relay/tests/lib.rs
View File

@ -36,7 +36,7 @@ use libp2p_plaintext::PlainText2Config;
use libp2p_relay::{Relay, RelayConfig};
use libp2p_swarm::protocols_handler::KeepAlive;
use libp2p_swarm::{
DialError, DummyBehaviour, NetworkBehaviour, NetworkBehaviourAction,
dial_opts::DialOpts, DialError, DummyBehaviour, NetworkBehaviour, NetworkBehaviourAction,
NetworkBehaviourEventProcess, PollParameters, Swarm, SwarmEvent,
};
use std::task::{Context, Poll};
@ -146,7 +146,7 @@ fn src_connect_to_dst_listening_via_relay() {
}
};
src_swarm.dial_addr(dst_addr_via_relay).unwrap();
src_swarm.dial(dst_addr_via_relay).unwrap();
let src = async move {
// Source Node dialing Relay to connect to Destination Node.
match src_swarm.select_next_some().await {
@ -221,7 +221,7 @@ fn src_connect_to_dst_not_listening_via_active_relay() {
dst_swarm.listen_on(Protocol::P2pCircuit.into()).unwrap();
spawn_swarm_on_pool(&pool, dst_swarm);
src_swarm.dial_addr(dst_addr_via_relay).unwrap();
src_swarm.dial(dst_addr_via_relay).unwrap();
pool.run_until(async move {
// Source Node dialing Relay to connect to Destination Node.
match src_swarm.select_next_some().await {
@ -307,7 +307,7 @@ fn src_connect_to_dst_via_established_connection_to_relay() {
spawn_swarm_on_pool(&pool, dst_swarm);
pool.run_until(async move {
src_swarm.dial_addr(relay_addr).unwrap();
src_swarm.dial(relay_addr).unwrap();
// Source Node establishing connection to Relay.
loop {
@ -321,7 +321,7 @@ fn src_connect_to_dst_via_established_connection_to_relay() {
}
}
src_swarm.dial_addr(dst_addr_via_relay).unwrap();
src_swarm.dial(dst_addr_via_relay).unwrap();
// Source Node establishing connection to destination node via Relay.
loop {
@ -375,7 +375,7 @@ fn src_try_connect_to_offline_dst() {
relay_swarm.listen_on(relay_addr.clone()).unwrap();
spawn_swarm_on_pool(&pool, relay_swarm);
src_swarm.dial_addr(dst_addr_via_relay.clone()).unwrap();
src_swarm.dial(dst_addr_via_relay.clone()).unwrap();
pool.run_until(async move {
// Source Node dialing Relay to connect to Destination Node.
match src_swarm.select_next_some().await {
@ -433,7 +433,7 @@ fn src_try_connect_to_unsupported_dst() {
dst_swarm.listen_on(dst_addr.clone()).unwrap();
spawn_swarm_on_pool(&pool, dst_swarm);
src_swarm.dial_addr(dst_addr_via_relay.clone()).unwrap();
src_swarm.dial(dst_addr_via_relay.clone()).unwrap();
pool.run_until(async move {
// Source Node dialing Relay to connect to Destination Node.
match src_swarm.select_next_some().await {
@ -484,7 +484,7 @@ fn src_try_connect_to_offline_dst_via_offline_relay() {
.with(dst_addr.into_iter().next().unwrap())
.with(Protocol::P2p(dst_peer_id.clone().into()));
src_swarm.dial_addr(dst_addr_via_relay.clone()).unwrap();
src_swarm.dial(dst_addr_via_relay.clone()).unwrap();
pool.run_until(async move {
// Source Node dialing Relay to connect to Destination Node.
match src_swarm.select_next_some().await {
@ -761,7 +761,7 @@ fn inactive_connection_timeout() {
spawn_swarm_on_pool(&pool, dst_swarm);
pool.run_until(async move {
src_swarm.dial_addr(relay_addr).unwrap();
src_swarm.dial(relay_addr).unwrap();
// Source Node dialing Relay.
loop {
match src_swarm.select_next_some().await {
@ -773,7 +773,7 @@ fn inactive_connection_timeout() {
}
}
src_swarm.dial_addr(dst_addr_via_relay).unwrap();
src_swarm.dial(dst_addr_via_relay).unwrap();
// Source Node establishing connection to destination node via Relay.
match src_swarm.select_next_some().await {
@ -841,8 +841,8 @@ fn concurrent_connection_same_relay_same_dst() {
spawn_swarm_on_pool(&pool, dst_swarm);
pool.run_until(async move {
src_swarm.dial_addr(dst_addr_via_relay.clone()).unwrap();
src_swarm.dial_addr(dst_addr_via_relay).unwrap();
src_swarm.dial(dst_addr_via_relay.clone()).unwrap();
src_swarm.dial(dst_addr_via_relay).unwrap();
// Source Node establishing two connections to destination node via Relay.
let mut num_established = 0;
@ -987,8 +987,8 @@ fn yield_incoming_connection_through_correct_listener() {
}
});
src_1_swarm.dial_addr(dst_addr_via_relay_1.clone()).unwrap();
src_2_swarm.dial_addr(dst_addr_via_relay_2.clone()).unwrap();
src_1_swarm.dial(dst_addr_via_relay_1.clone()).unwrap();
src_2_swarm.dial(dst_addr_via_relay_2.clone()).unwrap();
spawn_swarm_on_pool(&pool, src_1_swarm);
spawn_swarm_on_pool(&pool, src_2_swarm);
@ -1047,7 +1047,7 @@ fn yield_incoming_connection_through_correct_listener() {
// Expect destination node to reject incoming connection from unknown relay given that
// destination node is not listening for such connections.
src_3_swarm.dial_addr(dst_addr_via_relay_3.clone()).unwrap();
src_3_swarm.dial(dst_addr_via_relay_3.clone()).unwrap();
pool.run_until(async {
loop {
futures::select! {
@ -1117,7 +1117,7 @@ fn yield_incoming_connection_through_correct_listener() {
// Expect destination node to accept incoming connection from "unknown" relay, i.e. the
// connection from source node 3 via relay 3.
src_3_swarm.dial_addr(dst_addr_via_relay_3.clone()).unwrap();
src_3_swarm.dial(dst_addr_via_relay_3.clone()).unwrap();
pool.run_until(async move {
loop {
match src_3_swarm.select_next_some().await {

4
protocols/rendezvous/CHANGELOG.md
View File

@ -1,7 +1,9 @@
# 0.1.1 [unreleased]
# 0.2.0 [unreleased]
- Use `instant` and `futures-timer` instead of `wasm-timer` (see [PR 2245]).
- Update dependencies.
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
# 0.1.0 [2021-11-01]

4
protocols/rendezvous/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-rendezvous"
edition = "2018"
description = "Rendezvous protocol for libp2p"
version = "0.1.1"
version = "0.2.0"
authors = ["The COMIT guys <hello@comit.network>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -12,7 +12,7 @@ categories = ["network-programming", "asynchronous"]
[dependencies]
asynchronous-codec = "0.6"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
prost = "0.9"
void = "1"
log = "0.4"

4
protocols/rendezvous/examples/discover.rs
View File

@ -55,7 +55,7 @@ async fn main() {
log::info!("Local peer id: {}", swarm.local_peer_id());
let _ = swarm.dial_addr(rendezvous_point_address.clone());
let _ = swarm.dial(rendezvous_point_address.clone()).unwrap();
let mut discover_tick = tokio::time::interval(Duration::from_secs(30));
let mut cookie = None;
@ -96,7 +96,7 @@ async fn main() {
address.clone()
};
swarm.dial_addr(address_with_p2p).unwrap()
swarm.dial(address_with_p2p).unwrap()
}
}
}

5
protocols/rendezvous/examples/register.rs
View File

@ -22,8 +22,7 @@ use futures::StreamExt;
use libp2p::core::identity;
use libp2p::core::PeerId;
use libp2p::ping::{Ping, PingConfig, PingEvent, PingSuccess};
use libp2p::swarm::Swarm;
use libp2p::swarm::SwarmEvent;
use libp2p::swarm::{Swarm, SwarmEvent};
use libp2p::{development_transport, rendezvous};
use libp2p::{Multiaddr, NetworkBehaviour};
use libp2p_swarm::AddressScore;
@ -58,7 +57,7 @@ async fn main() {
let _ = swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse().unwrap());
swarm.dial_addr(rendezvous_point_address).unwrap();
swarm.dial(rendezvous_point_address).unwrap();
while let Some(event) = swarm.next().await {
match event {

5
protocols/rendezvous/examples/register_with_identify.rs
View File

@ -23,8 +23,7 @@ use libp2p::core::identity;
use libp2p::core::PeerId;
use libp2p::identify::{Identify, IdentifyConfig, IdentifyEvent};
use libp2p::ping::{Ping, PingConfig, PingEvent, PingSuccess};
use libp2p::swarm::Swarm;
use libp2p::swarm::SwarmEvent;
use libp2p::swarm::{Swarm, SwarmEvent};
use libp2p::{development_transport, rendezvous};
use libp2p::{Multiaddr, NetworkBehaviour};
use std::time::Duration;
@ -61,7 +60,7 @@ async fn main() {
let _ = swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse().unwrap());
swarm.dial_addr(rendezvous_point_address).unwrap();
swarm.dial(rendezvous_point_address).unwrap();
while let Some(event) = swarm.next().await {
match event {

6
protocols/rendezvous/tests/harness.rs
View File

@ -29,7 +29,9 @@ use libp2p::core::upgrade::SelectUpgrade;
use libp2p::core::{identity, Multiaddr, PeerId, Transport};
use libp2p::mplex::MplexConfig;
use libp2p::noise::{Keypair, NoiseConfig, X25519Spec};
use libp2p::swarm::{AddressScore, NetworkBehaviour, Swarm, SwarmBuilder, SwarmEvent};
use libp2p::swarm::{
dial_opts::DialOpts, AddressScore, NetworkBehaviour, Swarm, SwarmBuilder, SwarmEvent,
};
use libp2p::yamux::YamuxConfig;
use std::fmt::Debug;
use std::time::Duration;
@ -163,7 +165,7 @@ where
{
let addr_to_dial = other.external_addresses().next().unwrap().addr.clone();
self.dial_addr(addr_to_dial.clone()).unwrap();
self.dial(addr_to_dial.clone()).unwrap();
let mut dialer_done = false;
let mut listener_done = false;

7
protocols/rendezvous/tests/rendezvous.rs
View File

@ -26,8 +26,7 @@ use futures::stream::FuturesUnordered;
use futures::StreamExt;
use libp2p_core::identity;
use libp2p_rendezvous as rendezvous;
use libp2p_swarm::DialError;
use libp2p_swarm::{Swarm, SwarmEvent};
use libp2p_swarm::{dial_opts::DialOpts, DialError, Swarm, SwarmEvent};
use std::convert::TryInto;
use std::time::Duration;
@ -187,7 +186,7 @@ async fn discover_allows_for_dial_by_peer_id() {
let alices_peer_id = *alice.local_peer_id();
let bobs_peer_id = *bob.local_peer_id();
bob.dial(&alices_peer_id).unwrap();
bob.dial(alices_peer_id).unwrap();
let alice_connected_to = tokio::spawn(async move {
loop {
@ -296,7 +295,7 @@ async fn registration_on_clients_expire() {
tokio::time::sleep(Duration::from_secs(registration_ttl + 5)).await;
let event = bob.select_next_some().await;
let error = bob.dial(alice.local_peer_id()).unwrap_err();
let error = bob.dial(*alice.local_peer_id()).unwrap_err();
assert!(matches!(
event,

4
protocols/request-response/CHANGELOG.md
View File

@ -1,7 +1,9 @@
# 0.13.1 [unreleased]
# 0.14.0 [unreleased]
- Use `instant` instead of `wasm-timer` (see [PR 2245]).
- Update dependencies.
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
# 0.13.0 [2021-11-01]

4
protocols/request-response/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-request-response"
edition = "2018"
description = "Generic Request/Response Protocols"
version = "0.13.1"
version = "0.14.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -15,7 +15,7 @@ bytes = "1"
futures = "0.3.1"
instant = "0.1.11"
libp2p-core = { version = "0.30.0", path = "../../core", default-features = false }
libp2p-swarm = { version = "0.31.0", path = "../../swarm" }
libp2p-swarm = { version = "0.32.0", path = "../../swarm" }
log = "0.4.11"
lru = "0.7"
rand = "0.7"

13
protocols/request-response/src/lib.rs
View File

@ -66,8 +66,9 @@ use futures::channel::oneshot;
use handler::{RequestProtocol, RequestResponseHandler, RequestResponseHandlerEvent};
use libp2p_core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
use libp2p_swarm::{
DialError, DialPeerCondition, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction,
NotifyHandler, PollParameters,
dial_opts::{self, DialOpts},
DialError, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
PollParameters,
};
use smallvec::SmallVec;
use std::{
@ -385,10 +386,10 @@ where
if let Some(request) = self.try_send_request(peer, request) {
let handler = self.new_handler();
self.pending_events
.push_back(NetworkBehaviourAction::DialPeer {
peer_id: *peer,
condition: DialPeerCondition::Disconnected,
self.pending_events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(*peer)
.condition(dial_opts::PeerCondition::Disconnected)
.build(),
handler,
});
self.pending_outbound_requests

16
src/tutorial.rs
View File

@ -222,9 +222,9 @@
//!
//! ```rust
//! use futures::executor::block_on;
//! use libp2p::{identity, PeerId};
//! use libp2p::{identity, Multiaddr, PeerId};
//! use libp2p::ping::{Ping, PingConfig};
//! use libp2p::swarm::Swarm;
//! use libp2p::swarm::{Swarm, dial_opts::DialOpts};
//! use std::error::Error;
//!
//! fn main() -> Result<(), Box<dyn Error>> {
@ -250,8 +250,8 @@
//! // Dial the peer identified by the multi-address given as the second
//! // command-line argument, if any.
//! if let Some(addr) = std::env::args().nth(1) {
//! let remote = addr.parse()?;
//! swarm.dial_addr(remote)?;
//! let remote: Multiaddr = addr.parse()?;
//! swarm.dial(remote)?;
//! println!("Dialed {}", addr)
//! }
//!
@ -269,8 +269,8 @@
//! use futures::executor::block_on;
//! use futures::prelude::*;
//! use libp2p::ping::{Ping, PingConfig};
//! use libp2p::swarm::{Swarm, SwarmEvent};
//! use libp2p::{identity, PeerId};
//! use libp2p::swarm::{Swarm, SwarmEvent, dial_opts::DialOpts};
//! use libp2p::{identity, Multiaddr, PeerId};
//! use std::error::Error;
//! use std::task::Poll;
//!
@ -297,8 +297,8 @@
//! // Dial the peer identified by the multi-address given as the second
//! // command-line argument, if any.
//! if let Some(addr) = std::env::args().nth(1) {
//! let remote = addr.parse()?;
//! swarm.dial_addr(remote)?;
//! let remote: Multiaddr = addr.parse()?;
//! swarm.dial(remote)?;
//! println!("Dialed {}", addr)
//! }
//!

2
swarm-derive/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-swarm-derive"
edition = "2018"
description = "Procedural macros of libp2p-core"
version = "0.25.0"
version = "0.26.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"

15
swarm-derive/src/lib.rs
View File

@ -554,10 +554,10 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
wrapped_event = quote!{ #either_ident::First(#wrapped_event) };
}
// `DialPeer` and `DialAddress` each provide a handler of the specific
// behaviour triggering the event. Though in order for the final handler
// to be able to handle protocols of all behaviours, the provided
// handler needs to be combined with handlers of all other behaviours.
// `Dial` provides a handler of the specific behaviour triggering the
// event. Though in order for the final handler to be able to handle
// protocols of all behaviours, the provided handler needs to be
// combined with handlers of all other behaviours.
let provided_handler_and_new_handlers = {
let mut out_handler = None;
@ -608,11 +608,8 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
loop {
match #trait_to_impl::poll(&mut #field_name, cx, poll_params) {
#generate_event_match_arm
std::task::Poll::Ready(#network_behaviour_action::DialAddress { address, handler: provided_handler }) => {
return std::task::Poll::Ready(#network_behaviour_action::DialAddress { address, handler: #provided_handler_and_new_handlers });
}
std::task::Poll::Ready(#network_behaviour_action::DialPeer { peer_id, condition, handler: provided_handler }) => {
return std::task::Poll::Ready(#network_behaviour_action::DialPeer { peer_id, condition, handler: #provided_handler_and_new_handlers });
std::task::Poll::Ready(#network_behaviour_action::Dial { opts, handler: provided_handler }) => {
return std::task::Poll::Ready(#network_behaviour_action::Dial { opts, handler: #provided_handler_and_new_handlers });
}
std::task::Poll::Ready(#network_behaviour_action::NotifyHandler { peer_id, handler, event }) => {
return std::task::Poll::Ready(#network_behaviour_action::NotifyHandler {

42
swarm/CHANGELOG.md
View File

@ -1,8 +1,48 @@
# 0.31.1 [unreleased]
# 0.32.0 [unreleased]
- Use `instant` and `futures-timer` instead of `wasm-timer` (see [PR 2245]).
- Enable advanced dialing requests both on `Swarm::dial` and via
`NetworkBehaviourAction::Dial`. Users can now trigger a dial with a specific
set of addresses, optionally extended via
`NetworkBehaviour::addresses_of_peer`.
Changes required to maintain status quo:
- Previously `swarm.dial(peer_id)`
now `swarm.dial(DialOpts::peer_id(peer_id).build())`
or `swarm.dial(peer_id)` given that `DialOpts` implements `From<PeerId>`.
- Previously `swarm.dial_addr(addr)`
now `swarm.dial(DialOpts::unknown_peer_id().address(addr).build())`
or `swarm.dial(addr)` given that `DialOpts` implements `From<Multiaddr>`.
- Previously `NetworkBehaviourAction::DialPeer { peer_id, condition, handler }`
now
```rust
NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(peer_id)
.condition(condition)
.build(),
handler,
}
```
- Previously `NetworkBehaviourAction::DialAddress { address, handler }`
now
```rust
NetworkBehaviourAction::Dial {
opts: DialOpts::unknown_peer_id()
.address(address)
.build(),
handler,
}
```
See [PR 2317].
[PR 2245]: https://github.com/libp2p/rust-libp2p/pull/2245
[PR 2317]: https://github.com/libp2p/rust-libp2p/pull/2317
# 0.31.0 [2021-11-01]

2
swarm/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-swarm"
edition = "2018"
description = "The libp2p swarm"
version = "0.31.1"
version = "0.32.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"

136
swarm/src/behaviour.rs
View File

@ -18,6 +18,7 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::dial_opts::DialOpts;
use crate::protocols_handler::{IntoProtocolsHandler, ProtocolsHandler};
use crate::{AddressRecord, AddressScore, DialError};
use libp2p_core::{
@ -265,31 +266,7 @@ pub enum NetworkBehaviourAction<
/// Instructs the `Swarm` to return an event when it is being polled.
GenerateEvent(TOutEvent),
/// Instructs the swarm to dial the given multiaddress optionally including a [`PeerId`].
///
/// On success, [`NetworkBehaviour::inject_connection_established`] is invoked.
/// On failure, [`NetworkBehaviour::inject_dial_failure`] is invoked.
///
/// Note that the provided handler is returned to the [`NetworkBehaviour`] on connection failure
/// and connection closing. Thus it can be used to carry state, which otherwise would have to be
/// tracked in the [`NetworkBehaviour`] itself. E.g. a message destined to an unconnected peer
/// can be included in the handler, and thus directly send on connection success or extracted by
/// the [`NetworkBehaviour`] on connection failure. See [`NetworkBehaviourAction::DialPeer`] for
/// example.
DialAddress {
/// The address to dial.
address: Multiaddr,
/// The handler to be used to handle the connection to the peer.
handler: THandler,
},
/// Instructs the swarm to dial a known `PeerId`.
///
/// The [`NetworkBehaviour::addresses_of_peer`] method is called to determine which addresses to
/// attempt to reach.
///
/// If we were already trying to dial this node, the addresses that are not yet in the queue of
/// addresses to try are added back to this queue.
/// Instructs the swarm to start a dial.
///
/// On success, [`NetworkBehaviour::inject_connection_established`] is invoked.
/// On failure, [`NetworkBehaviour::inject_dial_failure`] is invoked.
@ -300,7 +277,7 @@ pub enum NetworkBehaviourAction<
/// can be included in the handler, and thus directly send on connection success or extracted by
/// the [`NetworkBehaviour`] on connection failure.
///
/// # Example
/// # Example carrying state in the handler
///
/// ```rust
/// # use futures::executor::block_on;
@ -312,10 +289,11 @@ pub enum NetworkBehaviourAction<
/// # use libp2p::core::PeerId;
/// # use libp2p::plaintext::PlainText2Config;
/// # use libp2p::swarm::{
/// # DialError, DialPeerCondition, IntoProtocolsHandler, KeepAlive, NegotiatedSubstream,
/// # DialError, IntoProtocolsHandler, KeepAlive, NegotiatedSubstream,
/// # NetworkBehaviour, NetworkBehaviourAction, PollParameters, ProtocolsHandler,
/// # ProtocolsHandlerEvent, ProtocolsHandlerUpgrErr, SubstreamProtocol, Swarm, SwarmEvent,
/// # };
/// # use libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
/// # use libp2p::yamux;
/// # use std::collections::VecDeque;
/// # use std::task::{Context, Poll};
@ -350,21 +328,22 @@ pub enum NetworkBehaviourAction<
/// );
/// });
///
/// # #[derive(Default)]
/// # struct MyBehaviour {
/// # outbox_to_swarm: VecDeque<NetworkBehaviourAction<PreciousMessage, MyHandler>>,
/// # }
/// #
/// # impl MyBehaviour {
/// # fn send(&mut self, peer_id: PeerId, msg: PreciousMessage) {
/// # self.outbox_to_swarm
/// # .push_back(NetworkBehaviourAction::DialPeer {
/// # peer_id,
/// # condition: DialPeerCondition::Always,
/// # handler: MyHandler { message: Some(msg) },
/// # });
/// # }
/// # }
/// #[derive(Default)]
/// struct MyBehaviour {
/// outbox_to_swarm: VecDeque<NetworkBehaviourAction<PreciousMessage, MyHandler>>,
/// }
///
/// impl MyBehaviour {
/// fn send(&mut self, peer_id: PeerId, msg: PreciousMessage) {
/// self.outbox_to_swarm
/// .push_back(NetworkBehaviourAction::Dial {
/// opts: DialOpts::peer_id(peer_id)
/// .condition(PeerCondition::Always)
/// .build(),
/// handler: MyHandler { message: Some(msg) },
/// });
/// }
/// }
/// #
/// impl NetworkBehaviour for MyBehaviour {
/// # type ProtocolsHandler = MyHandler;
@ -472,14 +451,7 @@ pub enum NetworkBehaviourAction<
/// # #[derive(Debug, PartialEq, Eq)]
/// # struct PreciousMessage(String);
/// ```
DialPeer {
/// The peer to try reach.
peer_id: PeerId,
/// The condition for initiating a new dialing attempt.
condition: DialPeerCondition,
/// The handler to be used to handle the connection to the peer.
handler: THandler,
},
Dial { opts: DialOpts, handler: THandler },
/// Instructs the `Swarm` to send an event to the handler dedicated to a
/// connection with a peer.
@ -549,18 +521,9 @@ impl<TOutEvent, THandler: IntoProtocolsHandler, TInEventOld>
) -> NetworkBehaviourAction<TOutEvent, THandler, TInEventNew> {
match self {
NetworkBehaviourAction::GenerateEvent(e) => NetworkBehaviourAction::GenerateEvent(e),
NetworkBehaviourAction::DialAddress { address, handler } => {
NetworkBehaviourAction::DialAddress { address, handler }
NetworkBehaviourAction::Dial { opts, handler } => {
NetworkBehaviourAction::Dial { opts, handler }
}
NetworkBehaviourAction::DialPeer {
peer_id,
condition,
handler,
} => NetworkBehaviourAction::DialPeer {
peer_id,
condition,
handler,
},
NetworkBehaviourAction::NotifyHandler {
peer_id,
handler,
@ -589,18 +552,9 @@ impl<TOutEvent, THandler: IntoProtocolsHandler> NetworkBehaviourAction<TOutEvent
pub fn map_out<E>(self, f: impl FnOnce(TOutEvent) -> E) -> NetworkBehaviourAction<E, THandler> {
match self {
NetworkBehaviourAction::GenerateEvent(e) => NetworkBehaviourAction::GenerateEvent(f(e)),
NetworkBehaviourAction::DialAddress { address, handler } => {
NetworkBehaviourAction::DialAddress { address, handler }
NetworkBehaviourAction::Dial { opts, handler } => {
NetworkBehaviourAction::Dial { opts, handler }
}
NetworkBehaviourAction::DialPeer {
peer_id,
condition,
handler,
} => NetworkBehaviourAction::DialPeer {
peer_id,
condition,
handler,
},
NetworkBehaviourAction::NotifyHandler {
peer_id,
handler,
@ -640,19 +594,8 @@ where
{
match self {
NetworkBehaviourAction::GenerateEvent(e) => NetworkBehaviourAction::GenerateEvent(e),
NetworkBehaviourAction::DialAddress { address, handler } => {
NetworkBehaviourAction::DialAddress {
address,
handler: f(handler),
}
}
NetworkBehaviourAction::DialPeer {
peer_id,
condition,
handler,
} => NetworkBehaviourAction::DialPeer {
peer_id,
condition,
NetworkBehaviourAction::Dial { opts, handler } => NetworkBehaviourAction::Dial {
opts,
handler: f(handler),
},
NetworkBehaviourAction::NotifyHandler {
@ -687,29 +630,6 @@ pub enum NotifyHandler {
Any,
}
/// The available conditions under which a new dialing attempt to
/// a peer is initiated when requested by [`NetworkBehaviourAction::DialPeer`].
#[derive(Debug, Copy, Clone)]
pub enum DialPeerCondition {
/// A new dialing attempt is initiated _only if_ the peer is currently
/// considered disconnected, i.e. there is no established connection
/// and no ongoing dialing attempt.
Disconnected,
/// A new dialing attempt is initiated _only if_ there is currently
/// no ongoing dialing attempt, i.e. the peer is either considered
/// disconnected or connected but without an ongoing dialing attempt.
NotDialing,
/// A new dialing attempt is always initiated, only subject to the
/// configured connection limits.
Always,
}
impl Default for DialPeerCondition {
fn default() -> Self {
DialPeerCondition::Disconnected
}
}
/// The options which connections to close.
#[derive(Debug, Clone)]
pub enum CloseConnection {

217
swarm/src/dial_opts.rs Normal file
View File

@ -0,0 +1,217 @@
// Copyright 2019 Parity Technologies (UK) Ltd.
// Copyright 2021 Protocol Labs.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use libp2p_core::{Multiaddr, PeerId};
/// Options to configure a dial to a known or unknown peer.
///
/// Used in [`Swarm::dial`](crate::Swarm::dial) and
/// [`NetworkBehaviourAction::Dial`](crate::behaviour::NetworkBehaviourAction::Dial).
///
/// To construct use either of:
///
/// - [`DialOpts::peer_id`] dialing a known peer
///
/// - [`DialOpts::unknown_peer_id`] dialing an unknown peer
#[derive(Debug)]
pub struct DialOpts(pub(super) Opts);
impl DialOpts {
/// Dial a known peer.
///
/// ```
/// # use libp2p_swarm::dial_opts::{DialOpts, PeerCondition};
/// # use libp2p_core::PeerId;
/// DialOpts::peer_id(PeerId::random())
/// .condition(PeerCondition::Disconnected)
/// .addresses(vec!["/ip6/::1/tcp/12345".parse().unwrap()])
/// .extend_addresses_through_behaviour()
/// .build();
/// ```
pub fn peer_id(peer_id: PeerId) -> WithPeerId {
WithPeerId {
peer_id,
condition: Default::default(),
}
}
/// Dial an unknown peer.
///
/// ```
/// # use libp2p_swarm::dial_opts::DialOpts;
/// DialOpts::unknown_peer_id()
/// .address("/ip6/::1/tcp/12345".parse().unwrap())
/// .build();
/// ```
pub fn unknown_peer_id() -> WithoutPeerId {
WithoutPeerId {}
}
/// Get the [`PeerId`] specified in a [`DialOpts`] if any.
pub fn get_peer_id(&self) -> Option<PeerId> {
match self {
DialOpts(Opts::WithPeerId(WithPeerId { peer_id, .. })) => Some(*peer_id),
DialOpts(Opts::WithPeerIdWithAddresses(WithPeerIdWithAddresses {
peer_id, ..
})) => Some(*peer_id),
DialOpts(Opts::WithoutPeerIdWithAddress(_)) => None,
}
}
}
impl From<Multiaddr> for DialOpts {
fn from(address: Multiaddr) -> Self {
DialOpts::unknown_peer_id().address(address).build()
}
}
impl From<PeerId> for DialOpts {
fn from(peer_id: PeerId) -> Self {
DialOpts::peer_id(peer_id).build()
}
}
/// Internal options type.
///
/// Not to be constructed manually. Use either of the below instead:
///
/// - [`DialOpts::peer_id`] dialing a known peer
/// - [`DialOpts::unknown_peer_id`] dialing an unknown peer
#[derive(Debug)]
pub(super) enum Opts {
WithPeerId(WithPeerId),
WithPeerIdWithAddresses(WithPeerIdWithAddresses),
WithoutPeerIdWithAddress(WithoutPeerIdWithAddress),
}
#[derive(Debug)]
pub struct WithPeerId {
pub(crate) peer_id: PeerId,
pub(crate) condition: PeerCondition,
}
impl WithPeerId {
/// Specify a [`PeerCondition`] for the dial.
pub fn condition(mut self, condition: PeerCondition) -> Self {
self.condition = condition;
self
}
/// Specify a set of addresses to be used to dial the known peer.
pub fn addresses(self, addresses: Vec<Multiaddr>) -> WithPeerIdWithAddresses {
WithPeerIdWithAddresses {
peer_id: self.peer_id,
condition: self.condition,
addresses,
extend_addresses_through_behaviour: false,
}
}
/// Build the final [`DialOpts`].
///
/// Addresses to dial the peer are retrieved via
/// [`NetworkBehaviour::addresses_of_peer`](crate::behaviour::NetworkBehaviour::addresses_of_peer).
pub fn build(self) -> DialOpts {
DialOpts(Opts::WithPeerId(self))
}
}
#[derive(Debug)]
pub struct WithPeerIdWithAddresses {
pub(crate) peer_id: PeerId,
pub(crate) condition: PeerCondition,
pub(crate) addresses: Vec<Multiaddr>,
pub(crate) extend_addresses_through_behaviour: bool,
}
impl WithPeerIdWithAddresses {
/// Specify a [`PeerCondition`] for the dial.
pub fn condition(mut self, condition: PeerCondition) -> Self {
self.condition = condition;
self
}
/// In addition to the provided addresses, extend the set via
/// [`NetworkBehaviour::addresses_of_peer`](crate::behaviour::NetworkBehaviour::addresses_of_peer).
pub fn extend_addresses_through_behaviour(mut self) -> Self {
self.extend_addresses_through_behaviour = true;
self
}
/// Build the final [`DialOpts`].
pub fn build(self) -> DialOpts {
DialOpts(Opts::WithPeerIdWithAddresses(self))
}
}
#[derive(Debug)]
pub struct WithoutPeerId {}
impl WithoutPeerId {
/// Specify a single address to dial the unknown peer.
pub fn address(self, address: Multiaddr) -> WithoutPeerIdWithAddress {
WithoutPeerIdWithAddress { address }
}
}
#[derive(Debug)]
pub struct WithoutPeerIdWithAddress {
pub(crate) address: Multiaddr,
}
impl WithoutPeerIdWithAddress {
/// Build the final [`DialOpts`].
pub fn build(self) -> DialOpts {
DialOpts(Opts::WithoutPeerIdWithAddress(self))
}
}
/// The available conditions under which a new dialing attempt to
/// a known peer is initiated.
///
/// ```
/// # use libp2p_swarm::dial_opts::{DialOpts, PeerCondition};
/// # use libp2p_core::PeerId;
/// #
/// DialOpts::peer_id(PeerId::random())
/// .condition(PeerCondition::Disconnected)
/// .build();
/// ```
#[derive(Debug, Copy, Clone)]
pub enum PeerCondition {
/// A new dialing attempt is initiated _only if_ the peer is currently
/// considered disconnected, i.e. there is no established connection
/// and no ongoing dialing attempt.
Disconnected,
/// A new dialing attempt is initiated _only if_ there is currently
/// no ongoing dialing attempt, i.e. the peer is either considered
/// disconnected or connected but without an ongoing dialing attempt.
NotDialing,
/// A new dialing attempt is always initiated, only subject to the
/// configured connection limits.
Always,
}
impl Default for PeerCondition {
fn default() -> Self {
PeerCondition::Disconnected
}
}

213
swarm/src/lib.rs
View File

@ -59,12 +59,13 @@ mod registry;
mod test;
mod upgrade;
pub mod dial_opts;
pub mod protocols_handler;
pub mod toggle;
pub use behaviour::{
CloseConnection, DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction,
NetworkBehaviourEventProcess, NotifyHandler, PollParameters,
CloseConnection, NetworkBehaviour, NetworkBehaviourAction, NetworkBehaviourEventProcess,
NotifyHandler, PollParameters,
};
pub use protocols_handler::{
IntoProtocolsHandler, IntoProtocolsHandlerSelect, KeepAlive, OneShotHandler,
@ -73,6 +74,7 @@ pub use protocols_handler::{
};
pub use registry::{AddAddressResult, AddressRecord, AddressScore};
use dial_opts::{DialOpts, PeerCondition};
use futures::{executor::ThreadPoolBuilder, prelude::*, stream::FusedStream};
use libp2p_core::{
connection::{
@ -321,72 +323,151 @@ where
self.network.remove_listener(id)
}
/// Initiates a new dialing attempt to the given address.
pub fn dial_addr(&mut self, addr: Multiaddr) -> Result<(), DialError> {
/// Dial a known or unknown peer.
///
/// See also [`DialOpts`].
///
/// ```
/// # use libp2p_swarm::Swarm;
/// # use libp2p_swarm::dial_opts::{DialOpts, PeerCondition};
/// # use libp2p_core::{Multiaddr, PeerId, Transport};
/// # use libp2p_core::transport::dummy::DummyTransport;
/// # use libp2p_swarm::DummyBehaviour;
/// #
/// let mut swarm = Swarm::new(
/// DummyTransport::new().boxed(),
/// DummyBehaviour::default(),
/// PeerId::random(),
/// );
///
/// // Dial a known peer.
/// swarm.dial(PeerId::random());
///
/// // Dial an unknown peer.
/// swarm.dial("/ip6/::1/tcp/12345".parse::<Multiaddr>().unwrap());
/// ```
pub fn dial(&mut self, opts: impl Into<DialOpts>) -> Result<(), DialError> {
let handler = self.behaviour.new_handler();
self.dial_addr_with_handler(addr, handler)
.map_err(DialError::from_network_dial_error)
.map_err(|(e, _)| e)
}
fn dial_addr_with_handler(
&mut self,
addr: Multiaddr,
handler: <TBehaviour as NetworkBehaviour>::ProtocolsHandler,
) -> Result<(), network::DialError<NodeHandlerWrapperBuilder<THandler<TBehaviour>>>> {
let handler = handler
.into_node_handler_builder()
.with_substream_upgrade_protocol_override(self.substream_upgrade_protocol_override);
self.network.dial(&addr, handler).map(|_id| ())
}
/// Initiates a new dialing attempt to the given peer.
pub fn dial(&mut self, peer_id: &PeerId) -> Result<(), DialError> {
let handler = self.behaviour.new_handler();
self.dial_with_handler(peer_id, handler)
self.dial_with_handler(opts.into(), handler)
}
fn dial_with_handler(
&mut self,
peer_id: &PeerId,
opts: DialOpts,
handler: <TBehaviour as NetworkBehaviour>::ProtocolsHandler,
) -> Result<(), DialError> {
if self.banned_peers.contains(peer_id) {
match opts.0 {
// Dial a known peer.
dial_opts::Opts::WithPeerId(dial_opts::WithPeerId { peer_id, condition })
| dial_opts::Opts::WithPeerIdWithAddresses(dial_opts::WithPeerIdWithAddresses {
peer_id,
condition,
..
}) => {
// Check [`PeerCondition`] if provided.
let condition_matched = match condition {
PeerCondition::Disconnected => self.network.is_disconnected(&peer_id),
PeerCondition::NotDialing => !self.network.is_dialing(&peer_id),
PeerCondition::Always => true,
};
if !condition_matched {
self.behaviour.inject_dial_failure(
Some(peer_id),
handler,
&DialError::DialPeerConditionFalse(condition),
);
return Err(DialError::DialPeerConditionFalse(condition));
}
// Check if peer is banned.
if self.banned_peers.contains(&peer_id) {
let error = DialError::Banned;
self.behaviour
.inject_dial_failure(Some(*peer_id), handler, &error);
.inject_dial_failure(Some(peer_id), handler, &error);
return Err(error);
}
let self_listening = self.listened_addrs.clone();
let mut addrs = self
.behaviour
.addresses_of_peer(peer_id)
.into_iter()
.filter(move |a| !self_listening.contains(a))
.peekable();
// Retrieve the addresses to dial.
let addresses = {
let mut addresses = match opts.0 {
dial_opts::Opts::WithPeerId(dial_opts::WithPeerId { .. }) => {
self.behaviour.addresses_of_peer(&peer_id)
}
dial_opts::Opts::WithPeerIdWithAddresses(
dial_opts::WithPeerIdWithAddresses {
peer_id,
mut addresses,
extend_addresses_through_behaviour,
..
},
) => {
if extend_addresses_through_behaviour {
addresses.extend(self.behaviour.addresses_of_peer(&peer_id))
}
addresses
}
dial_opts::Opts::WithoutPeerIdWithAddress { .. } => {
unreachable!("Due to outer match.")
}
};
if addrs.peek().is_none() {
let mut unique_addresses = HashSet::new();
addresses.retain(|a| {
!self.listened_addrs.contains(a) && unique_addresses.insert(a.clone())
});
if addresses.is_empty() {
let error = DialError::NoAddresses;
self.behaviour
.inject_dial_failure(Some(*peer_id), handler, &error);
.inject_dial_failure(Some(peer_id), handler, &error);
return Err(error);
};
addresses
};
let handler = handler
.into_node_handler_builder()
.with_substream_upgrade_protocol_override(self.substream_upgrade_protocol_override);
match self.network.peer(*peer_id).dial(addrs, handler) {
.with_substream_upgrade_protocol_override(
self.substream_upgrade_protocol_override,
);
match self.network.peer(peer_id).dial(addresses, handler) {
Ok(_connection_id) => Ok(()),
Err(error) => {
let (error, handler) = DialError::from_network_dial_error(error);
self.behaviour.inject_dial_failure(
Some(*peer_id),
Some(peer_id),
handler.into_protocols_handler(),
&error,
);
Err(error)
return Err(error);
}
}
}
// Dial an unknown peer.
dial_opts::Opts::WithoutPeerIdWithAddress(dial_opts::WithoutPeerIdWithAddress {
address,
}) => {
let handler = handler
.into_node_handler_builder()
.with_substream_upgrade_protocol_override(
self.substream_upgrade_protocol_override,
);
match self.network.dial(&address, handler).map(|_id| ()) {
Ok(_connection_id) => Ok(()),
Err(error) => {
let (error, handler) = DialError::from_network_dial_error(error);
self.behaviour.inject_dial_failure(
None,
handler.into_protocols_handler(),
&error,
);
return Err(error);
}
}
}
}
}
@ -808,35 +889,12 @@ where
Poll::Ready(NetworkBehaviourAction::GenerateEvent(event)) => {
return Poll::Ready(SwarmEvent::Behaviour(event))
}
Poll::Ready(NetworkBehaviourAction::DialAddress { address, handler }) => {
let _ = Swarm::dial_addr_with_handler(&mut *this, address, handler);
}
Poll::Ready(NetworkBehaviourAction::DialPeer {
peer_id,
condition,
handler,
}) => {
let condition_matched = match condition {
DialPeerCondition::Disconnected => this.network.is_disconnected(&peer_id),
DialPeerCondition::NotDialing => !this.network.is_dialing(&peer_id),
DialPeerCondition::Always => true,
};
if condition_matched {
if Swarm::dial_with_handler(this, &peer_id, handler).is_ok() {
Poll::Ready(NetworkBehaviourAction::Dial { opts, handler }) => {
let peer_id = opts.get_peer_id();
if let Ok(()) = this.dial_with_handler(opts, handler) {
if let Some(peer_id) = peer_id {
return Poll::Ready(SwarmEvent::Dialing(peer_id));
}
} else {
log::trace!(
"Condition for new dialing attempt to {:?} not met: {:?}",
peer_id,
condition
);
this.behaviour.inject_dial_failure(
Some(peer_id),
handler,
&DialError::DialPeerConditionFalse(condition),
);
}
}
Poll::Ready(NetworkBehaviourAction::NotifyHandler {
@ -1214,8 +1272,9 @@ pub enum DialError {
/// [`NetworkBehaviour::addresses_of_peer`] returned no addresses
/// for the peer to dial.
NoAddresses,
/// The provided [`DialPeerCondition`] evaluated to false and thus the dial was aborted.
DialPeerConditionFalse(DialPeerCondition),
/// The provided [`dial_opts::PeerCondition`] evaluated to false and thus
/// the dial was aborted.
DialPeerConditionFalse(dial_opts::PeerCondition),
/// Pending connection attempt has been aborted.
Aborted,
/// The peer identity obtained on the connection did not
@ -1457,7 +1516,7 @@ mod tests {
let num_connections = 10;
for _ in 0..num_connections {
swarm1.dial_addr(addr2.clone()).unwrap();
swarm1.dial(addr2.clone()).unwrap();
}
let mut state = State::Connecting;
@ -1489,7 +1548,7 @@ mod tests {
swarm2.behaviour.reset();
unbanned = true;
for _ in 0..num_connections {
swarm2.dial_addr(addr1.clone()).unwrap();
swarm2.dial(addr1.clone()).unwrap();
}
state = State::Connecting;
}
@ -1532,7 +1591,7 @@ mod tests {
let num_connections = 10;
for _ in 0..num_connections {
swarm1.dial_addr(addr2.clone()).unwrap();
swarm1.dial(addr2.clone()).unwrap();
}
let mut state = State::Connecting;
@ -1562,7 +1621,7 @@ mod tests {
swarm1.behaviour.reset();
swarm2.behaviour.reset();
for _ in 0..num_connections {
swarm2.dial_addr(addr1.clone()).unwrap();
swarm2.dial(addr1.clone()).unwrap();
}
state = State::Connecting;
}
@ -1605,7 +1664,7 @@ mod tests {
let num_connections = 10;
for _ in 0..num_connections {
swarm1.dial_addr(addr2.clone()).unwrap();
swarm1.dial(addr2.clone()).unwrap();
}
let mut state = State::Connecting;
@ -1638,7 +1697,7 @@ mod tests {
swarm1.behaviour.reset();
swarm2.behaviour.reset();
for _ in 0..num_connections {
swarm2.dial_addr(addr1.clone()).unwrap();
swarm2.dial(addr1.clone()).unwrap();
}
state = State::Connecting;
}
@ -1680,7 +1739,7 @@ mod tests {
let num_connections = 10;
for _ in 0..num_connections {
swarm1.dial_addr(addr2.clone()).unwrap();
swarm1.dial(addr2.clone()).unwrap();
}
let mut state = State::Connecting;
let mut disconnected_conn_id = None;