Struct FlowDriver 

pub struct FlowDriver<E, F, S = ()>
where
    E: FlowExtractor,
    F: ReassemblerFactory<E::Key>,
    S: Send + 'static,
{ /* private fields */ }

Available on crate feature reassembler only.

Sync flow driver: tracker + per-(flow, side) reassembler dispatch.

Use this when you want both flow events and TCP byte streams in one synchronous loop (typical for pcap replay, embedded use, non-tokio CLI tools). For L7 parsing on top, use crate::session::SessionDriver or the typed crate::driver::Driver, which are built on it.

The tracker config is the single source of truth for reassembly limits: the factory receives it through ReassemblerFactory::apply_config on construction and on Self::set_config.

Implementations

impl<E, F> FlowDriver<E, F, ()>

where E: FlowExtractor, F: ReassemblerFactory<E::Key>,

pub fn new(extractor: E, factory: F) -> Self

Construct with default config and S = () (no per-flow user state). Annotation-free.

pub fn with_config(extractor: E, factory: F, config: FlowTrackerConfig) -> Self

Construct with explicit config and S = ().

impl<E, F, S> FlowDriver<E, F, S>

where E: FlowExtractor, F: ReassemblerFactory<E::Key>, S: Default + Send + 'static,

pub fn with_state(extractor: E, factory: F) -> Self

Construct with default config and per-flow state initialised via S::default().

pub fn with_state_and_config( extractor: E, factory: F, config: FlowTrackerConfig, ) -> Self

Construct with explicit config and per-flow state initialised via S::default().

impl<E, F, S> FlowDriver<E, F, S>

where E: FlowExtractor, F: ReassemblerFactory<E::Key>, S: Send + 'static,

pub fn with_state_init<G>(extractor: E, factory: F, init: G) -> Self

where G: FnMut(&E::Key) -> S + Send + Sync + 'static,

Construct with default config and a custom per-flow state initialiser.

pub fn with_state_init_and_config<G>( extractor: E, factory: F, config: FlowTrackerConfig, init: G, ) -> Self

where G: FnMut(&E::Key) -> S + Send + Sync + 'static,

Construct with explicit config and a custom per-flow state initialiser.

pub fn from_tracker(tracker: FlowTracker<E, S>, factory: F) -> Self

Wrap an existing tracker (keeps its config, idle-timeout predicate and live flows).

pub fn with_emit_anomalies(self, enable: bool) -> Self

Opt in to emitting FlowEvent::FlowAnomaly / FlowEvent::TrackerAnomaly for buffer overflows, gaps, late / stray segments, retransmits, overlap inconsistencies, high-watermark crossings, memcap hits and tracker eviction pressure. Default: false — no anomaly events emitted; counters still accumulate in crate::FlowStats.

Anomalies are coalesced per (flow, side, kind) per packet / sweep so a pathological flow doesn’t swamp the stream.

pub fn with_idle_timeout_fn<G>(self, f: G) -> Self

where G: Fn(&E::Key, Option<L4Proto>) -> Option<Duration> + Send + Sync + 'static,

Set a per-key idle-timeout override on the underlying tracker (see FlowTracker::set_idle_timeout_fn).

pub fn with_dedup(self, dedup: Dedup) -> Self

Filter incoming PacketViews through a content-hash crate::Dedup before extraction. Views the dedup classifies as duplicates produce zero events; they count in Self::dedup().dropped() and in flowscope_packets_deduplicated_total. Useful for loopback captures (tcpdump -i lo) where every packet arrives twice via the kernel’s outgoing/host reinjection.

pub fn dedup(&self) -> Option<&Dedup>

Borrow the dedup state. None when no dedup is configured.

pub fn with_monotonic_timestamps(self, enable: bool) -> Self

Opt in to strictly non-decreasing timestamps across the stream. Each packet’s view.timestamp is clamped to max(view.timestamp, last_emitted_timestamp); the clamp also applies to Self::sweep’s now argument. Default: off.

