Use std::ops::Sub for manipulating DateTime

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Author: jarhodes314

crates/extensions/tedge_flows/src/flow.rs

@@ -1,3 +1,6 @@
+use std::ops::Sub;
+use std::time::Duration;
+
 use crate::js_runtime::JsRuntime;
 use crate::js_script::JsScript;
 use crate::stats::Counter;
@@ -239,11 +242,24 @@ impl DateTime {
         let tick_every_secs = tick_every.as_secs();
         tick_every_secs != 0 && (self.seconds % tick_every_secs == 0)
     }
+}
+
+impl Sub<Duration> for DateTime {
+    type Output = DateTime;
 
-    pub fn sub_duration(&self, duration: std::time::Duration) -> Self {
-        DateTime {
-            seconds: self.seconds - duration.as_secs(),
-            nanoseconds: self.nanoseconds,
+    fn sub(self, rhs: Duration) -> Self::Output {
+        if rhs.subsec_nanos() > self.nanoseconds {
+            let seconds = self.seconds - 1;
+            let nanoseconds = self.nanoseconds + 1_000_000_000 - rhs.subsec_nanos();
+            DateTime {
+                seconds: seconds - rhs.as_secs(),
+                nanoseconds,
+            }
+        } else {
+            DateTime {
+                seconds: self.seconds - rhs.as_secs(),
+                nanoseconds: self.nanoseconds - rhs.subsec_nanos(),
+            }
         }
     }
 }

crates/extensions/tedge_flows/src/runtime.rs

@@ -214,7 +214,7 @@ impl MessageProcessor {
             } = &flow.input
             {
                 if timestamp.tick_now(*input_frequency) {
-                    let cutoff_time = timestamp.sub_duration(*input_span);
+                    let cutoff_time = timestamp - *input_span;
                     let drained_messages = self
                         .database
                         .drain_older_than(cutoff_time, input_series)