Move mdns (#1458)

* move mdns

* alphabetic ordering in Cargo.toml

Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>

protocols/mdns/src/behaviour.rs

@@ -0,0 +1,346 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+//
+// 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 crate::service::{MdnsService, MdnsPacket, build_query_response, build_service_discovery_response};
+use futures::prelude::*;
+use libp2p_core::{address_translation, ConnectedPoint, Multiaddr, PeerId, 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);
+
+/// A `NetworkBehaviour` for mDNS. Automatically discovers peers on the local network and adds
+/// them to the topology.
+pub struct Mdns {
+    /// The inner service.
+    service: MaybeBusyMdnsService,
+
+    /// List of nodes that we have discovered, the address, and when their TTL expires.
+    ///
+    /// Each combination of `PeerId` and `Multiaddr` can only appear once, but the same `PeerId`
+    /// can appear multiple times.
+    discovered_nodes: SmallVec<[(PeerId, Multiaddr, Instant); 8]>,
+
+    /// Future that fires when the TTL of at least one node in `discovered_nodes` expires.
+    ///
+    /// `None` if `discovered_nodes` is empty.
+    closest_expiration: Option<Delay>,
+}
+
+/// `MdnsService::next` takes ownership of `self`, returning a future that resolves with both itself
+/// and a `MdnsPacket` (similar to the old Tokio socket send style). The two states are thus `Free`
+/// with an `MdnsService` or `Busy` with a future returning the original `MdnsService` and an
+/// `MdnsPacket`.
+enum MaybeBusyMdnsService {
+    Free(MdnsService),
+    Busy(Pin<Box<dyn Future<Output = (MdnsService, MdnsPacket)> + Send>>),
+    Poisoned,
+}
+
+impl fmt::Debug for MaybeBusyMdnsService {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            MaybeBusyMdnsService::Free(service) => {
+                fmt.debug_struct("MaybeBusyMdnsService::Free")
+                    .field("service", service)
+                    .finish()
+            },
+            MaybeBusyMdnsService::Busy(_) => {
+                fmt.debug_struct("MaybeBusyMdnsService::Busy")
+                    .finish()
+            }
+            MaybeBusyMdnsService::Poisoned => {
+                fmt.debug_struct("MaybeBusyMdnsService::Poisoned")
+                    .finish()
+            }
+        }
+    }
+}
+
+impl Mdns {
+    /// Builds a new `Mdns` behaviour.
+    pub fn new() -> io::Result<Mdns> {
+        Ok(Mdns {
+            service: MaybeBusyMdnsService::Free(MdnsService::new()?),
+            discovered_nodes: SmallVec::new(),
+            closest_expiration: None,
+        })
+    }
+
+    /// Returns true if the given `PeerId` is in the list of nodes discovered through mDNS.
+    pub fn has_node(&self, peer_id: &PeerId) -> bool {
+        self.discovered_nodes.iter().any(|(p, _, _)| p == peer_id)
+    }
+}
+
+/// Event that can be produced by the `Mdns` behaviour.
+#[derive(Debug)]
+pub enum MdnsEvent {
+    /// Discovered nodes through mDNS.
+    Discovered(DiscoveredAddrsIter),
+
+    /// The given combinations of `PeerId` and `Multiaddr` have expired.
+    ///
+    /// Each discovered record has a time-to-live. When this TTL expires and the address hasn't
+    /// been refreshed, we remove it from the list and emit it as an `Expired` event.
+    Expired(ExpiredAddrsIter),
+}
+
+/// Iterator that produces the list of addresses that have been discovered.
+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()
+    }
+}
+
+/// Iterator that produces the list of addresses that have expired.
+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()
+    }
+}
+
+impl NetworkBehaviour for Mdns {
+    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, _: ConnectedPoint) {}
+
+    fn inject_disconnected(&mut self, _: &PeerId, _: ConnectedPoint) {}
+
+    fn inject_node_event(
+        &mut self,
+        _: PeerId,
+        _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,
+        >,
+    > {
+        // Remove expired peers.
+        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),
+            }
+        }
+
+        // Polling the mDNS service, and obtain the list of nodes discovered this round.
+        let discovered = loop {
+            let service = mem::replace(&mut self.service, MaybeBusyMdnsService::Poisoned);
+
+            let packet = match service {
+                MaybeBusyMdnsService::Free(service) => {
+                    self.service = MaybeBusyMdnsService::Busy(Box::pin(service.next()));
+                    continue;
+                },
+                MaybeBusyMdnsService::Busy(mut fut) => {
+                    match fut.as_mut().poll(cx) {
+                        Poll::Ready((service, packet)) => {
+                            self.service = MaybeBusyMdnsService::Free(service);
+                            packet
+                        },
+                        Poll::Pending => {
+                            self.service = MaybeBusyMdnsService::Busy(fut);
+                            return Poll::Pending;
+                        }
+                    }
+                },
+                MaybeBusyMdnsService::Poisoned => panic!("Mdns poisoned"),
+            };
+
+            match packet {
+                MdnsPacket::Query(query) => {
+                    // MaybeBusyMdnsService should always be Free.
+                    if let MaybeBusyMdnsService::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) => {
+                    // We replace the IP address with the address we observe the
+                    // remote as and the address they listen on.
+                    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) => {
+                    // MaybeBusyMdnsService should always be Free.
+                    if let MaybeBusyMdnsService::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); }
+                },
+            }
+        };
+
+        // Getting this far implies that we discovered new nodes. As the final step, we need to
+        // refresh `closest_expiration`.
+        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 Mdns {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("Mdns")
+            .field("service", &self.service)
+            .finish()
+    }
+}

