Mock Cloud Connector

Cargo.toml

@@ -21,6 +21,7 @@ members = [
   "build_common",
   "common",
   "freyja",
+  "mocks/mock_cloud_connector",
   "mocks/mock_digital_twin",
   "mocks/mock_mapping_service",
   "proc_macros",

adapters/cloud/grpc_cloud_adapter/src/grpc_cloud_adapter.rs

@@ -2,17 +2,16 @@
 // Licensed under the MIT license.
 // SPDX-License-Identifier: MIT
 
+use std::sync::Arc;
 use std::time::Duration;
-use std::{str::FromStr, sync::Arc};
 
 use async_trait::async_trait;
 use log::debug;
 use tokio::sync::Mutex;
 use tonic::transport::Channel;
 
-use cloud_connector_proto::{
-    prost_types::Timestamp,
-    v1::{cloud_connector_client::CloudConnectorClient, UpdateDigitalTwinRequestBuilder},
+use cloud_connector_proto::v1::{
+    cloud_connector_client::CloudConnectorClient, UpdateDigitalTwinRequestBuilder,
 };
 use freyja_build_common::config_file_stem;
 use freyja_common::{
@@ -81,12 +80,9 @@ impl CloudAdapter for GRPCCloudAdapter {
     ) -> Result<CloudMessageResponse, CloudAdapterError> {
         debug!("Received a request to send to the cloud");
 
-        let timestamp = Timestamp::from_str(cloud_message.signal_timestamp.as_str())
-            .map_err(CloudAdapterError::deserialize)?;
-
         let request = UpdateDigitalTwinRequestBuilder::new()
             .string_value(cloud_message.signal_value)
-            .timestamp(timestamp)
+            .timestamp_offset(cloud_message.signal_timestamp)
             .metadata(cloud_message.metadata)
             .build();
 

adapters/cloud/in_memory_mock_cloud_adapter/src/in_memory_mock_cloud_adapter.rs

@@ -75,7 +75,7 @@ mod in_memory_mock_cloud_adapter_tests {
         let cloud_message = CloudMessageRequest {
             metadata: HashMap::new(),
             signal_value: String::from("72"),
-            signal_timestamp: OffsetDateTime::now_utc().to_string(),
+            signal_timestamp: OffsetDateTime::now_utc(),
         };
 
         assert!(cloud_adapter.send_to_cloud(cloud_message).await.is_ok());

adapters/data/in_memory_mock_data_adapter/res/in_memory_mock_data_adapter_config.default.json

@@ -2,19 +2,19 @@
     "signal_update_frequency_ms": 1000,
     "entities": [
         {
-            "entity_id": "dtmi:sdv:Vehicle:Cabin:HVAC:AmbientAirTemperature;1",
+            "entity_id": "dtmi:sdv:HVAC:AmbientAirTemperature;1",
             "values": {
                 "Static": 42.0
             }
         },
         {
-            "entity_id": "dtmi:sdv:Vehicle:Cabin:HVAC:IsAirConditioningActive;1",
+            "entity_id": "dtmi:sdv:HVAC:IsAirConditioningActive;1",
             "values": {
                 "Static": 0.0
             }
         },
         {
-            "entity_id": "dtmi:sdv:Vehicle:OBD:HybridBatteryRemaining;1",
+            "entity_id": "dtmi:sdv:OBD:HybridBatteryRemaining;1",
             "values": {
                 "Stepwise": {
                     "start": 77.7,

adapters/data/in_memory_mock_data_adapter/src/in_memory_mock_data_adapter.rs

@@ -138,9 +138,10 @@ impl DataAdapter for InMemoryMockDataAdapter {
                 {
                     let data = data.lock().await;
                     for entity_id in entities_with_subscribe {
-                        if Self::generate_signal_value(&entity_id, signals.clone(), &data).is_err()
+                        if let Err(e) =
+                            Self::generate_signal_value(&entity_id, signals.clone(), &data)
                         {
-                            warn!("Attempt to set value for non-existent entity {entity_id}");
+                            warn!("Attempt to set value for non-existent entity {entity_id}: {e}");
                         }
                     }
                 }

common/Cargo.toml

@@ -17,4 +17,5 @@ proc-macros = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 strum_macros = { workspace = true }
+time = { workspace = true }
 tokio = { workspace = true }

common/src/cloud_adapter.rs

@@ -6,6 +6,7 @@ use std::{collections::HashMap, sync::Arc};
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
+use time::OffsetDateTime;
 use tokio::sync::Mutex;
 
 use crate::service_discovery_adapter_selector::ServiceDiscoveryAdapterSelector;
@@ -42,7 +43,7 @@ pub struct CloudMessageRequest {
     pub signal_value: String,
 
     // Timestamp of when the signal was emitted
-    pub signal_timestamp: String,
+    pub signal_timestamp: OffsetDateTime,
 }
 
 /// Represents a response to a message sent to the cloud digital twin

freyja/src/emitter.rs

@@ -173,7 +173,7 @@ impl<TCloudAdapter: CloudAdapter, TDataAdapterSelector: DataAdapterSelector>
         let cloud_message = CloudMessageRequest {
             metadata: signal.target.metadata.clone(),
             signal_value: converted,
-            signal_timestamp: OffsetDateTime::now_utc().to_string(),
+            signal_timestamp: OffsetDateTime::now_utc(),
         };
 
         let response = self

mocks/mock_cloud_connector/Cargo.toml

@@ -0,0 +1,24 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+# SPDX-License-Identifier: MIT
+
+[package]
+name = "mock-cloud-connector"
+version = "0.1.0"
+edition = "2021"
+license = "MIT"
+
+[dependencies]
+async-trait = { workspace = true }
+cloud-connector-proto = { workspace = true }
+env_logger = { workspace = true }
+freyja-build-common = { workspace = true }
+freyja-common = { workspace = true }
+log = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true }
+tonic = { workspace = true }
+
+[build-dependencies]
+freyja-build-common = { workspace = true }

mocks/mock_cloud_connector/README.md

@@ -0,0 +1,25 @@
+# Mock Cloud Connector
+
+The Mock Cloud Connector mocks the behavior of a Cloud Connector. This enables functionality similar to the [In-Memory Mock Cloud Adapter](../../adapters/cloud/in_memory_mock_cloud_adapter/README.md) while also utilizing the standard Freyja application.
+
+The Mock Cloud Connector implements the [Cloud Connector API](../../interfaces/cloud_connector/v1/cloud_connector.proto), making it compatible with the [gRPC Cloud Adapter](../../adapters/cloud/grpc_cloud_adapter/README.md).
+
+## Configuration
+
+This mock supports the following configuration:
+
+- `server_authority`: The authority that will be used for hosting the mock cloud connector service.
+
+This mock supports [config overrides](../../docs/tutorials/config-overrides.md). The override filename is `mock_cloud_connector_config.json`, and the default config is located at `res/mock_cloud_connector_config.default.json`.
+
+## Behavior
+
+This cloud connector prints the requests that it receives to the console, enabling users to verify that data is flowing from Freyja to the cloud connector. As a mock, this cloud connector does not have any cloud connectivity.
+
+## Build and Run
+
+To build and run the Mock Cloud Connector, run the following command:
+
+```shell
+cargo run -p mock-cloud-connector
+```

mocks/mock_cloud_connector/build.rs

@@ -0,0 +1,11 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+use freyja_build_common::copy_config;
+
+const CONFIG_FILE_STEM: &str = "mock_cloud_connector_config";
+
+fn main() {
+    copy_config(CONFIG_FILE_STEM);
+}

mocks/mock_cloud_connector/res/mock_cloud_connector_config.default.json

@@ -0,0 +1,3 @@
+{
+    "server_authority": "0.0.0.0:5176"
+}

mocks/mock_cloud_connector/src/config.rs

@@ -0,0 +1,12 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+use serde::{Deserialize, Serialize};
+
+/// Config for the mock cloud connector
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Config {
+    /// The server authority for hosting a gRPC server
+    pub server_authority: String,
+}

mocks/mock_cloud_connector/src/main.rs

@@ -0,0 +1,65 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+mod config;
+mod mock_cloud_connector_impl;
+
+use std::env;
+
+use cloud_connector_proto::v1::cloud_connector_server::CloudConnectorServer;
+use env_logger::Target;
+use log::{info, LevelFilter};
+use tonic::transport::Server;
+
+use crate::{config::Config, mock_cloud_connector_impl::MockCloudConnectorImpl};
+use freyja_build_common::config_file_stem;
+use freyja_common::{
+    cmd_utils::{get_log_level, parse_args},
+    config_utils, out_dir,
+};
+
+/// Starts the following threads and tasks:
+/// - A thread which listens for input from the command window
+/// - A task which handles async get responses
+/// - A task which handles publishing to subscribers
+/// - A gRPC server to accept incoming requests
+#[tokio::main]
+async fn main() {
+    let args = parse_args(env::args()).expect("Failed to parse args");
+
+    // Setup logging
+    let log_level = get_log_level(&args, LevelFilter::Info).expect("Could not parse log level");
+    env_logger::Builder::new()
+        .filter(None, log_level)
+        .target(Target::Stdout)
+        .init();
+
+    let config: Config = config_utils::read_from_files(
+        config_file_stem!(),
+        config_utils::JSON_EXT,
+        out_dir!(),
+        |e| log::error!("{}", e),
+        |e| log::error!("{}", e),
+    )
+    .unwrap();
+
+    // Server setup
+    info!(
+        "Mock Cloud Connector Server starting at {}",
+        config.server_authority
+    );
+
+    let addr = config
+        .server_authority
+        .parse()
+        .expect("Unable to parse server address");
+
+    let mock_cloud_connector = MockCloudConnectorImpl {};
+
+    Server::builder()
+        .add_service(CloudConnectorServer::new(mock_cloud_connector))
+        .serve(addr)
+        .await
+        .unwrap();
+}

mocks/mock_cloud_connector/src/mock_cloud_connector_impl.rs

@@ -0,0 +1,32 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+// SPDX-License-Identifier: MIT
+
+use async_trait::async_trait;
+use cloud_connector_proto::v1::{
+    cloud_connector_server::CloudConnector, UpdateDigitalTwinRequest, UpdateDigitalTwinResponse,
+};
+use log::info;
+use tonic::{Request, Response, Status};
+
+/// Implements a Mock Cloud Connector
+pub struct MockCloudConnectorImpl {}
+
+#[async_trait]
+impl CloudConnector for MockCloudConnectorImpl {
+    /// Update the digital twin
+    ///
+    /// # Arguments
+    /// - `request`: the update request+
+    async fn update_digital_twin(
+        &self,
+        request: Request<UpdateDigitalTwinRequest>,
+    ) -> Result<Response<UpdateDigitalTwinResponse>, Status> {
+        let message_json = serde_json::to_string_pretty(&request.into_inner())
+            .map_err(|_| Status::invalid_argument("Could not parse request"))?;
+
+        info!("Mock Cloud Connector received a message!\n{message_json}");
+
+        Ok(Response::new(UpdateDigitalTwinResponse {}))
+    }
+}

proto/cloud_connector/Cargo.toml

@@ -11,7 +11,9 @@ license = "MIT"
 [dependencies]
 prost = { workspace = true }
 prost-types = { workspace = true }
-time = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+time = { workspace = true, features = ["serde-human-readable"] }
 tonic = { workspace = true }
 
 [build-dependencies]