24: Builder Api

Provide a builder style api for creating pending handles, then activating them.
Config now supports specifying interface to simplify creation, and a stream can
be created from a config without needing to first create a handle.

Author: dbcfd

README.md

@@ -24,25 +24,25 @@ First, add this to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-pcap-async = "0.3"
+pcap-async = "0.5"
 ```
 
 Next, add this to your crate:
 
 ```rust
 use futures::StreamExt;
 use pcap_async::{Config, Handle, PacketStream};
+use std::convert::TryFrom;
 
 fn main() {
     smol::run(async move {
-        let handle = Handle::lookup().expect("No handle created");
-        let mut provider = PacketStream::new(Config::default(), handle)
-            .expect("Could not create provider")
-            .fuse();
+        let cfg = Config::default();
+        let mut provider = PacketStream::try_from(cfg)
+            .expect("Could not create provider");
         while let Some(packets) = provider.next().await {
 
         }
-        handle.interrupt();
+        provider.interrupt();
     })
 }
 ```

src/bridge_stream.rs

@@ -19,22 +19,19 @@ use crate::errors::Error;
 use crate::handle::Handle;
 use crate::packet::Packet;
 use crate::pcap_util;
-use crate::stream::StreamItem;
+use crate::stream::{Interruptable, StreamItem};
 
 #[pin_project]
-struct CallbackFuture<E, T>
+struct CallbackFuture<T>
 where
-    E: Sync + Send,
-    T: Stream<Item = StreamItem<E>> + Sized + Unpin,
+    T: Stream<Item = StreamItem> + Sized + Unpin,
 {
     idx: usize,
     stream: Option<T>,
 }
 
-impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> Future
-    for CallbackFuture<E, T>
-{
-    type Output = (usize, Option<(T, StreamItem<E>)>);
+impl<T: Stream<Item = StreamItem> + Sized + Unpin> Future for CallbackFuture<T> {
+    type Output = (usize, Option<(T, StreamItem)>);
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
@@ -60,17 +57,22 @@ impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> Future
     }
 }
 
-struct BridgeStreamState<E, T>
+struct BridgeStreamState<T>
 where
-    E: Sync + Send,
-    T: Stream<Item = StreamItem<E>> + Sized + Unpin,
+    T: Interruptable + Sized + Unpin,
 {
     stream: Option<T>,
     current: Vec<Vec<Packet>>,
     complete: bool,
 }
 
-impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> BridgeStreamState<E, T> {
+impl<T: Interruptable + Sized + Unpin> BridgeStreamState<T> {
+    fn interrupt(&self) {
+        if let Some(st) = &self.stream {
+            st.interrupt();
+        }
+    }
+
     fn is_complete(&self) -> bool {
         self.complete && self.current.is_empty()
     }
@@ -100,22 +102,22 @@ impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> BridgeStre
 // `max_buffer_time` will check the spread of packets, and if it to large it will sort what it has and pass it on.
 
 #[pin_project]
-pub struct BridgeStream<E: Sync + Send, T>
+pub struct BridgeStream<T>
 where
-    T: Stream<Item = StreamItem<E>> + Sized + Unpin,
+    T: Interruptable + Sized + Unpin,
 {
-    stream_states: VecDeque<BridgeStreamState<E, T>>,
+    stream_states: VecDeque<BridgeStreamState<T>>,
     max_buffer_time: Duration,
     min_states_needed: usize,
-    poll_queue: FuturesUnordered<CallbackFuture<E, T>>,
+    poll_queue: FuturesUnordered<CallbackFuture<T>>,
 }
 
-impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> BridgeStream<E, T> {
+impl<T: Interruptable + Sized + Unpin> BridgeStream<T> {
     pub fn new(
         streams: Vec<T>,
         max_buffer_time: Duration,
         min_states_needed: usize,
-    ) -> Result<BridgeStream<E, T>, Error> {
+    ) -> Result<BridgeStream<T>, Error> {
         let poll_queue = FuturesUnordered::new();
         let mut stream_states = VecDeque::with_capacity(streams.len());
         for (idx, stream) in streams.into_iter().enumerate() {
@@ -139,10 +141,16 @@ impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> BridgeStre
             poll_queue,
         })
     }
+
+    pub fn interrupt(&self) {
+        for st in &self.stream_states {
+            st.interrupt();
+        }
+    }
 }
 
-fn gather_packets<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin>(
-    stream_states: &mut VecDeque<BridgeStreamState<E, T>>,
+fn gather_packets<T: Interruptable + Sized + Unpin>(
+    stream_states: &mut VecDeque<BridgeStreamState<T>>,
 ) -> Vec<Packet> {
     let mut result = vec![];
     let mut gather_to: Option<SystemTime> = None;
@@ -183,10 +191,11 @@ fn gather_packets<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpi
     result
 }
 
-impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> Stream
-    for BridgeStream<E, T>
+impl<T> Stream for BridgeStream<T>
+where
+    T: Interruptable + Sized + Unpin,
 {
-    type Item = StreamItem<E>;
+    type Item = StreamItem;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.project();
@@ -195,12 +204,12 @@ impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> Stream
             this.stream_states.len(),
             this.poll_queue.len()
         );
-        let states: &mut VecDeque<BridgeStreamState<E, T>> = this.stream_states;
+        let states: &mut VecDeque<BridgeStreamState<T>> = this.stream_states;
         let min_states_needed: usize = *this.min_states_needed;
         let max_buffer_time = this.max_buffer_time;
         let mut max_time_spread: Duration = Duration::from_millis(0);
         let mut not_pending: usize = 0;
-        let mut poll_queue: &mut FuturesUnordered<CallbackFuture<E, T>> = this.poll_queue;
+        let mut poll_queue: &mut FuturesUnordered<CallbackFuture<T>> = this.poll_queue;
 
         loop {
             match Pin::new(&mut poll_queue).poll_next(cx) {
@@ -284,6 +293,7 @@ impl<E: Sync + Send, T: Stream<Item = StreamItem<E>> + Sized + Unpin> Stream
 
 #[cfg(test)]
 mod tests {
+    use std::convert::TryFrom;
     use std::io::Cursor;
     use std::ops::Range;
     use std::path::PathBuf;
@@ -293,9 +303,10 @@ mod tests {
     use futures::{Future, Stream};
     use rand;
 
-    use crate::PacketStream;
+    use crate::{Interface, PacketStream};
 
     use super::*;
+    use std::sync::atomic::{AtomicBool, Ordering};
 
     fn make_packet(ts: usize) -> Packet {
         Packet {
@@ -316,11 +327,10 @@ mod tests {
 
         info!("Testing against {:?}", pcap_path);
 
-        let handle = Handle::file_capture(pcap_path.to_str().expect("No path found"))
-            .expect("No handle created");
+        let mut cfg = Config::default();
+        cfg.with_interface(Interface::File(pcap_path));
 
-        let packet_stream =
-            PacketStream::new(Config::default(), Arc::clone(&handle)).expect("Failed to build");
+        let packet_stream = PacketStream::try_from(cfg).expect("Failed to build");
 
         let packet_provider = BridgeStream::new(vec![packet_stream], Duration::from_millis(100), 2)
             .expect("Failed to build");
@@ -335,8 +345,6 @@ mod tests {
                 .filter(|p| p.data().len() == p.actual_length() as usize)
                 .collect();
 
-            handle.interrupt();
-
             packets
         });
 
@@ -373,11 +381,10 @@ mod tests {
 
         info!("Testing against {:?}", pcap_path);
 
-        let handle = Handle::file_capture(pcap_path.to_str().expect("No path found"))
-            .expect("No handle created");
+        let mut cfg = Config::default();
+        cfg.with_interface(Interface::File(pcap_path));
 
-        let packet_stream =
-            PacketStream::new(Config::default(), Arc::clone(&handle)).expect("Failed to build");
+        let packet_stream = PacketStream::try_from(cfg).expect("Failed to build");
 
         let packet_provider = BridgeStream::new(vec![packet_stream], Duration::from_millis(100), 2)
             .expect("Failed to build");
@@ -396,11 +403,9 @@ mod tests {
                 .await
                 .into_iter()
                 .flatten()
-                .filter(|p| p.data().len() == p.actual_length() as _)
+                .filter(|p| p.data().len() == p.actual_length() as usize)
                 .count();
 
-            handle.interrupt();
-
             packets
         });
 
@@ -411,9 +416,8 @@ mod tests {
     fn packets_from_lookup_bridge() {
         let _ = env_logger::try_init();
 
-        let handle = Handle::lookup().expect("No handle created");
-        let packet_stream =
-            PacketStream::new(Config::default(), Arc::clone(&handle)).expect("Failed to build");
+        let cfg = Config::default();
+        let packet_stream = PacketStream::try_from(cfg).expect("Failed to build");
 
         let stream = BridgeStream::new(vec![packet_stream], Duration::from_millis(100), 2);
 
@@ -432,9 +436,7 @@ mod tests {
             "(not (net 172.16.0.0/16 and port 443)) and (not (host 172.17.76.33 and port 443))"
                 .to_owned(),
         );
-        let handle = Handle::lookup().expect("No handle created");
-        let packet_stream =
-            PacketStream::new(Config::default(), Arc::clone(&handle)).expect("Failed to build");
+        let packet_stream = PacketStream::try_from(cfg).expect("Failed to build");
 
         let stream = BridgeStream::new(vec![packet_stream], Duration::from_millis(100), 2);
 
@@ -444,6 +446,33 @@ mod tests {
         );
     }
 
+    #[pin_project]
+    struct IterStream {
+        inner: Vec<Packet>,
+        interrupted: AtomicBool,
+    }
+
+    impl Interruptable for IterStream {
+        fn interrupt(&self) {
+            self.interrupted.store(true, Ordering::Relaxed);
+        }
+    }
+
+    impl Stream for IterStream {
+        type Item = StreamItem;
+
+        fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+            let mut this = self;
+            if !this.interrupted.load(Ordering::Relaxed) {
+                let d = std::mem::replace(&mut this.inner, vec![]);
+                this.interrupted.store(true, Ordering::Relaxed);
+                return Poll::Ready(Some(Ok(d)));
+            } else {
+                return Poll::Ready(None);
+            }
+        }
+    }
+
     #[test]
     fn packets_come_out_time_ordered() {
         let mut packets1 = vec![];
@@ -463,18 +492,21 @@ mod tests {
             packets2.push(p)
         }
 
-        let item1: StreamItem<Error> = Ok(packets1.clone());
-        let item2: StreamItem<Error> = Ok(packets2.clone());
-
-        let stream1 = futures::stream::iter(vec![item1]);
-        let stream2 = futures::stream::iter(vec![item2]);
+        let stream1 = IterStream {
+            interrupted: AtomicBool::default(),
+            inner: packets1.clone(),
+        };
+        let stream2 = IterStream {
+            interrupted: AtomicBool::default(),
+            inner: packets2.clone(),
+        };
 
         let result = smol::block_on(async move {
             let bridge = BridgeStream::new(vec![stream1, stream2], Duration::from_millis(100), 0);
 
             let result = bridge
                 .expect("Unable to create BridgeStream")
-                .collect::<Vec<StreamItem<Error>>>()
+                .collect::<Vec<StreamItem>>()
                 .await;
             result
                 .into_iter()

src/config.rs

@@ -1,7 +1,17 @@
 use std;
+use std::path::PathBuf;
+
+#[derive(Clone, Debug)]
+pub enum Interface {
+    Dead { linktype: i32, snaplen: i32 },
+    Live(String),
+    Lookup,
+    File(PathBuf),
+}
 
 #[derive(Clone, Debug)]
 pub struct Config {
+    interface: Interface,
     max_packets_read: usize,
     snaplen: u32,
     buffer_size: u32,
@@ -12,6 +22,15 @@ pub struct Config {
 }
 
 impl Config {
+    pub fn interface(&self) -> &Interface {
+        &self.interface
+    }
+
+    pub fn with_interface(&mut self, iface: Interface) -> &mut Self {
+        self.interface = iface;
+        self
+    }
+
     pub fn max_packets_read(&self) -> usize {
         self.max_packets_read
     }
@@ -74,31 +93,12 @@ impl Config {
         self.blocking = blocking;
         self
     }
-
-    pub fn new(
-        max_packets_read: usize,
-        snaplen: u32,
-        buffer_size: u32,
-        datalink: Option<i32>,
-        bpf: Option<String>,
-        buffer_for: std::time::Duration,
-        blocking: bool,
-    ) -> Config {
-        Config {
-            max_packets_read,
-            snaplen,
-            buffer_size,
-            datalink,
-            bpf,
-            buffer_for,
-            blocking,
-        }
-    }
 }
 
 impl Default for Config {
     fn default() -> Config {
         Config {
+            interface: Interface::Lookup,
             max_packets_read: 1000,
             snaplen: 65535,
             buffer_size: 16777216,