aw_transform/
heartbeat.rs

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
use aw_models::Event;

/// Returns a merged event if two events have the same data and are within the pulsetime
///
/// # Example
///
/// ```ignore
/// pulsetime: 1 second (one space)
/// input:  [a] [a]  [a][b]
/// output: [a    ]  [a][b]
/// ```
pub fn heartbeat(last_event: &Event, heartbeat: &Event, pulsetime: f64) -> Option<Event> {
    // Verify that data is the same
    if heartbeat.data != last_event.data {
        return None;
    }

    let last_event_endtime = last_event.calculate_endtime();
    let heartbeat_endtime = heartbeat.calculate_endtime();

    // Verify that timestamps intersect (including pulsetime)
    let pulsetime_ns: i64 = (pulsetime * 1_000_000_000.0).round() as i64;
    let last_endtime_allowed = last_event_endtime + chrono::Duration::nanoseconds(pulsetime_ns);
    if last_event.timestamp > heartbeat.timestamp {
        return None;
    }
    if heartbeat.timestamp > last_endtime_allowed {
        return None;
    }

    let starttime = if heartbeat.timestamp < last_event.timestamp {
        &heartbeat.timestamp
    } else {
        &last_event.timestamp
    };

    let endtime = if last_event_endtime > heartbeat_endtime {
        &last_event_endtime
    } else {
        &heartbeat_endtime
    };

    let duration = endtime.signed_duration_since(*starttime);
    if duration.num_nanoseconds().unwrap() < 0 {
        debug!("Merging heartbeats would result in a negative duration, refusing to merge!");
        return None;
    }

    // Success, return successful heartbeat last_event
    Some(Event {
        id: None,
        timestamp: *starttime,
        duration,
        data: last_event.data.clone(),
    })
}

#[cfg(test)]
mod tests {
    use chrono::Duration;
    use chrono::Utc;
    use serde_json::json;

    use aw_models::Event;

    use super::heartbeat;

    #[test]
    fn test_heartbeat_pulsetime() {
        let now = Utc::now();
        let event1 = Event {
            id: None,
            timestamp: now,
            duration: Duration::seconds(1),
            data: json_map! {"test": json!(1)},
        };
        let heartbeat1 = Event {
            id: None,
            timestamp: now + Duration::seconds(2),
            duration: Duration::seconds(1),
            data: json_map! {"test": json!(1)},
        };

        // Merge result
        let res_merge = heartbeat(&event1, &heartbeat1, 2.0).unwrap();
        assert_eq!(res_merge.timestamp, now);
        assert_eq!(res_merge.duration, Duration::seconds(3));
        assert_eq!(res_merge.data, event1.data);

        // No merge result
        let res_no_merge = heartbeat(&event1, &heartbeat1, 0.0);
        assert!(res_no_merge.is_none());

        // TODO: needs more tests!
    }

    #[test]
    fn test_heartbeat_data() {
        let now = Utc::now();
        let event = Event {
            id: None,
            timestamp: now.clone(),
            duration: Duration::seconds(0),
            data: json_map! {"test": json!(1)},
        };
        let heartbeat_same_data = Event {
            id: None,
            timestamp: now.clone(),
            duration: Duration::seconds(1),
            data: json_map! {"test": json!(1)},
        };

        // Data is same, should merge
        let res_merge = heartbeat(&event, &heartbeat_same_data, 1.0);
        assert!(res_merge.is_some());

        let heartbeat_different_data = Event {
            id: None,
            timestamp: now.clone(),
            duration: Duration::seconds(1),
            data: json_map! {"test": json!(2)},
        };
        // Data is different, should not merge
        let res_merge = heartbeat(&event, &heartbeat_different_data, 1.0);
        assert!(res_merge.is_none());
    }

    #[test]
    fn test_heartbeat_same_timestamp() {
        let now = Utc::now();
        let event = Event {
            id: None,
            timestamp: now.clone(),
            duration: Duration::seconds(0),
            data: json_map! {"test": json!(1)},
        };
        let heartbeat_same_data = Event {
            id: None,
            timestamp: now.clone(),
            duration: Duration::seconds(1),
            data: json_map! {"test": json!(1)},
        };

        // Should merge
        let res_merge = heartbeat(&event, &heartbeat_same_data, 1.0).unwrap();
        assert_eq!(Duration::seconds(1), res_merge.duration);

        // Order shouldn't matter, should merge anyway
        let res_merge = heartbeat(&heartbeat_same_data, &event, 1.0).unwrap();
        assert_eq!(Duration::seconds(1), res_merge.duration);
    }
}