Deprecate adhoc DateTime in favor of std SystemTime

didier-wenzek · didier-wenzek · commit bb5f52690034 · 2025-11-19T14:35:58.000+01:00
Signed-off-by: Didier Wenzek &lt;didier.wenzek@free.fr&gt;
diff --git a/crates/core/tedge/src/cli/flows/cli.rs b/crates/core/tedge/src/cli/flows/cli.rs
@@ -7,9 +7,9 @@ use anyhow::anyhow;
 use anyhow::Context;
 use anyhow::Error;
 use camino::Utf8PathBuf;
+use std::time::SystemTime;
 use tedge_config::TEdgeConfig;
 use tedge_flows::BaseFlowRegistry;
-use tedge_flows::DateTime;
 use tedge_flows::Message;
 use tedge_flows::MessageProcessor;
 
@@ -50,8 +50,8 @@ pub enum TEdgeFlowsCli {
 
         /// Processing time to be used for the test
         #[clap(long = "processing-time")]
-        #[arg(value_parser = parse_date)]
-        processing_time: Option<DateTime>,
+        #[arg(value_parser = humantime::parse_rfc3339_weak)]
+        processing_time: Option<SystemTime>,
 
         /// The input payloads are base64 encoded and have to be decoded first
         #[clap(long = "base64-input")]
@@ -145,9 +145,3 @@ impl TEdgeFlowsCli {
         Ok(processor)
     }
 }
-
-fn parse_date(src: &str) -> Result<DateTime, String> {
-    let time = humantime::parse_rfc3339_weak(src)
-        .map_err(|e| format!("Unable to parse RFC3339 date: {e}"))?;
-    time.try_into()
-}
diff --git a/crates/core/tedge/src/cli/flows/test.rs b/crates/core/tedge/src/cli/flows/test.rs
@@ -5,9 +5,9 @@ use anyhow::Error;
 use base64::prelude::BASE64_STANDARD;
 use base64::prelude::*;
 use camino::Utf8PathBuf;
+use std::time::SystemTime;
 use tedge_config::TEdgeConfig;
 use tedge_flows::BaseFlowRegistry;
-use tedge_flows::DateTime;
 use tedge_flows::FlowResult;
 use tedge_flows::Message;
 use tedge_flows::MessageProcessor;
@@ -22,7 +22,7 @@ pub struct TestCommand {
     pub flow: Option<Utf8PathBuf>,
     pub message: Option<Message>,
     pub final_on_interval: bool,
-    pub processing_time: Option<DateTime>,
+    pub processing_time: Option<SystemTime>,
     pub base64_input: bool,
     pub base64_output: bool,
 }