pub fn set_monotonic_timestamps(&mut self, enable: bool)

In-place variant of Self::with_monotonic_timestamps.

pub fn set_emit_anomalies(&mut self, enable: bool)

In-place variant of Self::with_emit_anomalies.

pub fn set_dedup(&mut self, dedup: Option<Dedup>)

In-place variant of Self::with_dedup; None removes it.

pub fn set_config(&mut self, config: FlowTrackerConfig)

Replace the config: the tracker adopts it and the factory receives it through ReassemblerFactory::apply_config. Reassemblers already created keep their settings.

pub fn pause_events(&mut self)

Stop emitting events (the “shunt mode” of FlowTracker::pause_events); per-flow state is still released as flows end.

pub fn resume_events(&mut self)

Resume emitting events after Self::pause_events.

pub fn max_timestamp(&self) -> Timestamp

Latest packet timestamp seen — what Self::finish stamps its output with.

pub fn track<'v>( &mut self, view: impl Into<PacketView<'v>>, ) -> FlowEvents<E::Key>

Process one packet. Drives the tracker and dispatches TCP payloads to the factory’s reassemblers. Reassemblers are created on demand and cleaned up on Ended.

pub fn track_pending<'v>( &mut self, view: impl Into<PacketView<'v>>, ) -> FlowEvents<E::Key>

Lower-level: process one packet and return events without finalizing reassemblers. Reassemblers stay accessible (via Self::reassembler / Self::drain_stream) until Self::finalize is called — which is how a consumer harvests the last bytes of a flow that ends on this packet.

For every Ended event in the result, both sides’ reassemblers have already been flushed (Reassembler::flush_pending) and their final diagnostics folded into the event’s stats. An Ended the event mask suppresses is finalized internally.

You MUST call Self::finalize before the next track_pending / sweep_pending / track / sweep call. Auto-sweeps (FlowTrackerConfig::auto_sweep_interval) only run from Self::track.

pub fn track_pending_with<'v, W>( &mut self, view: impl Into<PacketView<'v>>, want: W, ) -> FlowEvents<E::Key>

