massping/
lib.rs

1//! Asynchronous ICMP ping library using Linux DGRAM sockets and the
2//! tokio runtime.
3//!
4//! This crate uses `SOCK_DGRAM` sockets with `IPPROTO_ICMP`/`IPPROTO_ICMPV6`,
5//! which allows sending ICMP echo requests without root privileges on Linux.
6//!
7//! ## Features
8//!
9//! * `stream`: implements [`Stream`] for [`MeasureManyStream`].
10//!
11//! ## MSRV version policy
12//!
13//! This project has a CI job to prevent accidental bumping of the MSRV.
14//! We might bump MSRV version at any time. If you require a lower MSRV
15//! please open an issue.
16//!
17//! [`Stream`]: futures_core::Stream
18
19#![deny(
20    rust_2018_idioms,
21    unreachable_pub,
22    clippy::doc_markdown,
23    rustdoc::broken_intra_doc_links
24)]
25
26#[cfg(feature = "stream")]
27use std::pin::Pin;
28use std::{
29    io,
30    marker::PhantomData,
31    net::{IpAddr, Ipv4Addr, Ipv6Addr},
32    task::{Context, Poll},
33    time::Duration,
34};
35
36#[cfg(feature = "stream")]
37use futures_core::Stream;
38
39pub use self::{
40    ip_version::IpVersion,
41    pinger::{MeasureManyStream, Pinger, V4Pinger, V6Pinger},
42};
43
44mod ip_version;
45pub mod packet;
46mod pinger;
47pub mod raw_pinger;
48mod socket;
49
50/// A pinger for both [`Ipv4Addr`] and [`Ipv6Addr`] addresses.
51pub struct DualstackPinger {
52    v4: V4Pinger,
53    v6: V6Pinger,
54}
55
56impl DualstackPinger {
57    /// Construct a new `DualstackPinger`.
58    ///
59    /// For maximum efficiency the same instance of `DualstackPinger` should
60    /// be used for as long as possible, altough it might also
61    /// be beneficial to `Drop` the `DualstackPinger` and recreate it if
62    /// you are not going to be sending pings for a long period of time.
63    pub fn new() -> io::Result<Self> {
64        let v4 = V4Pinger::new()?;
65        let v6 = V6Pinger::new()?;
66        Ok(Self { v4, v6 })
67    }
68
69    /// Ping `addresses`
70    ///
71    /// Creates [`DualstackMeasureManyStream`] which **lazily** sends ping
72    /// requests and [`Stream`]s the responses as they arrive.
73    ///
74    /// [`Stream`]: futures_core::Stream
75    pub fn measure_many<I>(&self, addresses: I) -> DualstackMeasureManyStream<'_, I>
76    where
77        I: Iterator<Item = IpAddr> + Clone,
78    {
79        let addresses_v4 = FilterIpAddr {
80            iter: addresses.clone(),
81            _marker: PhantomData,
82        };
83        let addresses_v6 = FilterIpAddr {
84            iter: addresses,
85            _marker: PhantomData,
86        };
87
88        DualstackMeasureManyStream {
89            v4: self.v4.measure_many(addresses_v4),
90            v6: self.v6.measure_many(addresses_v6),
91            v4_done: false,
92            v6_done: false,
93        }
94    }
95}
96
97/// A [`Stream`] of ping responses.
98///
99/// No kind of `rtt` timeout is implemented, so an external mechanism
100/// like [`tokio::time::timeout`] should be used to prevent the program
101/// from hanging indefinitely.
102///
103/// Leaking this method might crate a slowly forever growing memory leak.
104///
105/// [`Stream`]: futures_core::Stream
106/// [`tokio::time::timeout`]: tokio::time::timeout
107pub struct DualstackMeasureManyStream<'a, I: Iterator<Item = IpAddr>> {
108    v4: MeasureManyStream<'a, Ipv4Addr, FilterIpAddr<I, Ipv4Addr>>,
109    v6: MeasureManyStream<'a, Ipv6Addr, FilterIpAddr<I, Ipv6Addr>>,
110    v4_done: bool,
111    v6_done: bool,
112}
113
114impl<I: Iterator<Item = IpAddr>> DualstackMeasureManyStream<'_, I> {
115    pub fn poll_next_unpin(&mut self, cx: &mut Context<'_>) -> Poll<Option<(IpAddr, Duration)>> {
116        if !self.v4_done {
117            match self.v4.poll_next_unpin(cx) {
118                Poll::Ready(Some((v4, rtt))) => return Poll::Ready(Some((IpAddr::V4(v4), rtt))),
119                Poll::Ready(None) => self.v4_done = true,
120                Poll::Pending => {}
121            }
122        }
123
124        if !self.v6_done {
125            match self.v6.poll_next_unpin(cx) {
126                Poll::Ready(Some((v6, rtt))) => return Poll::Ready(Some((IpAddr::V6(v6), rtt))),
127                Poll::Ready(None) => self.v6_done = true,
128                Poll::Pending => {}
129            }
130        }
131
132        if self.v4_done && self.v6_done {
133            return Poll::Ready(None);
134        }
135
136        Poll::Pending
137    }
138}
139
140#[cfg(feature = "stream")]
141impl<I: Iterator<Item = IpAddr> + Unpin> Stream for DualstackMeasureManyStream<'_, I> {
142    type Item = (IpAddr, Duration);
143
144    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
145        self.as_mut().poll_next_unpin(cx)
146    }
147}
148
149struct FilterIpAddr<I, V: IpVersion> {
150    iter: I,
151    _marker: PhantomData<V>,
152}
153
154impl<I: Iterator<Item = IpAddr>, V: IpVersion> Iterator for FilterIpAddr<I, V> {
155    type Item = V;
156
157    fn next(&mut self) -> Option<Self::Item> {
158        loop {
159            let item = self.iter.next()?;
160            if let Some(addr) = V::from_ip_addr(item) {
161                return Some(addr);
162            }
163        }
164    }
165}
massping/lib.rs

massping/
lib.rs
