fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-07-31 08:51:57 +00:00
Code Issues Projects Releases Wiki Activity

mdns: update if-watch to 3.0.0 (#3096)

Browse Source
This commit is contained in:
João Oliveira
2022-11-18 01:12:23 +00:00
committed by GitHub
parent 7803524a76
commit 08510dd523
12 changed files with 173 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
protocols/mdns/CHANGELOG.md
View File

@@ -4,11 +4,21 @@
- Update to `libp2p-swarm` `v0.41.0`.
- Update to `if-watch` `3.0.0` and both rename `TokioMdns` to `Behaviour` living in `tokio::Behaviour`,
and move and rename `Mdns` to `async_io::Behaviour`. See [PR 3096].
- Remove the remaning `Mdns` prefixes from types as per [discussion 2174].
I.e the `Mdns` prefix has been removed from various types like `MdnsEvent`.
Users should prefer importing the mdns protocol as a module (`use libp2p::mdns;`),
and refer to its types via `mdns::`. For example: `mdns::Behaviour` or `mdns::Event`.
- Replace `GenMdns`'s `NetworkBehaviour` implemention `inject_*` methods with the new `on_*` methods.
See [PR 3011].
- Use `trust-dns-proto` to parse DNS messages. See [PR 3102].
[discussion 2174]: https://github.com/libp2p/rust-libp2p/discussions/2174
[PR 3096]: https://github.com/libp2p/rust-libp2p/pull/3096
[PR 3011]: https://github.com/libp2p/rust-libp2p/pull/3011
[PR 3102]: https://github.com/libp2p/rust-libp2p/pull/3102

6
protocols/mdns/Cargo.toml
View File

@@ -14,7 +14,7 @@ categories = ["network-programming", "asynchronous"]
async-io = { version = "1.3.1", optional = true }
data-encoding = "2.3.2"
futures = "0.3.13"
if-watch = "2.0.0"
if-watch = "3.0.0"
libp2p-core = { version = "0.38.0", path = "../../core" }
libp2p-swarm = { version = "0.41.0", path = "../../swarm" }
log = "0.4.14"
@@ -26,8 +26,8 @@ trust-dns-proto = { version = "0.22.0", default-features = false, features = ["m
void = "1.0.2"
[features]
tokio = ["dep:tokio"]
async-io = ["dep:async-io"]
tokio = ["dep:tokio", "if-watch/tokio"]
async-io = ["dep:async-io", "if-watch/smol"]
[dev-dependencies]
async-std = { version = "1.9.0", features = ["attributes"] }

104
protocols/mdns/src/behaviour.rs
View File

@@ -24,9 +24,9 @@ mod timer;
use self::iface::InterfaceState;
use crate::behaviour::{socket::AsyncSocket, timer::Builder};
use crate::MdnsConfig;
use crate::Config;
use futures::Stream;
use if_watch::{IfEvent, IfWatcher};
use if_watch::IfEvent;
use libp2p_core::{Multiaddr, PeerId};
use libp2p_swarm::behaviour::{ConnectionClosed, FromSwarm};
use libp2p_swarm::{
@@ -36,32 +36,80 @@ use smallvec::SmallVec;
use std::collections::hash_map::{Entry, HashMap};
use std::{cmp, fmt, io, net::IpAddr, pin::Pin, task::Context, task::Poll, time::Instant};
/// An abstraction to allow for compatibility with various async runtimes.
pub trait Provider: 'static {
/// The Async Socket type.
type Socket: AsyncSocket;
/// The Async Timer type.
type Timer: Builder + Stream;
/// The IfWatcher type.
type Watcher: Stream<Item = std::io::Result<IfEvent>> + fmt::Debug + Unpin;
/// Create a new instance of the `IfWatcher` type.
fn new_watcher() -> Result<Self::Watcher, std::io::Error>;
}
/// The type of a [`Behaviour`] using the `async-io` implementation.
#[cfg(feature = "async-io")]
use crate::behaviour::{socket::asio::AsyncUdpSocket, timer::asio::AsyncTimer};
pub mod async_io {
use super::Provider;
use crate::behaviour::{socket::asio::AsyncUdpSocket, timer::asio::AsyncTimer};
use if_watch::smol::IfWatcher;
/// The type of a [`GenMdns`] using the `async-io` implementation.
#[cfg(feature = "async-io")]
pub type Mdns = GenMdns<AsyncUdpSocket, AsyncTimer>;
#[doc(hidden)]
pub enum AsyncIo {}
impl Provider for AsyncIo {
type Socket = AsyncUdpSocket;
type Timer = AsyncTimer;
type Watcher = IfWatcher;
fn new_watcher() -> Result<Self::Watcher, std::io::Error> {
IfWatcher::new()
}
}
pub type Behaviour = super::Behaviour<AsyncIo>;
}
/// The type of a [`Behaviour`] using the `tokio` implementation.
#[cfg(feature = "tokio")]
use crate::behaviour::{socket::tokio::TokioUdpSocket, timer::tokio::TokioTimer};
pub mod tokio {
use super::Provider;
use crate::behaviour::{socket::tokio::TokioUdpSocket, timer::tokio::TokioTimer};
use if_watch::tokio::IfWatcher;
/// The type of a [`GenMdns`] using the `tokio` implementation.
#[cfg(feature = "tokio")]
pub type TokioMdns = GenMdns<TokioUdpSocket, TokioTimer>;
#[doc(hidden)]
pub enum Tokio {}
impl Provider for Tokio {
type Socket = TokioUdpSocket;
type Timer = TokioTimer;
type Watcher = IfWatcher;
fn new_watcher() -> Result<Self::Watcher, std::io::Error> {
IfWatcher::new()
}
}
pub type Behaviour = super::Behaviour<Tokio>;
}
/// A `NetworkBehaviour` for mDNS. Automatically discovers peers on the local network and adds
/// them to the topology.
#[derive(Debug)]
pub struct GenMdns<S, T> {
pub struct Behaviour<P>
where
P: Provider,
{
/// InterfaceState config.
config: MdnsConfig,
config: Config,
/// Iface watcher.
if_watch: IfWatcher,
if_watch: P::Watcher,
/// Mdns interface states.
iface_states: HashMap<IpAddr, InterfaceState<S, T>>,
iface_states: HashMap<IpAddr, InterfaceState<P::Socket, P::Timer>>,
/// List of nodes that we have discovered, the address, and when their TTL expires.
///
@@ -72,19 +120,18 @@ pub struct GenMdns<S, T> {
/// Future that fires when the TTL of at least one node in `discovered_nodes` expires.
///
/// `None` if `discovered_nodes` is empty.
closest_expiration: Option<T>,
closest_expiration: Option<P::Timer>,
}
impl<S, T> GenMdns<S, T>
impl<P> Behaviour<P>
where
T: Builder,
P: Provider,
{
/// Builds a new `Mdns` behaviour.
pub fn new(config: MdnsConfig) -> io::Result<Self> {
let if_watch = if_watch::IfWatcher::new()?;
pub fn new(config: Config) -> io::Result<Self> {
Ok(Self {
config,
if_watch,
if_watch: P::new_watcher()?,
iface_states: Default::default(),
discovered_nodes: Default::default(),
closest_expiration: Default::default(),
@@ -109,17 +156,16 @@ where
*expires = now;
}
}
self.closest_expiration = Some(T::at(now));
self.closest_expiration = Some(P::Timer::at(now));
}
}
impl<S, T> NetworkBehaviour for GenMdns<S, T>
impl<P> NetworkBehaviour for Behaviour<P>
where
T: Builder + Stream,
S: AsyncSocket,
P: Provider,
{
type ConnectionHandler = dummy::ConnectionHandler;
type OutEvent = MdnsEvent;
type OutEvent = Event;
fn new_handler(&mut self) -> Self::ConnectionHandler {
dummy::ConnectionHandler
@@ -226,7 +272,7 @@ where
}
}
if !discovered.is_empty() {
let event = MdnsEvent::Discovered(DiscoveredAddrsIter {
let event = Event::Discovered(DiscoveredAddrsIter {
inner: discovered.into_iter(),
});
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
@@ -245,13 +291,13 @@ where
true
});
if !expired.is_empty() {
let event = MdnsEvent::Expired(ExpiredAddrsIter {
let event = Event::Expired(ExpiredAddrsIter {
inner: expired.into_iter(),
});
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
}
if let Some(closest_expiration) = closest_expiration {
let mut timer = T::at(closest_expiration);
let mut timer = P::Timer::at(closest_expiration);
let _ = Pin::new(&mut timer).poll_next(cx);
self.closest_expiration = Some(timer);
@@ -262,7 +308,7 @@ where
/// Event that can be produced by the `Mdns` behaviour.
#[derive(Debug)]
pub enum MdnsEvent {
pub enum Event {
/// Discovered nodes through mDNS.
Discovered(DiscoveredAddrsIter),

4
protocols/mdns/src/behaviour/iface.rs
View File

@@ -24,7 +24,7 @@ mod query;
use self::dns::{build_query, build_query_response, build_service_discovery_response};
use self::query::MdnsPacket;
use crate::behaviour::{socket::AsyncSocket, timer::Builder};
use crate::MdnsConfig;
use crate::Config;
use libp2p_core::{Multiaddr, PeerId};
use libp2p_swarm::PollParameters;
use socket2::{Domain, Socket, Type};
@@ -74,7 +74,7 @@ where
T: Builder + futures::Stream,
{
/// Builds a new [`InterfaceState`].
pub fn new(addr: IpAddr, config: MdnsConfig) -> io::Result<Self> {
pub fn new(addr: IpAddr, config: Config) -> io::Result<Self> {
log::info!("creating instance on iface {}", addr);
let recv_socket = match addr {
IpAddr::V4(addr) => {

29
protocols/mdns/src/lib.rs
View File

@@ -38,14 +38,33 @@
use std::net::{Ipv4Addr, Ipv6Addr};
use std::time::Duration;
#[deprecated(
since = "0.42.0",
note = "Use re-exports that omit `Mdns` prefix, i.e. `libp2p::mdns::Config`"
)]
pub type MdnsConfig = Config;
#[deprecated(
since = "0.42.0",
note = "Use re-exports that omit `Mdns` prefix, i.e. `libp2p::mdns::Event`"
)]
pub type MdnsEvent = Event;
#[deprecated(
since = "0.42.0",
note = "Use the async-io prefixed `Mdns`, i.e. `libp2p::mdns::async_io::Mdns`"
)]
#[cfg(feature = "async-io")]
pub type Mdns = async_io::Behaviour;
mod behaviour;
pub use crate::behaviour::{GenMdns, MdnsEvent};
pub use crate::behaviour::{Behaviour, Event};
#[cfg(feature = "async-io")]
pub use crate::behaviour::Mdns;
pub use crate::behaviour::async_io;
#[cfg(feature = "tokio")]
pub use crate::behaviour::TokioMdns;
pub use crate::behaviour::tokio;
/// The DNS service name for all libp2p peers used to query for addresses.
const SERVICE_NAME: &[u8] = b"_p2p._udp.local";
@@ -61,7 +80,7 @@ pub const IPV6_MDNS_MULTICAST_ADDRESS: Ipv6Addr = Ipv6Addr::new(0xFF02, 0, 0, 0,
/// Configuration for mDNS.
#[derive(Debug, Clone)]
pub struct MdnsConfig {
pub struct Config {
/// TTL to use for mdns records.
pub ttl: Duration,
/// Interval at which to poll the network for new peers. This isn't
@@ -74,7 +93,7 @@ pub struct MdnsConfig {
pub enable_ipv6: bool,
}
impl Default for MdnsConfig {
impl Default for Config {
fn default() -> Self {
Self {
ttl: Duration::from_secs(6 * 60),

24
protocols/mdns/tests/use-async-std.rs
View File

@@ -21,7 +21,7 @@
use futures::StreamExt;
use libp2p::{
identity,
mdns::{Mdns, MdnsConfig, MdnsEvent},
mdns::{async_io::Behaviour, Config, Event},
swarm::{Swarm, SwarmEvent},
PeerId,
};
@@ -30,12 +30,12 @@ use std::time::Duration;
#[async_std::test]
async fn test_discovery_async_std_ipv4() -> Result<(), Box<dyn Error>> {
run_discovery_test(MdnsConfig::default()).await
run_discovery_test(Config::default()).await
}
#[async_std::test]
async fn test_discovery_async_std_ipv6() -> Result<(), Box<dyn Error>> {
let config = MdnsConfig {
let config = Config {
enable_ipv6: true,
..Default::default()
};
@@ -45,7 +45,7 @@ async fn test_discovery_async_std_ipv6() -> Result<(), Box<dyn Error>> {
#[async_std::test]
async fn test_expired_async_std() -> Result<(), Box<dyn Error>> {
env_logger::try_init().ok();
let config = MdnsConfig {
let config = Config {
ttl: Duration::from_secs(1),
query_interval: Duration::from_secs(10),
..Default::default()
@@ -57,17 +57,17 @@ async fn test_expired_async_std() -> Result<(), Box<dyn Error>> {
.map_err(|e| Box::new(e) as Box<dyn Error>)
}
async fn create_swarm(config: MdnsConfig) -> Result<Swarm<Mdns>, Box<dyn Error>> {
async fn create_swarm(config: Config) -> Result<Swarm<Behaviour>, Box<dyn Error>> {
let id_keys = identity::Keypair::generate_ed25519();
let peer_id = PeerId::from(id_keys.public());
let transport = libp2p::development_transport(id_keys).await?;
let behaviour = Mdns::new(config)?;
let behaviour = Behaviour::new(config)?;
let mut swarm = Swarm::with_async_std_executor(transport, behaviour, peer_id);
swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
Ok(swarm)
}
async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
async fn run_discovery_test(config: Config) -> Result<(), Box<dyn Error>> {
env_logger::try_init().ok();
let mut a = create_swarm(config.clone()).await?;
let mut b = create_swarm(config).await?;
@@ -75,7 +75,7 @@ async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
let mut discovered_b = false;
loop {
futures::select! {
ev = a.select_next_some() => if let SwarmEvent::Behaviour(MdnsEvent::Discovered(peers)) = ev {
ev = a.select_next_some() => if let SwarmEvent::Behaviour(Event::Discovered(peers)) = ev {
for (peer, _addr) in peers {
if peer == *b.local_peer_id() {
if discovered_a {
@@ -86,7 +86,7 @@ async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
}
}
},
ev = b.select_next_some() => if let SwarmEvent::Behaviour(MdnsEvent::Discovered(peers)) = ev {
ev = b.select_next_some() => if let SwarmEvent::Behaviour(Event::Discovered(peers)) = ev {
for (peer, _addr) in peers {
if peer == *a.local_peer_id() {
if discovered_b {
@@ -101,20 +101,20 @@ async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
}
}
async fn run_peer_expiration_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
async fn run_peer_expiration_test(config: Config) -> Result<(), Box<dyn Error>> {
let mut a = create_swarm(config.clone()).await?;
let mut b = create_swarm(config).await?;
loop {
futures::select! {
ev = a.select_next_some() => if let SwarmEvent::Behaviour(MdnsEvent::Expired(peers)) = ev {
ev = a.select_next_some() => if let SwarmEvent::Behaviour(Event::Expired(peers)) = ev {
for (peer, _addr) in peers {
if peer == *b.local_peer_id() {
return Ok(());
}
}
},
ev = b.select_next_some() => if let SwarmEvent::Behaviour(MdnsEvent::Expired(peers)) = ev {
ev = b.select_next_some() => if let SwarmEvent::Behaviour(Event::Expired(peers)) = ev {
for (peer, _addr) in peers {
if peer == *a.local_peer_id() {
return Ok(());

28
protocols/mdns/tests/use-tokio.rs
View File

@@ -20,7 +20,7 @@
use futures::StreamExt;
use libp2p::{
identity,
mdns::{MdnsConfig, MdnsEvent, TokioMdns},
mdns::{tokio::Behaviour, Config, Event},
swarm::{Swarm, SwarmEvent},
PeerId,
};
@@ -29,12 +29,12 @@ use std::time::Duration;
#[tokio::test]
async fn test_discovery_tokio_ipv4() -> Result<(), Box<dyn Error>> {
run_discovery_test(MdnsConfig::default()).await
run_discovery_test(Config::default()).await
}
#[tokio::test]
async fn test_discovery_tokio_ipv6() -> Result<(), Box<dyn Error>> {
let config = MdnsConfig {
let config = Config {
enable_ipv6: true,
..Default::default()
};
@@ -44,7 +44,7 @@ async fn test_discovery_tokio_ipv6() -> Result<(), Box<dyn Error>> {
#[tokio::test]
async fn test_expired_tokio() -> Result<(), Box<dyn Error>> {
env_logger::try_init().ok();
let config = MdnsConfig {
let config = Config {
ttl: Duration::from_secs(1),
query_interval: Duration::from_secs(10),
..Default::default()
@@ -53,17 +53,17 @@ async fn test_expired_tokio() -> Result<(), Box<dyn Error>> {
run_peer_expiration_test(config).await
}
async fn create_swarm(config: MdnsConfig) -> Result<Swarm<TokioMdns>, Box<dyn Error>> {
async fn create_swarm(config: Config) -> Result<Swarm<Behaviour>, Box<dyn Error>> {
let id_keys = identity::Keypair::generate_ed25519();
let peer_id = PeerId::from(id_keys.public());
let transport = libp2p::tokio_development_transport(id_keys)?;
let behaviour = TokioMdns::new(config)?;
let behaviour = Behaviour::new(config)?;
let mut swarm = Swarm::with_tokio_executor(transport, behaviour, peer_id);
swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
Ok(swarm)
}
async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
async fn run_discovery_test(config: Config) -> Result<(), Box<dyn Error>> {
env_logger::try_init().ok();
let mut a = create_swarm(config.clone()).await?;
let mut b = create_swarm(config).await?;
@@ -71,7 +71,7 @@ async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
let mut discovered_b = false;
loop {
futures::select! {
ev = a.select_next_some() => if let SwarmEvent::Behaviour(MdnsEvent::Discovered(peers)) = ev {
ev = a.select_next_some() => if let SwarmEvent::Behaviour(Event::Discovered(peers)) = ev {
for (peer, _addr) in peers {
if peer == *b.local_peer_id() {
if discovered_a {
@@ -82,7 +82,7 @@ async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
}
}
},
ev = b.select_next_some() => if let SwarmEvent::Behaviour(MdnsEvent::Discovered(peers)) = ev {
ev = b.select_next_some() => if let SwarmEvent::Behaviour(Event::Discovered(peers)) = ev {
for (peer, _addr) in peers {
if peer == *a.local_peer_id() {
if discovered_b {
@@ -97,7 +97,7 @@ async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
}
}
async fn run_peer_expiration_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
async fn run_peer_expiration_test(config: Config) -> Result<(), Box<dyn Error>> {
let mut a = create_swarm(config.clone()).await?;
let mut b = create_swarm(config).await?;
let expired_at = tokio::time::sleep(Duration::from_secs(15));
@@ -109,14 +109,14 @@ async fn run_peer_expiration_test(config: MdnsConfig) -> Result<(), Box<dyn Erro
panic!();
},
ev = a.select_next_some() => match ev {
SwarmEvent::Behaviour(MdnsEvent::Expired(peers)) => {
SwarmEvent::Behaviour(Event::Expired(peers)) => {
for (peer, _addr) in peers {
if peer == *b.local_peer_id() {
return Ok(());
}
}
}
SwarmEvent::Behaviour(MdnsEvent::Discovered(peers)) => {
SwarmEvent::Behaviour(Event::Discovered(peers)) => {
for (peer, _addr) in peers {
if peer == *b.local_peer_id() {
expired_at.as_mut().reset(tokio::time::Instant::now() + tokio::time::Duration::from_secs(2));
@@ -126,14 +126,14 @@ async fn run_peer_expiration_test(config: MdnsConfig) -> Result<(), Box<dyn Erro
_ => {}
},
ev = b.select_next_some() => match ev {
SwarmEvent::Behaviour(MdnsEvent::Expired(peers)) => {
SwarmEvent::Behaviour(Event::Expired(peers)) => {
for (peer, _addr) in peers {
if peer == *a.local_peer_id() {
return Ok(());
}
}
}
SwarmEvent::Behaviour(MdnsEvent::Discovered(peers)) => {
SwarmEvent::Behaviour(Event::Discovered(peers)) => {
for (peer, _addr) in peers {
if peer == *a.local_peer_id() {
expired_at.as_mut().reset(tokio::time::Instant::now() + tokio::time::Duration::from_secs(2));