rlink 0.6.16

High performance Stream Processing Framework
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

use std::time::Duration;

use crate::core::element::Record;
use crate::core::watermark::{Watermark, WatermarkGenerator, MIN_WATERMARK};

#[derive(Debug)]
pub struct BoundedOutOfOrdernessWatermarks {
    max_timestamp: u64,
    out_of_orderness_millis: u64,
}

impl BoundedOutOfOrdernessWatermarks {
    pub fn new(out_of_orderness_millis: Duration) -> Self {
        let out_of_orderness_millis = out_of_orderness_millis.as_millis() as u64;

        BoundedOutOfOrdernessWatermarks {
            max_timestamp: 0,
            out_of_orderness_millis,
        }
    }
}

impl WatermarkGenerator for BoundedOutOfOrdernessWatermarks {
    fn on_event(&mut self, _record: &mut Record, event_timestamp: u64) -> Option<Watermark> {
        if event_timestamp > self.max_timestamp {
            self.max_timestamp = event_timestamp;
        }
        None
    }

    fn on_periodic_emit(&mut self) -> Option<Watermark> {
        let timestamp = if self.max_timestamp > self.out_of_orderness_millis {
            self.max_timestamp - self.out_of_orderness_millis - 1
        } else {
            MIN_WATERMARK.timestamp
        };
        Some(Watermark::new(timestamp))
    }
}