use crate::service::{MdnsPacket, build_query_response, build_service_discovery_response};
use futures::prelude::*;
use libp2p_core::{
Multiaddr,
PeerId,
address_translation,
connection::ConnectionId,
multiaddr::Protocol
};
use libp2p_swarm::{
NetworkBehaviour,
NetworkBehaviourAction,
PollParameters,
ProtocolsHandler,
protocols_handler::DummyProtocolsHandler
};
use log::warn;
use smallvec::SmallVec;
use std::{cmp, fmt, io, iter, mem, pin::Pin, time::Duration, task::Context, task::Poll};
use wasm_timer::{Delay, Instant};
const MDNS_RESPONSE_TTL: std::time::Duration = Duration::from_secs(5 * 60);
macro_rules! codegen {
($feature_name:expr, $behaviour_name:ident, $maybe_busy_wrapper:ident, $service_name:ty) => {
pub struct $behaviour_name {
service: $maybe_busy_wrapper,
discovered_nodes: SmallVec<[(PeerId, Multiaddr, Instant); 8]>,
closest_expiration: Option<Delay>,
}
enum $maybe_busy_wrapper {
Free($service_name),
Busy(Pin<Box<dyn Future<Output = ($service_name, MdnsPacket)> + Send>>),
Poisoned,
}
impl fmt::Debug for $maybe_busy_wrapper {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
$maybe_busy_wrapper::Free(service) => {
fmt.debug_struct("$maybe_busy_wrapper::Free")
.field("service", service)
.finish()
},
$maybe_busy_wrapper::Busy(_) => {
fmt.debug_struct("$maybe_busy_wrapper::Busy")
.finish()
}
$maybe_busy_wrapper::Poisoned => {
fmt.debug_struct("$maybe_busy_wrapper::Poisoned")
.finish()
}
}
}
}
impl $behaviour_name {
pub fn new() -> io::Result<$behaviour_name> {
Ok($behaviour_name {
service: $maybe_busy_wrapper::Free(<$service_name>::new()?),
discovered_nodes: SmallVec::new(),
closest_expiration: None,
})
}
pub fn has_node(&self, peer_id: &PeerId) -> bool {
self.discovered_nodes().any(|p| p == peer_id)
}
pub fn discovered_nodes(&self) -> impl ExactSizeIterator<Item = &PeerId> {
self.discovered_nodes.iter().map(|(p, _, _)| p)
}
}
impl NetworkBehaviour for $behaviour_name {
type ProtocolsHandler = DummyProtocolsHandler;
type OutEvent = MdnsEvent;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
DummyProtocolsHandler::default()
}
fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
let now = Instant::now();
self.discovered_nodes
.iter()
.filter(move |(p, _, expires)| p == peer_id && *expires > now)
.map(|(_, addr, _)| addr.clone())
.collect()
}
fn inject_connected(&mut self, _: &PeerId) {}
fn inject_disconnected(&mut self, _: &PeerId) {}
fn inject_event(
&mut self,
_: PeerId,
_: ConnectionId,
_ev: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
) {
void::unreachable(_ev)
}
fn poll(
&mut self,
cx: &mut Context<'_>,
params: &mut impl PollParameters,
) -> Poll<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
if let Some(ref mut closest_expiration) = self.closest_expiration {
match Future::poll(Pin::new(closest_expiration), cx) {
Poll::Ready(Ok(())) => {
let now = Instant::now();
let mut expired = SmallVec::<[(PeerId, Multiaddr); 4]>::new();
while let Some(pos) = self.discovered_nodes.iter().position(|(_, _, exp)| *exp < now) {
let (peer_id, addr, _) = self.discovered_nodes.remove(pos);
expired.push((peer_id, addr));
}
if !expired.is_empty() {
let event = MdnsEvent::Expired(ExpiredAddrsIter {
inner: expired.into_iter(),
});
return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
}
},
Poll::Pending => (),
Poll::Ready(Err(err)) => warn!("timer has errored: {:?}", err),
}
}
let discovered = loop {
let service = mem::replace(&mut self.service, $maybe_busy_wrapper::Poisoned);
let packet = match service {
$maybe_busy_wrapper::Free(service) => {
self.service = $maybe_busy_wrapper::Busy(Box::pin(service.next()));
continue;
},
$maybe_busy_wrapper::Busy(mut fut) => {
match fut.as_mut().poll(cx) {
Poll::Ready((service, packet)) => {
self.service = $maybe_busy_wrapper::Free(service);
packet
},
Poll::Pending => {
self.service = $maybe_busy_wrapper::Busy(fut);
return Poll::Pending;
}
}
},
$maybe_busy_wrapper::Poisoned => panic!("Mdns poisoned"),
};
match packet {
MdnsPacket::Query(query) => {
if let $maybe_busy_wrapper::Free(ref mut service) = self.service {
let resp = build_query_response(
query.query_id(),
params.local_peer_id().clone(),
params.listened_addresses().into_iter(),
MDNS_RESPONSE_TTL,
);
service.enqueue_response(resp.unwrap());
} else { debug_assert!(false); }
},
MdnsPacket::Response(response) => {
let obs_ip = Protocol::from(response.remote_addr().ip());
let obs_port = Protocol::Udp(response.remote_addr().port());
let observed: Multiaddr = iter::once(obs_ip)
.chain(iter::once(obs_port))
.collect();
let mut discovered: SmallVec<[_; 4]> = SmallVec::new();
for peer in response.discovered_peers() {
if peer.id() == params.local_peer_id() {
continue;
}
let new_expiration = Instant::now() + peer.ttl();
let mut addrs: Vec<Multiaddr> = Vec::new();
for addr in peer.addresses() {
if let Some(new_addr) = address_translation(&addr, &observed) {
addrs.push(new_addr.clone())
}
addrs.push(addr.clone())
}
for addr in addrs {
if let Some((_, _, cur_expires)) = self.discovered_nodes.iter_mut()
.find(|(p, a, _)| p == peer.id() && *a == addr)
{
*cur_expires = cmp::max(*cur_expires, new_expiration);
} else {
self.discovered_nodes.push((peer.id().clone(), addr.clone(), new_expiration));
}
discovered.push((peer.id().clone(), addr));
}
}
break discovered;
},
MdnsPacket::ServiceDiscovery(disc) => {
if let $maybe_busy_wrapper::Free(ref mut service) = self.service {
let resp = build_service_discovery_response(
disc.query_id(),
MDNS_RESPONSE_TTL,
);
service.enqueue_response(resp);
} else { debug_assert!(false); }
},
}
};
self.closest_expiration = self.discovered_nodes.iter()
.fold(None, |exp, &(_, _, elem_exp)| {
Some(exp.map(|exp| cmp::min(exp, elem_exp)).unwrap_or(elem_exp))
})
.map(Delay::new_at);
Poll::Ready(NetworkBehaviourAction::GenerateEvent(MdnsEvent::Discovered(DiscoveredAddrsIter {
inner: discovered.into_iter(),
})))
}
}
impl fmt::Debug for $behaviour_name {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Mdns")
.field("service", &self.service)
.finish()
}
}
};
}
#[cfg(feature = "async-std")]
codegen!("async-std", Mdns, MaybeBusyMdnsService, crate::service::MdnsService);
#[cfg(feature = "tokio")]
codegen!("tokio", TokioMdns, MaybeBusyTokioMdnsService, crate::service::TokioMdnsService);
#[derive(Debug)]
pub enum MdnsEvent {
Discovered(DiscoveredAddrsIter),
Expired(ExpiredAddrsIter),
}
pub struct DiscoveredAddrsIter {
inner: smallvec::IntoIter<[(PeerId, Multiaddr); 4]>
}
impl Iterator for DiscoveredAddrsIter {
type Item = (PeerId, Multiaddr);
#[inline]
fn next(&mut self) -> Option<Self::Item> {
self.inner.next()
}
#[inline]
fn size_hint(&self) -> (usize, Option<usize>) {
self.inner.size_hint()
}
}
impl ExactSizeIterator for DiscoveredAddrsIter {
}
impl fmt::Debug for DiscoveredAddrsIter {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("DiscoveredAddrsIter")
.finish()
}
}
pub struct ExpiredAddrsIter {
inner: smallvec::IntoIter<[(PeerId, Multiaddr); 4]>
}
impl Iterator for ExpiredAddrsIter {
type Item = (PeerId, Multiaddr);
#[inline]
fn next(&mut self) -> Option<Self::Item> {
self.inner.next()
}
#[inline]
fn size_hint(&self) -> (usize, Option<usize>) {
self.inner.size_hint()
}
}
impl ExactSizeIterator for ExpiredAddrsIter {
}
impl fmt::Debug for ExpiredAddrsIter {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("ExpiredAddrsIter")
.finish()
}
}