async_pcap/
async_pcap.rs

1use std::sync::Arc;
2use std::sync::atomic::{AtomicBool, Ordering};
3
4use pcap::{Active, Capture, Error, PacketHeader};
5use tokio::sync::Mutex;
6use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel};
7
8/// Represents a network packet with its header and raw data.
9#[derive(Debug, Clone)]
10pub struct Packet {
11    /// Packet header information provided by pcap
12    pub header: PacketHeader,
13    /// Raw packet data
14    pub data: Vec<u8>,
15}
16
17/// An asynchronous wrapper around a `pcap::Capture`.
18///  
19/// `AsyncCapture` owns the receiver side of a channel that receives
20/// captured packets or a stop signal. It allows async code to
21/// `await` new packets without blocking a thread.
22pub struct AsyncCapture {
23    rx: Mutex<UnboundedReceiver<PacketOrStop>>,
24}
25
26/// Enum used internally to represent either a captured packet
27/// or a stop signal to terminate the capture.
28enum PacketOrStop {
29    /// A captured packet
30    Packet(Result<Packet, Error>),
31    /// Signal that capture has stopped
32    Stop,
33}
34
35/// Handle to control the asynchronous capture.
36///  
37/// `AsyncCaptureHandle` allows stopping the capture from another
38/// thread or async task.
39#[derive(Clone)]
40pub struct AsyncCaptureHandle {
41    stop_flag: Arc<AtomicBool>,
42}
43
44impl AsyncCapture {
45    /// Creates a new asynchronous capture from a `pcap::Capture<Active>`.
46    ///
47    /// Spawns a background thread that reads packets and sends them
48    /// through a channel for async consumption.
49    ///
50    /// Returns a tuple of `(AsyncCapture, AsyncCaptureHandle)`.
51    pub fn new(mut cap: Capture<Active>) -> (Self, AsyncCaptureHandle) {
52        let (tx, rx) = unbounded_channel::<PacketOrStop>();
53        let stop_flag = Arc::new(AtomicBool::new(false));
54        let handle = AsyncCaptureHandle {
55            stop_flag: stop_flag.clone(),
56        };
57
58        std::thread::spawn(move || {
59            loop {
60                let res = cap.next_packet();
61                let owned = res.map(|packet| Packet {
62                    header: *packet.header,
63                    data: packet.data.to_vec(),
64                });
65                if let Err(e) = tx.send(PacketOrStop::Packet(owned)) {
66                    // Receiver dropped, exit thread
67                    log::warn!("{e}");
68                    break;
69                }
70                if stop_flag.load(Ordering::Relaxed) {
71                    log::warn!("AsyncCapture thread is aborted.");
72                    break;
73                }
74            }
75            // Send a Stop message when capture thread ends
76            let _ = tx.send(PacketOrStop::Stop);
77        });
78
79        (Self { rx: Mutex::new(rx) }, handle)
80    }
81
82    /// Waits for the next packet asynchronously.
83    ///
84    /// Returns `Some(Result<Packet, Error>)` if a packet is received,
85    /// or `None` if the capture has stopped.
86    pub async fn next_packet(&self) -> Option<Result<Packet, Error>> {
87        let mut rx = self.rx.lock().await;
88        match rx.recv().await {
89            Some(PacketOrStop::Packet(pkt)) => Some(pkt),
90            Some(PacketOrStop::Stop) | None => None,
91        }
92    }
93}
94
95impl AsyncCaptureHandle {
96    /// Stops the capture from another thread or asynchronous task.
97    ///
98    /// This method sets the internal stop flag, signaling the background
99    /// capture thread to terminate gracefully. It also sends a `Stop`
100    /// message through the internal channel to ensure that any awaiting
101    /// calls to [`AsyncCapture::next_packet()`] will return `None`.
102    ///
103    /// # Notes
104    ///
105    /// - Calling this method multiple times is safe and idempotent.
106    /// - Once stopped, the background thread will no longer produce packets.
107    /// - After calling `stop`, any future calls to
108    ///   [`AsyncCapture::next_packet()`] will immediately return `None`.
109    pub fn stop(&self) {
110        self.stop_flag.store(true, Ordering::Relaxed);
111    }
112}
async_pcap/async_pcap.rs

async_pcap/
async_pcap.rs
