hick_reactor/
endpoint.rs

1//! Caller-side handle for an mDNS endpoint.
2
3use std::future::Future;
4
5use agnostic_net::Net;
6use async_channel::Sender;
7use hick_udp::{
8  MulticastOptionsV4, MulticastOptionsV6, try_bind_v4, try_bind_v6, try_join_v4, try_join_v6,
9};
10use mdns_proto::{QuerySpec, ServiceSpec};
11
12use hick_trace::*;
13
14use crate::{
15  command::{Command, QueryStarted, ServiceRegistered},
16  driver::{self, BoundSockets},
17  error::{RegisterError, ServerError, StartQueryError},
18  options::ServerOptions,
19  query::Query,
20  service::Service,
21};
22
23/// Handle to a running mDNS endpoint.
24///
25/// Cloneable; every clone shares the same underlying driver task. The driver
26/// task is spawned at [`Self::server`] time and exits when the last clone is
27/// dropped (the command channel closes).
28#[derive(Clone)]
29pub struct Endpoint {
30  cmd: Sender<Command>,
31  /// Shared stats handle cloned from the driver's proto endpoint. Present
32  /// only when the `stats` Cargo feature is enabled.
33  #[cfg(feature = "stats")]
34  stats: std::sync::Arc<stats::Stats>,
35}
36
37impl Endpoint {
38  /// Bind the multicast sockets configured in `opts` and spawn the driver
39  /// task on the runtime exposed by `N`.
40  pub async fn server<N: Net>(opts: ServerOptions) -> Result<Self, ServerError> {
41    if !opts.ipv4() && !opts.ipv6() {
42      return Err(ServerError::NoFamilyEnabled);
43    }
44
45    let interface_index = match opts.interface_index() {
46      Some(i) => i,
47      None => pick_default_interface_index(opts.ipv4(), opts.ipv6()).ok_or_else(|| {
48        ServerError::Io(std::io::Error::new(
49          std::io::ErrorKind::NotFound,
50          "no multicast-capable interface found",
51        ))
52      })?,
53    };
54
55    // gracefully degrade — if the chosen interface lacks one of
56    // the requested families, bind only the family it can support rather
57    // than failing the entire endpoint. This makes
58    // `with_ipv4(true).with_ipv6(true)` work on hosts with no global v6
59    // (the common single-stack case) while still honouring an explicit
60    // single-family request.
61    let iface_has_v4 = match getifs::interface_by_index(interface_index) {
62      Ok(Some(i)) => matches!(i.ipv4_addrs(), Ok(ref a) if !a.is_empty()),
63      _ => false,
64    };
65    let iface_has_v6 = match getifs::interface_by_index(interface_index) {
66      Ok(Some(i)) => matches!(i.ipv6_addrs(), Ok(ref a) if !a.is_empty()),
67      _ => false,
68    };
69    let bind_v4 = opts.ipv4() && iface_has_v4;
70    let bind_v6 = opts.ipv6() && iface_has_v6;
71    if !bind_v4 && !bind_v6 {
72      return Err(ServerError::Io(std::io::Error::new(
73        std::io::ErrorKind::AddrNotAvailable,
74        "interface has no address in any requested family",
75      )));
76    }
77
78    let v4 = if bind_v4 {
79      match try_bind_v4(MulticastOptionsV4::new(interface_index)) {
80        Ok(std_sock) => {
81          debug!(interface_index, "bound v4 mDNS socket");
82          match try_join_v4(&std_sock, interface_index) {
83            Ok(()) => {
84              debug!(interface_index, "joined v4 mDNS multicast group");
85            }
86            Err(e) => {
87              warn!(error = %e, interface_index, "failed to join v4 mDNS multicast group");
88              return Err(map_join_to_bind_v4(e));
89            }
90          }
91          std_sock.set_nonblocking(true)?;
92          let async_sock = N::UdpSocket::try_from(std_sock).map_err(ServerError::WrapSocket)?;
93          Some(async_sock)
94        }
95        Err(e) => {
96          warn!(error = %e, interface_index, "failed to bind v4 mDNS socket");
97          return Err(ServerError::BindV4(e));
98        }
99      }
100    } else {
101      None
102    };
103
104    let v6 = if bind_v6 {
105      match try_bind_v6(MulticastOptionsV6::new(interface_index)) {
106        Ok(std_sock) => {
107          debug!(interface_index, "bound v6 mDNS socket");
108          match try_join_v6(&std_sock, interface_index) {
109            Ok(()) => {
110              debug!(interface_index, "joined v6 mDNS multicast group");
111            }
112            Err(e) => {
113              warn!(error = %e, interface_index, "failed to join v6 mDNS multicast group");
114              return Err(map_join_to_bind_v6(e));
115            }
116          }
117          std_sock.set_nonblocking(true)?;
118          let async_sock = N::UdpSocket::try_from(std_sock).map_err(ServerError::WrapSocket)?;
119          Some(async_sock)
120        }
121        Err(e) => {
122          warn!(error = %e, interface_index, "failed to bind v6 mDNS socket");
123          return Err(ServerError::BindV6(e));
124        }
125      }
126    } else {
127      None
128    };
129
130    // unbounded so that `Service::drop` / `Query::drop` (which use
131    // `try_send` to issue cleanup commands synchronously) cannot silently
132    // lose the Unregister/Cancel. Drop only fails on a closed channel, which
133    // means the driver task has already exited and there is nothing to
134    // clean up.
135    let (cmd_tx, cmd_rx) = async_channel::unbounded::<Command>();
136    let sockets = BoundSockets {
137      v4,
138      v6,
139      interface_index,
140    };
141    #[cfg(feature = "stats")]
142    let mut stats_slot: Option<std::sync::Arc<stats::Stats>> = None;
143    driver::spawn::<N>(
144      opts,
145      sockets,
146      cmd_rx,
147      #[cfg(feature = "stats")]
148      &mut stats_slot,
149    );
150
151    Ok(Self {
152      cmd: cmd_tx,
153      #[cfg(feature = "stats")]
154      stats: stats_slot.expect("spawn always populates stats_slot when stats feature is enabled"),
155    })
156  }
157
158  /// Return a point-in-time snapshot of the I/O + protocol counters for this
159  /// endpoint.
160  ///
161  /// The snapshot includes both counters incremented by the `mdns-proto` layer
162  /// (parse errors, cache operations, service/query lifecycle) and counters
163  /// added by the driver layer (raw wire rx/tx byte counts, socket-level send
164  /// errors). All counters share the same [`stats::Stats`] instance
165  /// so the snapshot is a single consistent view.
166  #[cfg(feature = "stats")]
167  #[cfg_attr(docsrs, doc(cfg(feature = "stats")))]
168  pub fn stats(&self) -> stats::StatsSnapshot {
169    self.stats.snapshot()
170  }
171
172  /// Hand a detached discovery-lookup driver task to the driver to spawn (via
173  /// [`Command::SpawnLookup`]).
174  ///
175  /// Spawning happens inside the driver task, which always runs in the runtime
176  /// context the endpoint was created on. Routing it this way — rather than
177  /// spawning from the caller — means `browse()` works from any executor or
178  /// thread, even one with no entered runtime of its own, matching the
179  /// runtime-agnostic, channel-only nature of the rest of the endpoint API.
180  pub(crate) fn spawn_lookup<F>(&self, fut: F) -> Result<(), StartQueryError>
181  where
182    F: Future<Output = ()> + Send + 'static,
183  {
184    self
185      .cmd
186      .try_send(Command::SpawnLookup {
187        task: Box::pin(fut),
188      })
189      .map_err(|_| StartQueryError::DriverGone)
190  }
191
192  /// Register a new service with the responder. The returned [`Service`]
193  /// streams [`ServiceUpdate`](mdns_proto::ServiceUpdate) events; dropping
194  /// it unregisters the service.
195  pub async fn register_service(&self, spec: ServiceSpec) -> Result<Service, RegisterError> {
196    let (reply_tx, reply_rx) = futures::channel::oneshot::channel();
197    self
198      .cmd
199      .send(Command::RegisterService {
200        spec,
201        reply: reply_tx,
202      })
203      .await
204      .map_err(|_| RegisterError::DriverGone)?;
205    let ServiceRegistered {
206      handle,
207      mailbox,
208      doorbell,
209    } = reply_rx.await.map_err(|_| RegisterError::DriverGone)??;
210    Ok(Service::new(handle, mailbox, doorbell, self.cmd.clone()))
211  }
212
213  /// Start a new query. The returned [`Query`] streams
214  /// [`QueryEvent`](crate::QueryEvent) values; dropping it cancels the query.
215  pub async fn start_query(&self, spec: QuerySpec) -> Result<Query, StartQueryError> {
216    let (reply_tx, reply_rx) = futures::channel::oneshot::channel();
217    self
218      .cmd
219      .send(Command::StartQuery {
220        spec,
221        reply: reply_tx,
222      })
223      .await
224      .map_err(|_| StartQueryError::DriverGone)?;
225    let QueryStarted {
226      handle,
227      mailbox,
228      doorbell,
229    } = reply_rx.await.map_err(|_| StartQueryError::DriverGone)??;
230    Ok(Query::new(handle, mailbox, doorbell, self.cmd.clone()))
231  }
232}
233
234fn map_join_to_bind_v4(e: hick_udp::JoinError) -> ServerError {
235  match e {
236    hick_udp::JoinError::Io(io) => ServerError::BindV4(hick_udp::BindError::Io(io)),
237    hick_udp::JoinError::InterfaceNotFound(d) => {
238      ServerError::BindV4(hick_udp::BindError::InterfaceNotFound(d))
239    }
240    _ => ServerError::Io(std::io::Error::other("unknown JoinError variant")),
241  }
242}
243
244fn map_join_to_bind_v6(e: hick_udp::JoinError) -> ServerError {
245  match e {
246    hick_udp::JoinError::Io(io) => ServerError::BindV6(hick_udp::BindError::Io(io)),
247    hick_udp::JoinError::InterfaceNotFound(d) => {
248      ServerError::BindV6(hick_udp::BindError::InterfaceNotFound(d))
249    }
250    _ => ServerError::Io(std::io::Error::other("unknown JoinError variant")),
251  }
252}
253
254fn pick_default_interface_index(want_v4: bool, want_v6: bool) -> Option<u32> {
255  let ifs = getifs::interfaces().ok()?;
256  let has_v4 = |i: &getifs::Interface| matches!(i.ipv4_addrs(), Ok(ref v) if !v.is_empty());
257  let has_v6 = |i: &getifs::Interface| matches!(i.ipv6_addrs(), Ok(ref v) if !v.is_empty());
258  let multicast_up_non_loopback = |i: &getifs::Interface| -> bool {
259    let f = i.flags();
260    f.contains(getifs::Flags::UP)
261      && f.contains(getifs::Flags::MULTICAST)
262      && !f.contains(getifs::Flags::LOOPBACK)
263      && i.index() != 0
264  };
265  let loopback_up = |i: &getifs::Interface| -> bool {
266    i.flags().contains(getifs::Flags::LOOPBACK) && i.flags().contains(getifs::Flags::UP)
267  };
268  // prefer an interface that satisfies ALL requested families;
269  // fall back to one that satisfies at least one. Without the loose fallback
270  // an IPv4-only NIC on a host with no global IPv6 would be silently
271  // rejected even though it can serve `with_ipv4(true).with_ipv6(true)` over
272  // v4 perfectly well.
273  let strict =
274    |i: &&getifs::Interface| -> bool { (!want_v4 || has_v4(i)) && (!want_v6 || has_v6(i)) };
275  let loose = |i: &&getifs::Interface| -> bool { (want_v4 && has_v4(i)) || (want_v6 && has_v6(i)) };
276  let strict_non_loopback = ifs
277    .iter()
278    .find(|i| multicast_up_non_loopback(i) && strict(i));
279  let loose_non_loopback = ifs
280    .iter()
281    .find(|i| multicast_up_non_loopback(i) && loose(i));
282  let strict_loopback = ifs.iter().find(|i| loopback_up(i) && strict(i));
283  let loose_loopback = ifs.iter().find(|i| loopback_up(i) && loose(i));
284  strict_non_loopback
285    .or(loose_non_loopback)
286    .or(strict_loopback)
287    .or(loose_loopback)
288    .map(|i| i.index())
289}
290
291#[cfg(test)]
292mod tests;
hick_reactor/endpoint.rs

hick_reactor/
endpoint.rs