@@ -42,18 +42,18 @@ impl Command for TestCommand {
             Some(flow) => TEdgeFlowsCli::load_file(&self.flows_dir, flow).await?,
         };
         if let Some(message) = &self.message {
-            let timestamp = self.processing_time.unwrap_or_else(|| DateTime::now());
+            let timestamp = self.processing_time.unwrap_or_else(SystemTime::now);
             self.process(&mut processor, message.clone(), timestamp)
                 .await;
         } else {
             let mut stdin = BufReader::new(tokio::io::stdin());
             while let Some(message) = next_message(&mut stdin).await {
-                let timestamp = self.processing_time.unwrap_or_else(|| DateTime::now());
+                let timestamp = self.processing_time.unwrap_or_else(SystemTime::now);
                 self.process(&mut processor, message, timestamp).await;
             }
         }
         if self.final_on_interval {
-            let timestamp = DateTime::now();
+            let timestamp = SystemTime::now();
             let now = processor
                 .last_interval_deadline()
                 .unwrap_or_else(Instant::now);
@@ -68,7 +68,7 @@ impl TestCommand {
         &self,
         processor: &mut MessageProcessor<BaseFlowRegistry>,
         mut message: Message,
-        timestamp: DateTime,
+        timestamp: SystemTime,
     ) {
         if self.base64_input {
             match BASE64_STANDARD.decode(message.payload) {
@@ -90,7 +90,7 @@ impl TestCommand {
     async fn tick(
         &self,
         processor: &mut MessageProcessor<BaseFlowRegistry>,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) {
         processor
diff --git a/crates/extensions/tedge_flows/src/actor.rs b/crates/extensions/tedge_flows/src/actor.rs
@@ -1,5 +1,4 @@
 use crate::connected_flow::ConnectedFlowRegistry;
-use crate::flow::DateTime;
 use crate::flow::FlowError;
 use crate::flow::FlowOutput;
 use crate::flow::FlowResult;
@@ -16,6 +15,7 @@ use serde_json::json;
 use std::cmp::min;
 use std::collections::HashSet;
 use std::time::Duration;
+use std::time::SystemTime;
 use tedge_actors::Actor;
 use tedge_actors::DynSender;
 use tedge_actors::MessageReceiver;
@@ -207,7 +207,7 @@ impl FlowsMapper {
 
     async fn on_source_poll(&mut self) -> Result<(), RuntimeError> {
         let now = Instant::now();
-        let timestamp = DateTime::now();
+        let timestamp = SystemTime::now();
 
         let mut in_messages = vec![];
         for flow in self.processor.registry.flows_mut() {
@@ -241,7 +241,7 @@ impl FlowsMapper {
         source: SourceTag,
         message: Message,
     ) -> Result<(), RuntimeError> {
-        let timestamp = DateTime::now();
+        let timestamp = SystemTime::now();
         for messages in self
             .processor
             .on_message(timestamp, &source, &message)
@@ -255,7 +255,7 @@ impl FlowsMapper {
 
     async fn on_interval(&mut self) -> Result<(), RuntimeError> {
         let now = Instant::now();
-        let timestamp = DateTime::now();
+        let timestamp = SystemTime::now();
         if self.next_dump <= now {
             self.processor.dump_memory_stats().await;
             self.processor.dump_processing_stats().await;
@@ -295,7 +295,7 @@ impl FlowsMapper {
     ) -> Result<(), RuntimeError> {
         if let Some(flow) = self.processor.registry.flow(&flow_name) {
             let topic = flow.input_topic().to_string();
-            let timestamp = DateTime::now();
+            let timestamp = SystemTime::now();
             let message = Message::new(topic, line);
             if let Some(result) = self
                 .processor
diff --git a/crates/extensions/tedge_flows/src/connected_flow.rs b/crates/extensions/tedge_flows/src/connected_flow.rs
@@ -1,5 +1,4 @@
 use crate::config::ConfigError;
-use crate::flow::DateTime;
 use crate::flow::Flow;
 use crate::flow::FlowError;
 use crate::flow::FlowInput;
@@ -14,6 +13,7 @@ use crate::input_source::StreamingSource;
 use crate::registry::FlowRegistry;
 use crate::registry::FlowStore;
 use camino::Utf8Path;
+use std::time::SystemTime;
 use tedge_watch_ext::WatchRequest;
 use tokio::time::Instant;
 
@@ -65,14 +65,14 @@ impl ConnectedFlow {
         self.polling_source.as_ref().and_then(|p| p.next_deadline())
     }
 
-    pub async fn on_source_poll(&mut self, timestamp: DateTime, now: Instant) -> FlowResult {
+    pub async fn on_source_poll(&mut self, timestamp: SystemTime, now: Instant) -> FlowResult {
         let result = self.on_source_poll_steps(timestamp, now).await;
         self.flow.publish(result)
     }
 
     async fn on_source_poll_steps(
         &mut self,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) -> Result<Vec<Message>, FlowError> {
         let Some(source) = &mut self.polling_source.as_mut() else {
diff --git a/crates/extensions/tedge_flows/src/flow.rs b/crates/extensions/tedge_flows/src/flow.rs
@@ -15,7 +15,6 @@ use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::Topic;
 use tedge_mqtt_ext::TopicFilter;
 use tedge_watch_ext::WatchError;
-use time::OffsetDateTime;
 use tokio::time::Instant;
 use tracing::error;
 use tracing::warn;
@@ -113,17 +112,11 @@ impl FlowResult {
     }
 }
 
-#[derive(Copy, Clone, Debug, serde::Deserialize, serde::Serialize, Eq, PartialEq)]
-pub struct DateTime {
-    pub seconds: u64,
-    pub nanoseconds: u32,
-}
-
 #[derive(Clone, serde::Deserialize, serde::Serialize, Eq, PartialEq)]
 pub struct Message {
     pub topic: String,
     pub payload: Vec<u8>,
-    pub timestamp: Option<DateTime>,
+    pub timestamp: Option<SystemTime>,
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -199,7 +192,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         message: &Message,
     ) -> FlowResult {
         let stated_at = stats.flow_on_message_start(self.name());
@@ -219,7 +212,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         message: &Message,
     ) -> Result<Vec<Message>, FlowError> {
         let mut messages = vec![message.clone()];
@@ -247,7 +240,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) -> FlowResult {
         let stated_at = stats.flow_on_interval_start(self.name());
@@ -267,7 +260,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) -> Result<Vec<Message>, FlowError> {
         let mut messages = vec![];
@@ -400,45 +393,6 @@ impl FlowStep {
     }
 }
 
-impl DateTime {
-    pub fn now() -> Self {
-        DateTime::try_from(OffsetDateTime::now_utc()).unwrap()
-    }
-
-    pub fn json(&self) -> Value {
-        json!({"seconds": self.seconds, "nanoseconds": self.nanoseconds})
-    }
-}
-
-impl TryFrom<SystemTime> for DateTime {
-    type Error = String;
-
-    fn try_from(time: SystemTime) -> Result<Self, Self::Error> {
-        match time.duration_since(SystemTime::UNIX_EPOCH) {
-            Ok(elapsed) => Ok(DateTime {
-                seconds: elapsed.as_secs(),
-                nanoseconds: 0,
-            }),
-            Err(_) => Err("SystemTime before UNIX EPOCH!".to_string()),
-        }
-    }
-}
-
-impl TryFrom<OffsetDateTime> for DateTime {
-    type Error = FlowError;
-
-    fn try_from(value: OffsetDateTime) -> Result<Self, Self::Error> {
-        let seconds = u64::try_from(value.unix_timestamp()).map_err(|err| {
-            FlowError::UnsupportedMessage(format!("failed to convert timestamp: {}", err))
-        })?;
-
-        Ok(DateTime {
-            seconds,
-            nanoseconds: value.nanosecond(),
-        })
-    }
-}
-
 impl Message {
     pub fn new(topic: impl ToString, payload: impl Into<Vec<u8>>) -> Self {
         Message {
@@ -451,7 +405,7 @@ impl Message {
     pub fn with_timestamp(
         topic: impl ToString,
         payload: impl Into<Vec<u8>>,
-        timestamp: DateTime,
+        timestamp: SystemTime,
     ) -> Self {
         Message {
             topic: topic.to_string(),
@@ -462,7 +416,7 @@ impl Message {
 
     pub fn json(&self) -> Value {
         if let Some(timestamp) = &self.timestamp {
-            json!({"topic": self.topic, "payload": self.payload, "timestamp": timestamp.json()})
+            json!({"topic": self.topic, "payload": self.payload, "timestamp": epoch_ms(timestamp)})
         } else {
             json!({"topic": self.topic, "payload": self.payload, "timestamp": null})
         }
@@ -512,6 +466,13 @@ impl TryFrom<Message> for MqttMessage {
     }
 }
 
+pub(crate) fn epoch_ms(time: &SystemTime) -> u128 {
+    let duration = time
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .expect("SystemTime after UNIX EPOCH");
+    duration.as_millis()
+}
+
 pub fn error_from_js(err: LoadError) -> FlowError {
     FlowError::IncorrectSetting(format!("{err:#}"))
 }
diff --git a/crates/extensions/tedge_flows/src/input_source.rs b/crates/extensions/tedge_flows/src/input_source.rs
@@ -1,8 +1,8 @@
-use crate::flow::DateTime;
 use crate::flow::Message;
 use async_trait::async_trait;
 use camino::Utf8PathBuf;
 use std::time::Duration;
+use std::time::SystemTime;
 use tedge_watch_ext::WatchRequest;
 use tokio::time::Instant;
 
@@ -16,7 +16,7 @@ pub trait StreamingSource: Send + Sync {
 #[async_trait]
 pub trait PollingSource: Send + Sync {
     /// Poll the source for any available messages at the given timestamp
-    async fn poll(&mut self, timestamp: DateTime) -> Result<Vec<Message>, PollingSourceError>;
+    async fn poll(&mut self, timestamp: SystemTime) -> Result<Vec<Message>, PollingSourceError>;
 
     /// Get the next deadline when this source should be polled
     /// Returns None if the source doesn't have scheduled polling
@@ -53,7 +53,7 @@ impl CommandPollingSource {
 
 #[async_trait]
 impl PollingSource for CommandPollingSource {
-    async fn poll(&mut self, timestamp: DateTime) -> Result<Vec<Message>, PollingSourceError> {
+    async fn poll(&mut self, timestamp: SystemTime) -> Result<Vec<Message>, PollingSourceError> {
         let output = tedge_watch_ext::command_output(&self.command)
             .await
             .map_err(|err| PollingSourceError::CannotPoll {
@@ -119,7 +119,7 @@ impl FilePollingSource {
 
 #[async_trait]
 impl PollingSource for FilePollingSource {
-    async fn poll(&mut self, timestamp: DateTime) -> Result<Vec<Message>, PollingSourceError> {
+    async fn poll(&mut self, timestamp: SystemTime) -> Result<Vec<Message>, PollingSourceError> {
         let output = tokio::fs::read_to_string(&self.path).await.map_err(|err| {
             PollingSourceError::CannotPoll {
                 resource: self.path.clone().to_string(),
diff --git a/crates/extensions/tedge_flows/src/js_script.rs b/crates/extensions/tedge_flows/src/js_script.rs
diff --git a/crates/extensions/tedge_flows/src/js_value.rs b/crates/extensions/tedge_flows/src/js_value.rs
diff --git a/crates/extensions/tedge_flows/src/runtime.rs b/crates/extensions/tedge_flows/src/runtime.rs