protocols/mdns/src/dns.rs

@@ -0,0 +1,416 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+//
+// 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.
+
+//! Contains methods that handle the DNS encoding and decoding capabilities not available in the
+//! `dns_parser` library.
+
+use crate::{META_QUERY_SERVICE, SERVICE_NAME};
+use data_encoding;
+use libp2p_core::{Multiaddr, PeerId};
+use rand;
+use std::{borrow::Cow, cmp, error, fmt, str, time::Duration};
+
+/// Maximum size of a DNS label as per RFC1035
+const MAX_LABEL_LENGTH: usize = 63;
+
+/// Decodes a `<character-string>` (as defined by RFC1035) into a `Vec` of ASCII characters.
+// TODO: better error type?
+pub fn decode_character_string(mut from: &[u8]) -> Result<Cow<'_, [u8]>, ()> {
+    if from.is_empty() {
+        return Ok(Cow::Owned(Vec::new()));
+    }
+
+    // Remove the initial and trailing " if any.
+    if from[0] == b'"' {
+        if from.len() == 1 || from.last() != Some(&b'"') {
+            return Err(());
+        }
+        let len = from.len();
+        from = &from[1..len - 1];
+    }
+
+    // TODO: remove the backslashes if any
+    Ok(Cow::Borrowed(from))
+}
+
+/// Builds the binary representation of a DNS query to send on the network.
+pub fn build_query() -> Vec<u8> {
+    let mut out = Vec::with_capacity(33);
+
+    // Program-generated transaction ID; unused by our implementation.
+    append_u16(&mut out, rand::random());
+
+    // 0x0 flag for a regular query.
+    append_u16(&mut out, 0x0);
+
+    // Number of questions.
+    append_u16(&mut out, 0x1);
+
+    // Number of answers, authorities, and additionals.
+    append_u16(&mut out, 0x0);
+    append_u16(&mut out, 0x0);
+    append_u16(&mut out, 0x0);
+
+    // Our single question.
+    // The name.
+    append_qname(&mut out, SERVICE_NAME);
+
+    // Flags.
+    append_u16(&mut out, 0x0c);
+    append_u16(&mut out, 0x01);
+
+    // Since the output is constant, we reserve the right amount ahead of time.
+    // If this assert fails, adjust the capacity of `out` in the source code.
+    debug_assert_eq!(out.capacity(), out.len());
+    out
+}
+
+/// Builds the response to the DNS query.
+///
+/// If there are more than 2^16-1 addresses, ignores the rest.
+pub fn build_query_response(
+    id: u16,
+    peer_id: PeerId,
+    addresses: impl ExactSizeIterator<Item = Multiaddr>,
+    ttl: Duration,
+) -> Result<Vec<u8>, MdnsResponseError> {
+    // Convert the TTL into seconds.
+    let ttl = duration_to_secs(ttl);
+
+    // Add a limit to 2^16-1 addresses, as the protocol limits to this number.
+    let addresses = addresses.take(65535);
+
+    // This capacity was determined empirically and is a reasonable upper limit.
+    let mut out = Vec::with_capacity(320);
+
+    append_u16(&mut out, id);
+    // 0x84 flag for an answer.
+    append_u16(&mut out, 0x8400);
+    // Number of questions, answers, authorities, additionals.
+    append_u16(&mut out, 0x0);
+    append_u16(&mut out, 0x1);
+    append_u16(&mut out, 0x0);
+    append_u16(&mut out, addresses.len() as u16);
+
+    // Our single answer.
+    // The name.
+    append_qname(&mut out, SERVICE_NAME);
+
+    // Flags.
+    append_u16(&mut out, 0x000c);
+    append_u16(&mut out, 0x0001);
+
+    // TTL for the answer
+    append_u32(&mut out, ttl);
+
+    // Peer Id.
+    let peer_id_bytes = encode_peer_id(&peer_id);
+    debug_assert!(peer_id_bytes.len() <= 0xffff);
+    append_u16(&mut out, peer_id_bytes.len() as u16);
+    out.extend_from_slice(&peer_id_bytes);
+
+    // The TXT records for answers.
+    for addr in addresses {
+        let txt_to_send = format!("dnsaddr={}/p2p/{}", addr.to_string(), peer_id.to_base58());
+        let mut txt_to_send_bytes = Vec::with_capacity(txt_to_send.len());
+        append_character_string(&mut txt_to_send_bytes, txt_to_send.as_bytes())?;
+        append_txt_record(&mut out, &peer_id_bytes, ttl, Some(&txt_to_send_bytes[..]))?;
+    }
+
+    // The DNS specs specify that the maximum allowed size is 9000 bytes.
+    if out.len() > 9000 {
+        return Err(MdnsResponseError::ResponseTooLong);
+    }
+
+    Ok(out)
+}
+
+/// Builds the response to the DNS query.
+pub fn build_service_discovery_response(id: u16, ttl: Duration) -> Vec<u8> {
+    // Convert the TTL into seconds.
+    let ttl = duration_to_secs(ttl);
+
+    // This capacity was determined empirically.
+    let mut out = Vec::with_capacity(69);
+
+    append_u16(&mut out, id);
+    // 0x84 flag for an answer.
+    append_u16(&mut out, 0x8400);
+    // Number of questions, answers, authorities, additionals.
+    append_u16(&mut out, 0x0);
+    append_u16(&mut out, 0x1);
+    append_u16(&mut out, 0x0);
+    append_u16(&mut out, 0x0);
+
+    // Our single answer.
+    // The name.
+    append_qname(&mut out, META_QUERY_SERVICE);
+
+    // Flags.
+    append_u16(&mut out, 0x000c);
+    append_u16(&mut out, 0x8001);
+
+    // TTL for the answer
+    append_u32(&mut out, ttl);
+
+    // Service name.
+    {
+        let mut name = Vec::with_capacity(SERVICE_NAME.len() + 2);
+        append_qname(&mut name, SERVICE_NAME);
+        append_u16(&mut out, name.len() as u16);
+        out.extend_from_slice(&name);
+    }
+
+    // Since the output size is constant, we reserve the right amount ahead of time.
+    // If this assert fails, adjust the capacity of `out` in the source code.
+    debug_assert_eq!(out.capacity(), out.len());
+    out
+}
+
+/// Returns the number of secs of a duration.
+fn duration_to_secs(duration: Duration) -> u32 {
+    let secs = duration
+        .as_secs()
+        .saturating_add(if duration.subsec_nanos() > 0 { 1 } else { 0 });
+    cmp::min(secs, From::from(u32::max_value())) as u32
+}
+
+/// Appends a big-endian u32 to `out`.
+fn append_u32(out: &mut Vec<u8>, value: u32) {
+    out.push(((value >> 24) & 0xff) as u8);
+    out.push(((value >> 16) & 0xff) as u8);
+    out.push(((value >> 8) & 0xff) as u8);
+    out.push((value & 0xff) as u8);
+}
+
+/// Appends a big-endian u16 to `out`.
+fn append_u16(out: &mut Vec<u8>, value: u16) {
+    out.push(((value >> 8) & 0xff) as u8);
+    out.push((value & 0xff) as u8);
+}
+
+/// If a peer ID is longer than 63 characters, split it into segments to
+/// be compatible with RFC 1035.
+fn segment_peer_id(peer_id: String) -> String {
+    // Guard for the most common case
+    if peer_id.len() <= MAX_LABEL_LENGTH { return peer_id }
+
+    // This will only perform one allocation except in extreme circumstances.
+    let mut out = String::with_capacity(peer_id.len() + 8);
+
+    for (idx, chr) in peer_id.chars().enumerate() {
+        if idx > 0 && idx % MAX_LABEL_LENGTH == 0 {
+            out.push('.');
+        }
+        out.push(chr);
+    }
+    out
+}
+
+/// Combines and encodes a `PeerId` and service name for a DNS query.
+fn encode_peer_id(peer_id: &PeerId) -> Vec<u8> {
+    // DNS-safe encoding for the Peer ID 
+    let raw_peer_id = data_encoding::BASE32_DNSCURVE.encode(&peer_id.as_bytes());
+    // ensure we don't have any labels over 63 bytes long
+    let encoded_peer_id = segment_peer_id(raw_peer_id);
+    let service_name = str::from_utf8(SERVICE_NAME).expect("SERVICE_NAME is always ASCII");
+    let peer_name = [&encoded_peer_id, service_name].join(".");
+
+    // allocate with a little extra padding for QNAME encoding
+    let mut peer_id_bytes = Vec::with_capacity(peer_name.len() + 32);
+    append_qname(&mut peer_id_bytes, peer_name.as_bytes());
+
+    peer_id_bytes
+}
+
+/// Appends a `QNAME` (as defined by RFC1035) to the `Vec`.
+///
+/// # Panic
+///
+/// Panics if `name` has a zero-length component or a component that is too long.
+/// This is fine considering that this function is not public and is only called in a controlled
+/// environment.
+///
+fn append_qname(out: &mut Vec<u8>, name: &[u8]) {
+    debug_assert!(name.is_ascii());
+
+    for element in name.split(|&c| c == b'.') {
+        assert!(element.len() < 64, "Service name has a label too long");
+        assert_ne!(element.len(), 0, "Service name contains zero length label");
+        out.push(element.len() as u8);
+        for chr in element.iter() {
+            out.push(*chr);
+        }
+    }
+
+    out.push(0);
+}
+
+/// Appends a `<character-string>` (as defined by RFC1035) to the `Vec`.
+fn append_character_string(out: &mut Vec<u8>, ascii_str: &[u8]) -> Result<(), MdnsResponseError> {
+    if !ascii_str.is_ascii() {
+        return Err(MdnsResponseError::NonAsciiMultiaddr);
+    }
+
+    if !ascii_str.iter().any(|&c| c == b' ') {
+        for &chr in ascii_str.iter() {
+            out.push(chr);
+        }
+        return Ok(());
+    }
+
+    out.push(b'"');
+
+    for &chr in ascii_str.iter() {
+        if chr == b'\\' {
+            out.push(b'\\');
+            out.push(b'\\');
+        } else if chr == b'"' {
+            out.push(b'\\');
+            out.push(b'"');
+        } else {
+            out.push(chr);
+        }
+    }
+
+    out.push(b'"');
+    Ok(())
+}
+
+/// Appends a TXT record to the answer in `out`.
+fn append_txt_record<'a>(
+    out: &mut Vec<u8>,
+    name: &[u8],
+    ttl_secs: u32,
+    entries: impl IntoIterator<Item = &'a [u8]>,
+) -> Result<(), MdnsResponseError> {
+    // The name.
+    out.extend_from_slice(name);
+
+    // Flags.
+    out.push(0x00);
+    out.push(0x10);     // TXT record.
+    out.push(0x80);
+    out.push(0x01);
+
+    // TTL for the answer
+    append_u32(out, ttl_secs);
+
+    // Add the strings.
+    let mut buffer = Vec::new();
+    for entry in entries {
+        if entry.len() > u8::max_value() as usize {
+            return Err(MdnsResponseError::TxtRecordTooLong);
+        }
+        buffer.push(entry.len() as u8);
+        buffer.extend_from_slice(entry);
+    }
+
+    // It is illegal to have an empty TXT record, but we can have one zero-bytes entry, which does
+    // the same.
+    if buffer.is_empty() {
+        buffer.push(0);
+    }
+
+    if buffer.len() > u16::max_value() as usize {
+        return Err(MdnsResponseError::TxtRecordTooLong);
+    }
+    append_u16(out, buffer.len() as u16);
+    out.extend_from_slice(&buffer);
+    Ok(())
+}
+
+/// Error that can happen when producing a DNS response.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum MdnsResponseError {
+    TxtRecordTooLong,
+    NonAsciiMultiaddr,
+    ResponseTooLong,
+}
+
+impl fmt::Display for MdnsResponseError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            MdnsResponseError::TxtRecordTooLong => {
+                write!(f, "TXT record invalid because it is too long")
+            }
+            MdnsResponseError::NonAsciiMultiaddr => write!(
+                f,
+                "A multiaddr contains non-ASCII characters when serializd"
+            ),
+            MdnsResponseError::ResponseTooLong => write!(f, "DNS response is too long"),
+        }
+    }
+}
+
+impl error::Error for MdnsResponseError {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use dns_parser::Packet;
+    use libp2p_core::identity;
+    use std::time::Duration;
+
+    #[test]
+    fn build_query_correct() {
+        let query = build_query();
+        assert!(Packet::parse(&query).is_ok());
+    }
+
+    #[test]
+    fn build_query_response_correct() {
+        let my_peer_id = identity::Keypair::generate_ed25519().public().into_peer_id();
+        let addr1 = "/ip4/1.2.3.4/tcp/5000".parse().unwrap();
+        let addr2 = "/ip6/::1/udp/10000".parse().unwrap();
+        let query = build_query_response(
+            0xf8f8,
+            my_peer_id,
+            vec![addr1, addr2].into_iter(),
+            Duration::from_secs(60),
+        )
+        .unwrap();
+        assert!(Packet::parse(&query).is_ok());
+    }
+
+    #[test]
+    fn build_service_discovery_response_correct() {
+        let query = build_service_discovery_response(0x1234, Duration::from_secs(120));
+        assert!(Packet::parse(&query).is_ok());
+    }
+
+    #[test]
+    fn test_segment_peer_id() {
+        let str_32 = String::from_utf8(vec![b'x'; 32]).unwrap();
+        let str_63 = String::from_utf8(vec![b'x'; 63]).unwrap();
+        let str_64 = String::from_utf8(vec![b'x'; 64]).unwrap();
+        let str_126 = String::from_utf8(vec![b'x'; 126]).unwrap();
+        let str_127 = String::from_utf8(vec![b'x'; 127]).unwrap();
+
+        assert_eq!(segment_peer_id(str_32.clone()), str_32);
+        assert_eq!(segment_peer_id(str_63.clone()), str_63);
+
+        assert_eq!(segment_peer_id(str_64), [&str_63, "x"].join("."));
+        assert_eq!(segment_peer_id(str_126), [&str_63, str_63.as_str()].join("."));
+        assert_eq!(segment_peer_id(str_127), [&str_63, &str_63, "x"].join("."));
+    }
+
+    // TODO: test limits and errors
+}

