traffic preview (very very ugly and WIP)

src/gui/sniffer.rs

@@ -48,6 +48,7 @@
 use crate::mmdb::types::mmdb_reader::{MmdbReader, MmdbReaders};
 use crate::networking::parse_packets::BackendTrafficMessage;
 use crate::networking::parse_packets::parse_packets;
+use crate::networking::traffic_preview::traffic_preview;
 use crate::networking::types::capture_context::{
     CaptureContext, CaptureSource, CaptureSourcePicklist, MyPcapImport,
 };
@@ -269,6 +270,12 @@ pub fn update(&mut self, message: Message) -> Task<Message> {
         match message {
             Message::StartApp(id) => {
                 self.id = id;
+                let _ = thread::Builder::new()
+                    .name("thread_traffic_preview".to_string())
+                    .spawn(move || {
+                        traffic_preview();
+                    })
+                    .log_err(location!());
                 return Task::batch([
                     Sniffer::register_sigint_handler(),
                     Task::perform(set_newer_release_status(), Message::SetNewerReleaseStatus),

src/networking/mod.rs

@@ -1,3 +1,4 @@
 pub mod manage_packets;
 pub mod parse_packets;
+pub mod traffic_preview;
 pub mod types;

src/networking/parse_packets.rs

@@ -9,6 +9,7 @@
     analyze_headers, get_address_to_lookup, get_traffic_type, is_local_connection,
     modify_or_insert_in_map,
 };
+use crate::networking::traffic_preview::DevInfo;
 use crate::networking::types::address_port_pair::AddressPortPair;
 use crate::networking::types::arp_type::ArpType;
 use crate::networking::types::bogon::is_bogon;
@@ -61,7 +62,7 @@ pub fn parse_packets(
     let (pcap_tx, pcap_rx) = std::sync::mpsc::sync_channel(10_000);
     let _ = thread::Builder::new()
         .name("thread_packet_stream".to_string())
-        .spawn(move || packet_stream(cap, &pcap_tx, &mut freeze_rx_2, &filters))
+        .spawn(move || packet_stream(cap, &pcap_tx, &mut freeze_rx_2, &filters, None))
         .log_err(location!());
 
     loop {
@@ -301,7 +302,10 @@ pub fn parse_packets(
     }
 }
 
-fn get_sniffable_headers(packet: &[u8], my_link_type: MyLinkType) -> Option<LaxPacketHeaders<'_>> {
+pub(super) fn get_sniffable_headers(
+    packet: &[u8],
+    my_link_type: MyLinkType,
+) -> Option<LaxPacketHeaders<'_>> {
     match my_link_type {
         MyLinkType::Ethernet(_) | MyLinkType::Unsupported(_) | MyLinkType::NotYetAssigned => {
             LaxPacketHeaders::from_ethernet(packet).ok()
@@ -497,11 +501,12 @@ fn maybe_send_tick_run_offline(
     }
 }
 
-fn packet_stream(
+pub(super) fn packet_stream(
     mut cap: CaptureType,
     tx: &std::sync::mpsc::SyncSender<(Result<PacketOwned, pcap::Error>, Option<pcap::Stat>)>,
     freeze_rx: &mut Receiver<()>,
     filters: &Filters,
+    dev_info: Option<DevInfo>,
 ) {
     loop {
         // check if we need to freeze the parsing
@@ -518,6 +523,7 @@ fn packet_stream(
         let packet_owned = packet_res.map(|p| PacketOwned {
             header: *p.header,
             data: p.data.into(),
+            dev_info: dev_info.clone(),
         });
         if tx.send((packet_owned, cap.stats().ok())).is_err() {
             return;
@@ -525,7 +531,8 @@ fn packet_stream(
     }
 }
 
-struct PacketOwned {
+pub(super) struct PacketOwned {
     pub header: PacketHeader,
     pub data: Box<[u8]>,
+    pub dev_info: Option<DevInfo>,
 }

src/networking/traffic_preview.rs

@@ -0,0 +1,69 @@
+use crate::gui::types::filters::Filters;
+use crate::location;
+use crate::networking::parse_packets::{get_sniffable_headers, packet_stream};
+use crate::networking::types::capture_context::{CaptureContext, CaptureSource};
+use crate::networking::types::my_device::MyDevice;
+use crate::networking::types::my_link_type::MyLinkType;
+use crate::utils::error_logger::{ErrorLogger, Location};
+use pcap::Device;
+use std::thread;
+use std::time::{Duration, Instant};
+
+pub fn traffic_preview() {
+    let (freeze_tx, mut freeze_rx) = tokio::sync::broadcast::channel(1_048_575);
+    let (pcap_tx, pcap_rx) = std::sync::mpsc::sync_channel(10_000);
+    for dev in Device::list().unwrap_or_default() {
+        let mut freeze_rx = freeze_tx.subscribe();
+        let pcap_tx = pcap_tx.clone();
+        let dev_name = dev.name.clone();
+        let my_dev = MyDevice::from_pcap_device(dev);
+        let capture_source = CaptureSource::Device(my_dev);
+        let capture_context = CaptureContext::new(&capture_source, None, &Filters::default());
+        let my_link_type = capture_context.my_link_type();
+        let dev_info = Some(DevInfo {
+            name: dev_name,
+            my_link_type,
+        });
+        let (cap, _) = capture_context.consume();
+        let _ = thread::Builder::new()
+            .name("thread_device_traffic_preview".to_string())
+            .spawn(move || {
+                packet_stream(cap, &pcap_tx, &mut freeze_rx, &Filters::default(), dev_info);
+            })
+            .log_err(location!());
+    }
+
+    loop {
+        // check if we need to freeze the parsing
+        if freeze_rx.try_recv().is_ok() {
+            // wait until unfreeze
+            let _ = freeze_rx.blocking_recv();
+            // reset the first packet ticks
+            // first_packet_ticks = Some(Instant::now());
+        }
+
+        let (packet_res, cap_stats) = pcap_rx
+            .recv_timeout(Duration::from_millis(150))
+            .unwrap_or((Err(pcap::Error::TimeoutExpired), None));
+
+        // if tx.is_closed() {
+        //     return;
+        // }
+
+        if let Ok(packet) = packet_res {
+            let my_link_type = packet.dev_info.as_ref().unwrap().my_link_type;
+            if get_sniffable_headers(&packet.data, my_link_type).is_some() {
+                println!(
+                    "Received packet on {}",
+                    packet.dev_info.as_ref().map_or("?", |d| d.name.as_str()),
+                );
+            }
+        }
+    }
+}
+
+#[derive(Clone)]
+pub(super) struct DevInfo {
+    name: String,
+    my_link_type: MyLinkType,
+}