Struct MergedFlowStream 

pub struct MergedFlowStream<C, E>
where
    E: FlowExtractor,
{ /* private fields */ }

Source-agnostic tap merge: N captures feeding one shared FlowTracker, keyed by the bare bidirectional flow key.

Unlike MultiFlowStream (which builds one tracker per source and tags each event with its source_idx), MergedFlowStream coalesces the two directions of a tapped flow — TX on eth0, RX on eth1 — into one bidirectional flow. flowscope’s bidirectional 5-tuple canonicalizes endpoints by address order, so the a→b and b→a legs hash to the same key with opposite Orientation and coalesce by construction when they share a tracker. Yields a plain FlowEvent<E::Key> (no source_idx envelope — the whole point).

Each source’s packets are stamped with a 1-based with_source_idx(i + 1) (0 is flowscope’s “unused” sentinel) before tracking, so the merged flow reports which physical leg each canonical direction arrived on via FlowStats::source_idx_forward / source_idx_reverse, plus FlowStats::capture_leg_inconsistent — the tap-miswire / asymmetric-routing IOC (RFC 5103 biflow merge, flowscope #120). These ride on Ended / Tick events and snapshot_flow_stats.

Use a bidirectional extractor (e.g. FiveTuple::bidirectional()) — a per-direction key would defeat the merge. Pairs with MonitorBuilder::infer_tcp_initiator / MultiStreamConfig::with_infer_tcp_initiator for race-robust TCP roles across the two legs. See docs/scaling.md → “merge vs distinct”.

Implementations

impl<C, E> MergedFlowStream<C, E>

where E: FlowExtractor + Unpin + Send + 'static, E::Key: Clone + Unpin + Send + 'static,

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

Borrow the single shared tracker (stats / snapshot_flow_stats / live capture-leg introspection).

pub fn tracker_stats(&self) -> &FlowTrackerStats

Cumulative counters of the shared tracker.

pub fn active_flows(&self) -> usize

Live flow count in the shared tracker (each tapped flow counts once, not once per leg). O(n) walk.

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

Borrow-iterator over live (K, FlowStats) pairs of the merged tracker — the capture-leg fields (source_idx_forward / source_idx_reverse / capture_leg_inconsistent) are readable here mid-stream.

pub fn sources(&self) -> usize

Number of sources fed into the merge.

pub fn label(&self, source: usize) -> Option<&str>

Human-readable label for a 0-based source index (the index is source_idx - 1, since legs are stamped 1-based).

impl<E> MergedFlowStream<AsyncCapture<Capture>, E>

where E: FlowExtractor + Unpin + Send + 'static, E::Key: Clone + Unpin + Send + 'static,

pub fn per_source_capture_stats( &self, ) -> Vec<(String, Result<CaptureStats, Error>)>

Per-source kernel ring stats, in registration order.

pub fn capture_stats(&self) -> CaptureStats

Aggregate kernel ring stats across all sources. Err from any one source is skipped.

impl<E> MergedFlowStream<AsyncXdpCapture, E>

where E: FlowExtractor + Unpin + Send + 'static, E::Key: Clone + Unpin + Send + 'static,

pub fn per_source_capture_stats( &self, ) -> Vec<(String, Result<CaptureStats, Error>)>

Per-source AF_XDP ring stats, in registration order.

pub fn capture_stats(&self) -> CaptureStats

Aggregate AF_XDP ring stats across all sources. Err from any one source is skipped.

Trait Implementations

impl<C, E> Stream for MergedFlowStream<C, E>

where C: AsyncFlowSource + Unpin, E: FlowExtractor + Unpin, E::Key: Clone + Unpin,

type Item = Result<FlowEvent<<E as FlowExtractor>::Key>, Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.