where W: FnMut(&PacketContext<'_, E::Key>) -> bool,

Like Self::track_pending, but a reassembler is only created for a flow when want returns true for the packet that would create it (flows that already have stream state keep it). The session engines use it to buffer only flows some parser is interested in; answer consistently for every packet of a flow.

pub fn sweep(&mut self, now: Timestamp) -> Vec<FlowEvent<E::Key>>

Run the idle-timeout sweep and clean up reassemblers for ended flows.

pub fn finish(&mut self) -> Vec<FlowEvent<E::Key>>

Sweep every remaining flow to its end. Call once after the last track when input is exhausted. Output is stamped with the latest packet timestamp seen (never Timestamp::MAX), and the monotonic clock is left alone.

pub fn force_close( &mut self, key: &E::Key, now: Timestamp, ) -> Vec<FlowEvent<E::Key>>

Force-end the flow with this key. Returns the resulting FlowEvents (one Ended with crate::EndReason::ForceClosed); empty if key was not active.

pub fn force_close_pending( &mut self, key: &E::Key, now: Timestamp, ) -> Vec<FlowEvent<E::Key>>

Lower-level Self::force_close that leaves the flow’s reassemblers in place (flushed) so their last bytes can be drained; call Self::finalize afterwards.

pub fn sweep_pending(&mut self, now: Timestamp) -> Vec<FlowEvent<E::Key>>

Lower-level sweep variant. Like Self::sweep but does NOT finalize ended flows’ reassemblers. See Self::track_pending for the contract.

Before idling flows out, every live reassembler gets Reassembler::advance_time so holes past their deadline are skipped even on a side that went quiet (the released bytes wait in the reassembler for Self::drain_stream).

pub fn sweep_pending_drain<G>( &mut self, now: Timestamp, buf: &mut StreamChunks, f: G, ) -> Vec<FlowEvent<E::Key>>

where G: FnMut(&E::Key, FlowSide, &mut StreamChunks),

Self::sweep_pending that also hands over the bytes the sweep released: holes past their deadline are skipped by Reassembler::advance_time, and the data waiting behind them (with the Chunk::Gap marking the hole) is drained into buf and passed to f(key, side, buf) — once per side with output, buf cleared before each. Without it, a side that went quiet behind a hole only delivers at its next packet or at flow end.

Anomalies of the advanced sides are returned with the events (after the data f already saw). Same contract as Self::sweep_pending: call Self::finalize next.

pub fn finalize(&mut self, events: &mut [FlowEvent<E::Key>])

Drop the reassemblers of every flow that ended in events (calling fin / rst on the way out) and refund their bytes to the memcap pool. Called automatically by Self::track / Self::sweep; callers of the *_pending variants must call it before the next track* / sweep* call.

pub fn finalize_flow(&mut self, key: &E::Key, reason: EndReason)

Self::finalize for one ended flow: drop its reassemblers (fin for a graceful reason, rst otherwise) and refund their bytes to the memcap pool. No-op for unknown keys.

pub fn last_packet(&self) -> Option<&PacketInfo<E::Key>>

The packet most recently accepted by track* — None before the first packet, and after a packet that was deduplicated or matched no flow.

pub fn reassembly_memcap_bytes(&self) -> u64

Running total of bytes currently buffered across every live reassembler (issue #26) — what the tracker-wide memcap is compared against.

pub fn stream_count(&self) -> usize

Flows with stream state (live or tombstoned sides).

pub fn reassembler( &mut self, key: &E::Key, side: FlowSide, ) -> Option<&mut F::Reassembler>

Borrow the per-(flow, side) reassembler. None when no reassembler exists (no TCP payload seen on that side, the side was discarded / released, or the flow ended and was finalized).

pub fn drain_stream( &mut self, key: &E::Key, side: FlowSide, out: &mut StreamChunks, ) -> bool

Move the reassembled output of one flow side (bytes, gaps, stop) onto the end of out — see Reassembler::drain_into. A side stopped by the memcap reports its stop once. Returns false when there is nothing to report for the side. The memcap pool is re-synced immediately.

pub fn discard_stream(&mut self, key: &E::Key)

Stop reassembling a flow because no consumer needs its bytes any more (e.g. every parser on it has closed). Both sides drop their reassembler — including a side that has not sent data yet — and later segments are ignored. Unlike a reassembly stop, this is not reported anywhere: it is the consumer’s decision. Final counters are kept for the flow’s stats.

pub fn discard_side(&mut self, key: &E::Key, side: FlowSide)

Self::discard_stream for one side.

pub fn tracker(&self) -> &FlowTracker<E, S>

Borrow the inner tracker (for stats, introspection).

pub fn tracker_mut(&mut self) -> &mut FlowTracker<E, S>

Borrow the inner tracker mutably. Prefer the driver’s own methods: only Self::set_config keeps the reassembler factory in sync, and flows ended or forgotten directly on the tracker skip the driver’s end-of-flow flush (their stream state is dropped at the next sweep).

pub fn emits_anomalies(&self) -> bool

True when anomaly emission is on.

pub fn snapshot_flow_stats( &self, ) -> impl Iterator<Item = (E::Key, FlowStats)> + '_

Iterate (key, FlowStats) for every live flow, combining the tracker’s per-flow stats with live reassembler diagnostics (gaps, late / stray drops, oversize drops, peak watermark, retransmits, stop) so consumers get an up-to-date picture mid-flow. Lazy; each item clones the stats.

pub fn flow_stats(&self, key: &E::Key) -> Option<FlowStats>

Live stats of one flow (see Self::snapshot_flow_stats).