Merge pull request astarte-platform#439 from rgallor/refactor/timestamp

refactor(timestamp): reflect proto timestamp changes

Author: joshuachp

Cargo.lock

@@ -134,8 +134,8 @@ rustc-args = ["--cfg=docsrs"]
 [workspace.dependencies]
 astarte-device-sdk = { path = "./", version = "=0.9.6" }
 astarte-device-sdk-derive = { version = "=0.9.6", path = "./astarte-device-sdk-derive" }
-astarte-message-hub-proto = { git = "https://github.com/astarte-platform/astarte-message-hub-proto", rev = "29c9d6f99c2c1fa004a53d51f287d7e0ba342137" }
-astarte-message-hub-proto-mock = { git = "https://github.com/astarte-platform/astarte-message-hub-proto", rev = "29c9d6f99c2c1fa004a53d51f287d7e0ba342137" }
+astarte-message-hub-proto = { git = "https://github.com/astarte-platform/astarte-message-hub-proto", rev = "8e8198de9589c606e63ada8ec104bace81ae9cfa" }
+astarte-message-hub-proto-mock = { git = "https://github.com/astarte-platform/astarte-message-hub-proto", rev = "8e8198de9589c606e63ada8ec104bace81ae9cfa" }
 async-trait = "0.1.67"
 base64 = "0.22.0"
 bson = "2.7.0"

Cargo.toml

