flowscope 0.14.0

//! Cross-thread slot drain using `SlotHandle: Send + Sync` (0.12).
//!
//! ```bash
//! cargo run --features pcap,http \
//!     --example threaded_slot_drain -- tests/data/mixed_short.pcap
//! ```
//!
//! Architecture:
//!
//! - Main thread drives the capture loop: `driver.track_into(view,
//!   &mut events)` per packet. The `Driver<E>` itself remains
//!   `!Send` — its central `FlowTracker` holds `Rc<RefCell>`.
//! - Worker thread holds a clone of the typed `SlotHandle`, drains
//!   it in a tight loop, and forwards messages via an
//!   `std::sync::mpsc` channel for further processing.
//!
//! The slot handle is backed by `Arc<crossbeam_queue::SegQueue<…>>`
//! (lock-free MPMC). `Clone` hands out a competitive consumer:
//! each clone pops from the same queue; sum across all clones
//! equals total pushed. For broadcast semantics, drain into a
//! channel and fan out.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::thread;
use std::time::Duration;

use flowscope::PacketView;
use flowscope::driver::{Driver, Event, SlotMessage};
use flowscope::extract::{FiveTuple, FiveTupleKey};
use flowscope::http::{HttpMessage, HttpParser};
use flowscope::pcap::PcapFlowSource;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let path = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "tests/data/http_session.pcap".to_string());

    let mut builder = Driver::builder(FiveTuple::bidirectional());
    let mut http_slot = builder.session_on_ports(HttpParser::default(), [80, 8080]);
    let mut driver = builder.build();

    // Clone the slot handle for the worker thread.
    let drainer_handle = http_slot.clone();
    let done = Arc::new(AtomicBool::new(false));
    let done_for_worker = Arc::clone(&done);

    // Channel from worker → final consumer (could be a tokio
    // task, a writer, a metric pipeline, …).
    let (tx, rx) = mpsc::channel::<SlotMessage<HttpMessage, FiveTupleKey>>();

    let worker = thread::spawn(move || {
        let mut h = drainer_handle;
        let mut buf = Vec::new();
        while !done_for_worker.load(Ordering::Acquire) || h.pending() > 0 {
            h.drain(&mut buf);
            for m in buf.drain(..) {
                if tx.send(m).is_err() {
                    return; // consumer hung up
                }
            }
            thread::sleep(Duration::from_micros(50));
        }
    });

    // Capture loop on the main thread.
    let mut events: Vec<Event<FiveTupleKey>> = Vec::new();
    let mut flows = 0usize;
    for owned in PcapFlowSource::open(&path)?.views() {
        let owned = owned?;
        events.clear();
        driver.track_into(PacketView::from(&owned), &mut events);
        for ev in &events {
            if matches!(ev, Event::FlowStarted { .. }) {
                flows += 1;
            }
        }
    }
    // Drive the remaining lifecycle events through.
    driver.finish_into(&mut events);
    done.store(true, Ordering::Release);
    worker.join().expect("worker panicked");

    // Drain the channel — by now the worker has shut down and
    // we still have the rx end alive.
    let mut messages = 0usize;
    while let Ok(msg) = rx.try_recv() {
        let _ = msg.message; // keep the handler honest
        messages += 1;
    }
    // Also drain whatever the main-thread `http_slot` clone still
    // holds (the worker's clone shared the same Arc, but the
    // local `http_slot` is also a valid consumer).
    let mut leftover = Vec::new();
    http_slot.drain(&mut leftover);
    messages += leftover.len();

    println!("flows started: {flows}");
    println!("http messages observed: {messages}");
    Ok(())
}