Struct WatermarkTracker 

pub struct WatermarkTracker { /* private fields */ }

Tracks watermarks across multiple input sources.

For operators with multiple inputs (e.g., joins, unions), the combined watermark is the minimum across all sources. This ensures no late events from any source are missed.

Example

use laminar_core::time::{WatermarkTracker, Watermark};

let mut tracker = WatermarkTracker::new(3); // 3 sources

// Source 0 advances to 1000
let wm = tracker.update_source(0, 1000);
assert_eq!(wm, None); // Other sources still at MIN

// Source 1 advances to 2000
tracker.update_source(1, 2000);

// Source 2 advances to 500
let wm = tracker.update_source(2, 500);
assert_eq!(wm, Some(Watermark::new(500))); // Min of all sources

Implementations

impl WatermarkTracker

pub fn new(num_sources: usize) -> Self

Creates a new tracker for the specified number of sources.

pub fn with_idle_timeout(num_sources: usize, idle_timeout: Duration) -> Self

Creates a new tracker with a custom idle timeout.

pub fn update_source( &mut self, source_id: usize, watermark: i64, ) -> Option<Watermark>

Updates the watermark for a specific source.

Returns Some(Watermark) if the combined watermark advances.

pub fn mark_idle(&mut self, source_id: usize) -> Option<Watermark>

Marks a source as idle, excluding it from watermark calculation.

Idle sources don’t hold back the combined watermark.

pub fn check_idle_sources(&mut self) -> Option<Watermark>

Checks for sources that have been idle longer than the timeout.

Should be called periodically to detect stalled sources.

pub fn current_watermark(&self) -> Option<Watermark>

Returns the current combined watermark.

pub fn source_watermark(&self, source_id: usize) -> Option<i64>

Returns the watermark for a specific source.

pub fn is_idle(&self, source_id: usize) -> bool

Returns whether a source is marked as idle.

pub fn num_sources(&self) -> usize

Returns the number of sources being tracked.

pub fn active_source_count(&self) -> usize

Returns the number of active (non-idle) sources.

Trait Implementations

impl Debug for WatermarkTracker

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.