@@ -229,38 +229,40 @@ impl TryFrom<MessageHubEvent> for ReceivedEvent<GrpcPayload> {
             .payload
             .ok_or(MessageHubProtoError::ExpectedField("payload"))?;
 
-        let timestamp = message.timestamp.map(convert_timestamp).transpose()?;
-
         Ok(ReceivedEvent {
             interface: message.interface_name,
             path: message.path,
-            payload: GrpcPayload::new(payload, timestamp),
+            payload: GrpcPayload::new(payload),
         })
     }
 }
 
 impl From<ValidatedIndividual> for astarte_message_hub_proto::AstarteMessage {
     fn from(value: ValidatedIndividual) -> Self {
-        let timestamp = value.timestamp.map(|t| t.into());
+        let timestamp = value
+            .timestamp
+            .map(astarte_message_hub_proto::pbjson_types::Timestamp::from);
 
         let payload = Some(ProtoPayload::DatastreamIndividual(
             AstarteDatastreamIndividual {
                 data: Some(value.data.into()),
+                timestamp,
             },
         ));
 
         astarte_message_hub_proto::AstarteMessage {
             interface_name: value.interface,
             path: value.path,
-            timestamp,
             payload,
         }
     }
 }
 
 impl From<ValidatedObject> for astarte_message_hub_proto::AstarteMessage {
     fn from(value: ValidatedObject) -> Self {
-        let timestamp = value.timestamp.map(|t| t.into());
+        let timestamp = value
+            .timestamp
+            .map(astarte_message_hub_proto::pbjson_types::Timestamp::from);
 
         let data = value
             .data
@@ -269,13 +271,12 @@ impl From<ValidatedObject> for astarte_message_hub_proto::AstarteMessage {
             .collect();
 
         let payload = Some(ProtoPayload::DatastreamObject(
-            astarte_message_hub_proto::AstarteDatastreamObject { data },
+            astarte_message_hub_proto::AstarteDatastreamObject { data, timestamp },
         ));
 
         astarte_message_hub_proto::AstarteMessage {
             interface_name: value.interface,
             path: value.path,
-            timestamp,
             payload,
         }
     }
@@ -286,7 +287,6 @@ impl From<ValidatedUnset> for astarte_message_hub_proto::AstarteMessage {
         Self {
             interface_name: value.interface,
             path: value.path,
-            timestamp: None,
             payload: Some(ProtoPayload::PropertyIndividual(
                 astarte_message_hub_proto::AstartePropertyIndividual { data: None },
             )),
@@ -345,16 +345,18 @@ impl TryFrom<ProtoPayload> for Value {
             // Individual
             ProtoPayload::DatastreamIndividual(AstarteDatastreamIndividual {
                 data: Some(data),
+                timestamp: _,
             })
             | ProtoPayload::PropertyIndividual(AstartePropertyIndividual { data: Some(data) }) => {
                 let value = data.try_into()?;
 
                 Ok(Value::Individual(value))
             }
             // Individual error case
-            ProtoPayload::DatastreamIndividual(AstarteDatastreamIndividual { data: None }) => {
-                Err(MessageHubProtoError::ExpectedField("data"))
-            }
+            ProtoPayload::DatastreamIndividual(AstarteDatastreamIndividual {
+                data: None,
+                timestamp: _,
+            }) => Err(MessageHubProtoError::ExpectedField("data")),
             // Object
             ProtoPayload::DatastreamObject(object) => {
                 let value = AstarteObject::try_from(object)?;
@@ -369,65 +371,57 @@ impl TryFrom<ProtoPayload> for Value {
 // because the AstarteMessage is from the proto crate, while the AstarteDeviceDataEvent is ours.
 impl From<DeviceEvent> for astarte_message_hub_proto::AstarteMessage {
     fn from(value: DeviceEvent) -> Self {
-        let payload: ProtoPayload = value.data.into();
-
-        astarte_message_hub_proto::AstarteMessage {
-            interface_name: value.interface,
-            path: value.path,
-            timestamp: None,
-            payload: Some(payload),
-        }
-    }
-}
-
-// TODO this is incomplete, no way of creating a PropertyIndividual with a set value
-// refactor value to include a property type or find a way of stashing that information
-// somewhere
-impl From<Value> for ProtoPayload {
-    fn from(value: Value) -> Self {
-        match value {
+        let payload = match value.data {
             Value::Individual(val) => {
                 ProtoPayload::DatastreamIndividual(AstarteDatastreamIndividual {
                     data: Some(val.into()),
+                    // FIXME: how do i retrieve the timestamp if the Value type doesn't have it?
+                    timestamp: None,
                 })
             }
             Value::Object(val) => {
                 let data = val.inner.into_iter().map(|(k, v)| (k, v.into())).collect();
 
                 ProtoPayload::DatastreamObject(astarte_message_hub_proto::AstarteDatastreamObject {
                     data,
+                    // FIXME: how do i retrieve the timestamp if the Value type doesn't have it?
+                    timestamp: None,
                 })
             }
             Value::Unset => ProtoPayload::PropertyIndividual(
                 astarte_message_hub_proto::AstartePropertyIndividual { data: None },
             ),
+        };
+
+        astarte_message_hub_proto::AstarteMessage {
+            interface_name: value.interface,
+            path: value.path,
+            payload: Some(payload),
         }
     }
 }
 
 #[cfg(test)]
 pub(crate) mod test {
+    use std::collections::HashMap;
+
     use astarte_message_hub_proto::{
         AstarteData, AstarteDatastreamObject, AstarteMessage, AstartePropertyIndividual,
         InterfaceProperties,
     };
     use chrono::Utc;
     use pretty_assertions::assert_eq;
 
-    use crate::Timestamp;
-
     use super::*;
 
     pub(crate) fn new_astarte_message(
         interface_name: String,
         path: String,
-        timestamp: Option<Timestamp>,
         payload: ProtoPayload,
     ) -> AstarteMessage {
         AstarteMessage {
             interface_name,
             path,
-            timestamp: timestamp.map(pbjson_types::Timestamp::from),
             payload: Some(payload),
         }
     }
@@ -482,12 +476,12 @@ pub(crate) mod test {
         let payload: ProtoPayload =
             ProtoPayload::DatastreamIndividual(AstarteDatastreamIndividual {
                 data: Some(astarte_type.into()),
+                timestamp: None,
             });
 
         let astarte_message = AstarteMessage {
             interface_name: interface_name.clone(),
             path: interface_path.clone(),
-            timestamp: None,
             payload: Some(payload),
         };
 
@@ -504,13 +498,12 @@ pub(crate) mod test {
         let interface_name = "test.name.json".to_string();
         let interface_path = "test".to_string();
 
-        let astarte_type = Value::Unset;
-        let payload: ProtoPayload = astarte_type.into();
+        let payload: ProtoPayload =
+            ProtoPayload::PropertyIndividual(AstartePropertyIndividual { data: None });
 
         let astarte_message = AstarteMessage {
             interface_name: interface_name.clone(),
             path: interface_path.clone(),
-            timestamp: None,
             payload: Some(payload),
         };
 
@@ -971,6 +964,7 @@ pub(crate) mod test {
         let payload: ProtoPayload =
             ProtoPayload::DatastreamIndividual(AstarteDatastreamIndividual {
                 data: Some(astarte_sdk_type_double.into()),
+                timestamp: None,
             });
 
         let double_value = take_individual(payload)
@@ -997,13 +991,20 @@ pub(crate) mod test {
 
     #[test]
     fn from_sdk_astarte_aggregate_to_astarte_message_payload_success() {
+        use astarte_message_hub_proto::astarte_data::AstarteData as ProtoData;
+        use astarte_message_hub_proto::AstarteData;
+
         let expected_data: f64 = 15.5;
-        let astarte_type_map = Value::Object(AstarteObject::from_iter([(
-            "key1".to_string(),
-            AstarteType::Double(expected_data),
-        )]));
 
-        let payload_result: ProtoPayload = astarte_type_map.into();
+        let payload_result = ProtoPayload::DatastreamObject(AstarteDatastreamObject {
+            data: HashMap::from([(
+                "key1".to_string(),
+                AstarteData {
+                    astarte_data: Some(ProtoData::Double(expected_data)),
+                },
+            )]),
+            timestamp: None,
+        });
 
         let double_data = take_object(payload_result)
             .and_then(|mut obj| obj.data.remove("key1"))

src/transport/grpc/convert.rs

@@ -458,30 +458,30 @@ where
         mapping: &MappingRef<'_, &Interface>,
         payload: Self::Payload,
     ) -> Result<Option<(AstarteType, Option<Timestamp>)>, TransportError> {
-        let Self::Payload { data, timestamp } = payload;
+        let value: Value = payload.data.try_into().map_err(RecvError::connection)?;
 
-        let value: Value = data.try_into().map_err(RecvError::connection)?;
+        match value {
+            Value::Individual(astarte_type) => {
+                trace!("received {}", astarte_type.display_type());
+
+                // FIXME: replace None with the actual timestamp
+                Ok(Some((astarte_type, None)))
+            }
 
-        let astarte_type = match value {
-            Value::Individual(astarte_type) => Ok(astarte_type),
             Value::Object(_hash_map) => {
                 let aggr_err = AggregateError::for_payload(
                     mapping.interface().interface_name(),
                     mapping.path().to_string(),
                     Aggregation::Individual,
                     Aggregation::Object,
                 );
-                Err(RecvError::Aggregation(aggr_err))
+                Err(RecvError::Aggregation(aggr_err).into())
             }
             Value::Unset => {
                 debug!("unset received");
-                return Ok(None);
+                Ok(None)
             }
-        }?;
-
-        trace!("received {}", astarte_type.display_type());
-
-        Ok(Some((astarte_type, timestamp)))
+        }
     }
 
     fn deserialize_object(
@@ -505,7 +505,8 @@ where
 
         trace!("object received");
 
-        Ok((data, payload.timestamp))
+        // FIXME: replace None with the actual timestamp
+        Ok((data, None))
     }
 }
 
@@ -551,12 +552,11 @@ where
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct GrpcPayload {
     data: ProtoPayload,
-    timestamp: Option<Timestamp>,
 }
 
 impl GrpcPayload {
-    pub(crate) fn new(data: ProtoPayload, timestamp: Option<Timestamp>) -> Self {
-        Self { data, timestamp }
+    pub(crate) fn new(data: ProtoPayload) -> Self {
+        Self { data }
     }
 }
 
@@ -658,7 +658,9 @@ mod test {
 
     use astarte_message_hub_proto::tonic::Request;
     use astarte_message_hub_proto::AstarteMessage;
-    use astarte_message_hub_proto::{pbjson_types, tonic};
+    use astarte_message_hub_proto::{
+        pbjson_types, tonic, AstarteDatastreamObject, AstartePropertyIndividual,
+    };
     use astarte_message_hub_proto_mock::mockall::{predicate, Sequence};
     use itertools::Itertools;
     use uuid::uuid;
@@ -1127,13 +1129,17 @@ mod test {
         const PATH: &str = "/1";
         let interface = Interface::from_str(crate::test::E2E_SERVER_DATASTREAM).unwrap();
         let interface_name = interface.interface_name().to_owned();
-        let expected_object = Value::Object(MockServerObject::mock_object());
-        let proto_payload: astarte_message_hub_proto::astarte_message::Payload =
-            expected_object.into();
+        let proto_payload = ProtoPayload::DatastreamObject(AstarteDatastreamObject {
+            data: MockServerObject::mock_object()
+                .into_key_values()
+                .map(|(k, v)| (k, v.into()))
+                .collect(),
+            timestamp: None,
+        });
+
         let astarte_message = super::convert::test::new_astarte_message(
             interface_name.clone(),
             PATH.to_string(),
-            None,
             proto_payload.clone(),
         );
 
@@ -1178,11 +1184,11 @@ mod test {
         const PATH: &str = "/1/enable";
         let interface = Interface::from_str(crate::test::SERVER_PROPERTIES).unwrap();
         let interface_name = interface.interface_name().to_owned();
-        let proto_payload: ProtoPayload = Value::Unset.into();
+        let proto_payload: ProtoPayload =
+            ProtoPayload::PropertyIndividual(AstartePropertyIndividual { data: None });
         let astarte_message = super::convert::test::new_astarte_message(
             interface_name.clone(),
             PATH.to_string(),
-            None,
             proto_payload.clone(),
         );