protocols/mdns/src/lib.rs

@@ -0,0 +1,44 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+//
+// 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.
+
+//! mDNS is a protocol defined by [RFC 6762](https://tools.ietf.org/html/rfc6762) that allows
+//! querying nodes that correspond to a certain domain name.
+//!
+//! In the context of libp2p, the mDNS protocol is used to discover other nodes on the local
+//! network that support libp2p.
+//!
+//! # Usage
+//!
+//! This crate provides the `Mdns` struct which implements the `NetworkBehaviour` trait. This
+//! struct will automatically discover other libp2p nodes on the local network.
+//!
+
+/// Hardcoded name of the mDNS service. Part of the mDNS libp2p specifications.
+const SERVICE_NAME: &[u8] = b"_p2p._udp.local";
+/// Hardcoded name of the service used for DNS-SD.
+const META_QUERY_SERVICE: &[u8] = b"_services._dns-sd._udp.local";
+
+pub use self::behaviour::{Mdns, MdnsEvent};
+pub use self::service::MdnsService;
+
+mod behaviour;
+mod dns;
+
+pub mod service;

protocols/mdns/src/service.rs

@@ -0,0 +1,652 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+//
+// 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 crate::{SERVICE_NAME, META_QUERY_SERVICE, dns};
+use async_std::net::UdpSocket;
+use dns_parser::{Packet, RData};
+use either::Either::{Left, Right};
+use futures::{future, prelude::*};
+use libp2p_core::{multiaddr::{Multiaddr, Protocol}, PeerId};
+use std::{convert::TryFrom as _, fmt, io, net::Ipv4Addr, net::SocketAddr, str, time::{Duration, Instant}};
+use wasm_timer::Interval;
+use lazy_static::lazy_static;
+
+pub use dns::{MdnsResponseError, build_query_response, build_service_discovery_response};
+
+lazy_static! {
+    static ref IPV4_MDNS_MULTICAST_ADDRESS: SocketAddr = SocketAddr::from((
+        Ipv4Addr::new(224, 0, 0, 251),
+        5353,
+    ));
+}
+
+/// A running service that discovers libp2p peers and responds to other libp2p peers' queries on
+/// the local network.
+///
+/// # Usage
+///
+/// In order to use mDNS to discover peers on the local network, use the `MdnsService`. This is
+/// done by creating a `MdnsService` then polling it in the same way as you would poll a stream.
+///
+/// Polling the `MdnsService` can produce either an `MdnsQuery`, corresponding to an mDNS query
+/// received by another node on the local network, or an `MdnsResponse` corresponding to a response
+/// to a query previously emitted locally. The `MdnsService` will automatically produce queries,
+/// which means that you will receive responses automatically.
+///
+/// When you receive an `MdnsQuery`, use the `respond` method to send back an answer to the node
+/// that emitted the query.
+///
+/// When you receive an `MdnsResponse`, use the provided methods to query the information received
+/// in the response.
+///
+/// # Example
+///
+/// ```rust
+/// # use futures::prelude::*;
+/// # use futures::executor::block_on;
+/// # use libp2p_core::{identity, Multiaddr, PeerId};
+/// # use libp2p_mdns::service::{MdnsService, MdnsPacket, build_query_response, build_service_discovery_response};
+/// # use std::{io, time::Duration, task::Poll};
+/// # fn main() {
+/// # let my_peer_id = PeerId::from(identity::Keypair::generate_ed25519().public());
+/// # let my_listened_addrs: Vec<Multiaddr> = vec![];
+/// # block_on(async {
+/// let mut service = MdnsService::new().expect("Error while creating mDNS service");
+/// let _future_to_poll = async {
+///     let (mut service, packet) = service.next().await;
+///
+///     match packet {
+///         MdnsPacket::Query(query) => {
+///             println!("Query from {:?}", query.remote_addr());
+///             let resp = build_query_response(
+///                 query.query_id(),
+///                 my_peer_id.clone(),
+///                 vec![].into_iter(),
+///                 Duration::from_secs(120),
+///             ).unwrap();
+///             service.enqueue_response(resp);
+///         }
+///         MdnsPacket::Response(response) => {
+///             for peer in response.discovered_peers() {
+///                 println!("Discovered peer {:?}", peer.id());
+///                 for addr in peer.addresses() {
+///                     println!("Address = {:?}", addr);
+///                 }
+///             }
+///         }
+///         MdnsPacket::ServiceDiscovery(disc) => {
+///             let resp = build_service_discovery_response(
+///                 disc.query_id(),
+///                 Duration::from_secs(120),
+///             );
+///             service.enqueue_response(resp);
+///         }
+///     }
+/// };
+/// # })
+/// # }
+pub struct MdnsService {
+    /// Main socket for listening.
+    socket: UdpSocket,
+    /// Socket for sending queries on the network.
+    query_socket: UdpSocket,
+    /// Interval for sending queries.
+    query_interval: Interval,
+    /// Whether we send queries on the network at all.
+    /// Note that we still need to have an interval for querying, as we need to wake up the socket
+    /// regularly to recover from errors. Otherwise we could simply use an `Option<Interval>`.
+    silent: bool,
+    /// Buffer used for receiving data from the main socket.
+    recv_buffer: [u8; 2048],
+    /// Buffers pending to send on the main socket.
+    send_buffers: Vec<Vec<u8>>,
+    /// Buffers pending to send on the query socket.
+    query_send_buffers: Vec<Vec<u8>>,
+}
+
+impl MdnsService {
+    /// Starts a new mDNS service.
+    pub fn new() -> io::Result<MdnsService> {
+        Self::new_inner(false)
+    }
+
+    /// Same as `new`, but we don't send automatically send queries on the network.
+    pub fn silent() -> io::Result<MdnsService> {
+        Self::new_inner(true)
+    }
+
+    /// Starts a new mDNS service.
+    fn new_inner(silent: bool) -> io::Result<MdnsService> {
+        let socket = {
+            #[cfg(unix)]
+            fn platform_specific(s: &net2::UdpBuilder) -> io::Result<()> {
+                net2::unix::UnixUdpBuilderExt::reuse_port(s, true)?;
+                Ok(())
+            }
+            #[cfg(not(unix))]
+            fn platform_specific(_: &net2::UdpBuilder) -> io::Result<()> { Ok(()) }
+            let builder = net2::UdpBuilder::new_v4()?;
+            builder.reuse_address(true)?;
+            platform_specific(&builder)?;
+            builder.bind(("0.0.0.0", 5353))?
+        };
+
+        let socket = UdpSocket::from(socket);
+        socket.set_multicast_loop_v4(true)?;
+        socket.set_multicast_ttl_v4(255)?;
+        // TODO: correct interfaces?
+        socket.join_multicast_v4(From::from([224, 0, 0, 251]), Ipv4Addr::UNSPECIFIED)?;
+
+        Ok(MdnsService {
+            socket,
+            // Given that we pass an IP address to bind, which does not need to be resolved, we can
+            // use std::net::UdpSocket::bind, instead of its async counterpart from async-std.
+            query_socket: std::net::UdpSocket::bind((Ipv4Addr::from([0u8, 0, 0, 0]), 0u16))?.into(),
+            query_interval: Interval::new_at(Instant::now(), Duration::from_secs(20)),
+            silent,
+            recv_buffer: [0; 2048],
+            send_buffers: Vec::new(),
+            query_send_buffers: Vec::new(),
+        })
+    }
+
+    pub fn enqueue_response(&mut self, rsp: Vec<u8>) {
+        self.send_buffers.push(rsp);
+    }
+
+    /// Returns a future resolving to itself and the next received `MdnsPacket`.
+    //
+    // **Note**: Why does `next` take ownership of itself?
+    //
+    // `MdnsService::next` needs to be called from within `NetworkBehaviour`
+    // implementations. Given that traits cannot have async methods the
+    // respective `NetworkBehaviour` implementation needs to somehow keep the
+    // Future returned by `MdnsService::next` across classic `poll`
+    // invocations. The instance method `next` can either take a reference or
+    // ownership of itself:
+    //
+    // 1. Taking a reference - If `MdnsService::poll` takes a reference to
+    // `&self` the respective `NetworkBehaviour` implementation would need to
+    // keep both the Future as well as its `MdnsService` instance across poll
+    // invocations. Given that in this case the Future would have a reference
+    // to `MdnsService`, the `NetworkBehaviour` implementation struct would
+    // need to be self-referential which is not possible without unsafe code in
+    // Rust.
+    //
+    // 2. Taking ownership - Instead `MdnsService::next` takes ownership of
+    // self and returns it alongside an `MdnsPacket` once the actual future
+    // resolves, not forcing self-referential structures on the caller.
+    pub async fn next(mut self) -> (Self, MdnsPacket) {
+        loop {
+            // Flush the send buffer of the main socket.
+            while !self.send_buffers.is_empty() {
+                let to_send = self.send_buffers.remove(0);
+
+                match self.socket.send_to(&to_send, *IPV4_MDNS_MULTICAST_ADDRESS).await {
+                    Ok(bytes_written) => {
+                        debug_assert_eq!(bytes_written, to_send.len());
+                    }
+                    Err(_) => {
+                        // Errors are non-fatal because they can happen for example if we lose
+                        // connection to the network.
+                        self.send_buffers.clear();
+                        break;
+                    }
+                }
+            }
+
+            // Flush the query send buffer.
+            while !self.query_send_buffers.is_empty() {
+                let to_send = self.query_send_buffers.remove(0);
+
+                match self.query_socket.send_to(&to_send, *IPV4_MDNS_MULTICAST_ADDRESS).await {
+                    Ok(bytes_written) => {
+                        debug_assert_eq!(bytes_written, to_send.len());
+                    }
+                    Err(_) => {
+                        // Errors are non-fatal because they can happen for example if we lose
+                        // connection to the network.
+                        self.query_send_buffers.clear();
+                        break;
+                    }
+                }
+            }
+
+            // Either (left) listen for incoming packets or (right) send query packets whenever the
+            // query interval fires.
+            let selected_output = match futures::future::select(
+                Box::pin(self.socket.recv_from(&mut self.recv_buffer)),
+                Box::pin(self.query_interval.next()),
+            ).await {
+                future::Either::Left((recved, _)) => Left(recved),
+                future::Either::Right(_) => Right(()),
+            };
+
+            match selected_output {
+                Left(left) => match left {
+                    Ok((len, from)) => {
+                        match MdnsPacket::new_from_bytes(&self.recv_buffer[..len], from) {
+                            Some(packet) => return (self, packet),
+                            None => {},
+                        }
+                    },
+                    Err(_) => {
+                        // Errors are non-fatal and can happen if we get disconnected from the network.
+                        // The query interval will wake up the task at some point so that we can try again.
+                    },
+                },
+                Right(_) => {
+                    // Ensure underlying task is woken up on the next interval tick.
+                    while let Some(_) = self.query_interval.next().now_or_never() {};
+
+                    if !self.silent {
+                        let query = dns::build_query();
+                        self.query_send_buffers.push(query.to_vec());
+                    }
+                }
+            };
+        }
+    }
+}
+
+impl fmt::Debug for MdnsService {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("MdnsService")
+            .field("silent", &self.silent)
+            .finish()
+    }
+}
+
+/// A valid mDNS packet received by the service.
+#[derive(Debug)]
+pub enum MdnsPacket {
+    /// A query made by a remote.
+    Query(MdnsQuery),
+    /// A response sent by a remote in response to one of our queries.
+    Response(MdnsResponse),
+    /// A request for service discovery.
+    ServiceDiscovery(MdnsServiceDiscovery),
+}
+
+impl MdnsPacket {
+    fn new_from_bytes(buf: &[u8], from: SocketAddr) -> Option<MdnsPacket> {
+        match Packet::parse(buf) {
+            Ok(packet) => {
+                if packet.header.query {
+                    if packet
+                        .questions
+                        .iter()
+                        .any(|q| q.qname.to_string().as_bytes() == SERVICE_NAME)
+                    {
+                        let query = MdnsPacket::Query(MdnsQuery {
+                            from,
+                            query_id: packet.header.id,
+                        });
+                        return Some(query);
+                    } else if packet
+                        .questions
+                        .iter()
+                        .any(|q| q.qname.to_string().as_bytes() == META_QUERY_SERVICE)
+                    {
+                        // TODO: what if multiple questions, one with SERVICE_NAME and one with META_QUERY_SERVICE?
+                        let discovery = MdnsPacket::ServiceDiscovery(
+                            MdnsServiceDiscovery {
+                                from,
+                                query_id: packet.header.id,
+                            },
+                        );
+                        return Some(discovery);
+                    } else {
+                        return None;
+                    }
+                } else {
+                    let resp = MdnsPacket::Response(MdnsResponse::new (
+                        packet,
+                        from,
+                    ));
+                    return Some(resp);
+                }
+            }
+            Err(_) => {
+                return None;
+            }
+        }
+    }
+}
+
+/// A received mDNS query.
+pub struct MdnsQuery {
+    /// Sender of the address.
+    from: SocketAddr,
+    /// Id of the received DNS query. We need to pass this ID back in the results.
+    query_id: u16,
+}
+
+impl MdnsQuery {
+    /// Source address of the packet.
+    pub fn remote_addr(&self) -> &SocketAddr {
+        &self.from
+    }
+
+    /// Query id of the packet.
+    pub fn query_id(&self) -> u16 {
+        self.query_id
+    }
+}
+
+impl fmt::Debug for MdnsQuery {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("MdnsQuery")
+            .field("from", self.remote_addr())
+            .field("query_id", &self.query_id)
+            .finish()
+    }
+}
+
+/// A received mDNS service discovery query.
+pub struct MdnsServiceDiscovery {
+    /// Sender of the address.
+    from: SocketAddr,
+    /// Id of the received DNS query. We need to pass this ID back in the results.
+    query_id: u16,
+}
+
+impl MdnsServiceDiscovery {
+    /// Source address of the packet.
+    pub fn remote_addr(&self) -> &SocketAddr {
+        &self.from
+    }
+
+    /// Query id of the packet.
+    pub fn query_id(&self) -> u16 {
+        self.query_id
+    }
+}
+
+impl fmt::Debug for MdnsServiceDiscovery {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("MdnsServiceDiscovery")
+            .field("from", self.remote_addr())
+            .field("query_id", &self.query_id)
+            .finish()
+    }
+}
+
+/// A received mDNS response.
+pub struct MdnsResponse {
+    peers: Vec<MdnsPeer>,
+    from: SocketAddr,
+}
+
+impl MdnsResponse {
+    /// Creates a new `MdnsResponse` based on the provided `Packet`.
+    fn new(packet: Packet, from: SocketAddr) -> MdnsResponse {
+        let peers = packet.answers.iter().filter_map(|record| {
+            if record.name.to_string().as_bytes() != SERVICE_NAME {
+                return None;
+            }
+
+            let record_value = match record.data {
+                RData::PTR(record) => record.0.to_string(),
+                _ => return None,
+            };
+
+            let mut peer_name = match record_value.rsplitn(4, |c| c == '.').last() {
+                Some(n) => n.to_owned(),
+                None => return None,
+            };
+
+            // if we have a segmented name, remove the '.'
+            peer_name.retain(|c| c != '.');
+
+            let peer_id = match data_encoding::BASE32_DNSCURVE.decode(peer_name.as_bytes()) {
+                Ok(bytes) => match PeerId::from_bytes(bytes) {
+                    Ok(id) => id,
+                    Err(_) => return None,
+                },
+                Err(_) => return None,
+            };
+
+            Some(MdnsPeer::new (
+                &packet,
+                record_value,
+                peer_id,
+                record.ttl,
+            ))
+        }).collect();
+
+        MdnsResponse {
+            peers,
+            from,
+        }
+    }
+
+    /// Returns the list of peers that have been reported in this packet.
+    ///
+    /// > **Note**: Keep in mind that this will also contain the responses we sent ourselves.
+    pub fn discovered_peers(&self) -> impl Iterator<Item = &MdnsPeer> {
+        self.peers.iter()
+    }
+
+    /// Source address of the packet.
+    #[inline]
+    pub fn remote_addr(&self) -> &SocketAddr {
+        &self.from
+    }
+}
+
+impl fmt::Debug for MdnsResponse {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("MdnsResponse")
+            .field("from", self.remote_addr())
+            .finish()
+    }
+}
+
+/// A peer discovered by the service.
+pub struct MdnsPeer {
+    addrs: Vec<Multiaddr>,
+    /// Id of the peer.
+    peer_id: PeerId,
+    /// TTL of the record in seconds.
+    ttl: u32,
+}
+
+impl MdnsPeer {
+    /// Creates a new `MdnsPeer` based on the provided `Packet`.
+    pub fn new(packet: &Packet, record_value: String, my_peer_id: PeerId, ttl: u32) -> MdnsPeer {
+        let addrs = packet
+            .additional
+            .iter()
+            .filter_map(|add_record| {
+                if add_record.name.to_string() != record_value {
+                    return None;
+                }
+
+                if let RData::TXT(ref txt) = add_record.data {
+                    Some(txt)
+                } else {
+                    None
+                }
+            })
+            .flat_map(|txt| txt.iter())
+            .filter_map(|txt| {
+                // TODO: wrong, txt can be multiple character strings
+                let addr = match dns::decode_character_string(txt) {
+                    Ok(a) => a,
+                    Err(_) => return None,
+                };
+                if !addr.starts_with(b"dnsaddr=") {
+                    return None;
+                }
+                let addr = match str::from_utf8(&addr[8..]) {
+                    Ok(a) => a,
+                    Err(_) => return None,
+                };
+                let mut addr = match addr.parse::<Multiaddr>() {
+                    Ok(a) => a,
+                    Err(_) => return None,
+                };
+                match addr.pop() {
+                    Some(Protocol::P2p(peer_id)) => {
+                        if let Ok(peer_id) = PeerId::try_from(peer_id) {
+                            if peer_id != my_peer_id {
+                                return None;
+                            }
+                        } else {
+                            return None;
+                        }
+                    },
+                    _ => return None,
+                };
+                Some(addr)
+            }).collect();
+
+        MdnsPeer {
+            addrs,
+            peer_id: my_peer_id.clone(),
+            ttl,
+        }
+    }
+
+    /// Returns the id of the peer.
+    #[inline]
+    pub fn id(&self) -> &PeerId {
+        &self.peer_id
+    }
+
+    /// Returns the requested time-to-live for the record.
+    #[inline]
+    pub fn ttl(&self) -> Duration {
+        Duration::from_secs(u64::from(self.ttl))
+    }
+
+    /// Returns the list of addresses the peer says it is listening on.
+    ///
+    /// Filters out invalid addresses.
+    pub fn addresses(&self) -> &Vec<Multiaddr> {
+        &self.addrs
+    }
+}
+
+impl fmt::Debug for MdnsPeer {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("MdnsPeer")
+            .field("peer_id", &self.peer_id)
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use futures::executor::block_on;
+    use libp2p_core::{PeerId, multiaddr::multihash::*};
+    use std::{io::{Error, ErrorKind}, time::Duration};
+    use wasm_timer::ext::TryFutureExt;
+    use crate::service::{MdnsPacket, MdnsService};
+
+    fn discover(peer_id: PeerId) {
+        block_on(async {
+            let mut service = MdnsService::new().unwrap();
+            loop {
+                let next = service.next().await;
+                service = next.0;
+
+                match next.1 {
+                    MdnsPacket::Query(query) => {
+                        let resp = crate::dns::build_query_response(
+                            query.query_id(),
+                            peer_id.clone(),
+                            vec![].into_iter(),
+                            Duration::from_secs(120),
+                        ).unwrap();
+                        service.enqueue_response(resp);
+                    }
+                    MdnsPacket::Response(response) => {
+                        for peer in response.discovered_peers() {
+                            if peer.id() == &peer_id {
+                                return;
+                            }
+                        }
+                    }
+                    MdnsPacket::ServiceDiscovery(_) => panic!("did not expect a service discovery packet")
+                }
+            }
+        })
+    }
+
+    // As of today the underlying UDP socket is not stubbed out. Thus tests run in parallel to this
+    // unit tests inter fear with it. Test needs to be run in sequence to ensure test properties.
+    #[test]
+    fn respect_query_interval() {
+        let own_ips: Vec<std::net::IpAddr> = get_if_addrs::get_if_addrs().unwrap()
+            .into_iter()
+            .map(|i| i.addr.ip())
+            .collect();
+
+        let fut = async {
+            let mut service = MdnsService::new().unwrap();
+            let mut sent_queries = vec![];
+
+            loop {
+                let next = service.next().await;
+                service = next.0;
+
+                match next.1 {
+                    MdnsPacket::Query(query) => {
+                        // Ignore queries from other nodes.
+                        let source_ip = query.remote_addr().ip();
+                        if !own_ips.contains(&source_ip) {
+                            continue;
+                        }
+
+                        sent_queries.push(query);
+
+                        if sent_queries.len() > 1 {
+                            return Ok(())
+                        }
+                    }
+                    // Ignore response packets. We don't stub out the UDP socket, thus this is
+                    // either random noise from the network, or noise from other unit tests running
+                    // in parallel.
+                    MdnsPacket::Response(_) => {},
+                    MdnsPacket::ServiceDiscovery(_) => {
+                        return Err(Error::new(ErrorKind::Other, "did not expect a service discovery packet"));
+                    },
+                }
+            }
+        };
+
+        // TODO: This might be too long for a unit test.
+        block_on(fut.timeout(Duration::from_secs(41))).unwrap();
+    }
+
+    #[test]
+    fn discover_normal_peer_id() {
+        discover(PeerId::random())
+    }
+
+    #[test]
+    fn discover_long_peer_id() {
+        let max_value = String::from_utf8(vec![b'f'; 42]).unwrap();
+        let hash = encode(Hash::Identity, max_value.as_ref()).unwrap();
+        discover(PeerId::from_multihash(hash).unwrap())
+    }
+}