save data on backend

server/src/grpc.rs

@@ -1,10 +1,9 @@
-use std::collections::HashMap;
 use std::error::Error;
 use std::pin::Pin;
 use std::sync::Arc;
 use log::{error, trace};
 use tokio::select;
-use tokio::sync::{mpsc, Mutex};
+use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 use tonic::{Request, Response, Status, Streaming};
@@ -12,56 +11,39 @@ use tonic::codegen::tokio_stream::{Stream, StreamExt};
 use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
 use tonic::transport::Server;
 use crate::core::telemetry_service_server::{TelemetryService, TelemetryServiceServer};
-use crate::core::{TelemetryDefinitionRequest, TelemetryDefinitionResponse, TelemetryInsertResponse, TelemetryItem, Uuid};
-use crate::TelemetryDefinition;
+use crate::core::{TelemetryDataType, TelemetryDefinitionRequest, TelemetryDefinitionResponse, TelemetryInsertResponse, TelemetryItem, TelemetryValue, Uuid};
+use crate::core::telemetry_value::Value;
+use crate::telemetry::{TelemetryDataValue, TelemetryManagementService};
 
 pub struct CoreTelemetryService {
-    pub map: Arc<Mutex<HashMap<String, TelemetryDefinition>>>,
+    pub tlm_management: Arc<TelemetryManagementService>,
     pub cancellation_token: CancellationToken,
 }
 
 #[tonic::async_trait]
 impl TelemetryService for CoreTelemetryService {
     async fn new_telemetry(&self, request: Request<TelemetryDefinitionRequest>) -> Result<Response<TelemetryDefinitionResponse>, Status> {
-        trace!("TelemetryService::new_telemetry");
-
-        let telemetry_definition_request = request.into_inner();
-
-        let uuid = {
-            let mut lock = self.map.lock().await;
-            if let Some(definition) = lock.get(&telemetry_definition_request.name) {
-                trace!("Telemetry Definition Found {:?}", definition);
-                if definition.data_type != telemetry_definition_request.data_type() {
-                    return Err(Status::already_exists("A telemetry item of the same name already exists"));
-                }
-                definition.uuid.clone()
-            } else {
-                trace!("Adding New Telemetry Definition {:?}", telemetry_definition_request);
-                let uuid = Uuid::random().value;
-                lock.insert(telemetry_definition_request.name.clone(), TelemetryDefinition {
-                    uuid: uuid.clone(),
-                    name: telemetry_definition_request.name.clone(),
-                    data_type: telemetry_definition_request.data_type().clone(),
-                });
-                uuid
-            }
-        };
-
-        let reply = TelemetryDefinitionResponse {
-            uuid: Some(Uuid {
-                value: uuid
-            }),
-        };
-
-        Ok(Response::new(reply))
+        trace!("CoreTelemetryService::new_telemetry");
+        self.tlm_management.register(request.into_inner()).await
+            .map(|uuid|
+                Response::new(TelemetryDefinitionResponse {
+                    uuid: Some(Uuid {
+                        value: uuid
+                    }),
+                })
+            )
+            .map_err(|err|
+                Status::already_exists(err.to_string())
+            )
     }
 
     type InsertTelemetryStream = Pin<Box<dyn Stream<Item = Result<TelemetryInsertResponse, Status>> + Send>>;
 
     async fn insert_telemetry(&self, request: Request<Streaming<TelemetryItem>>) -> Result<Response<Self::InsertTelemetryStream>, Status> {
-        trace!("TelemetryService::insert_telemetry");
+        trace!("CoreTelemetryService::insert_telemetry");
 
         let cancel_token = self.cancellation_token.clone();
+        let tlm_management = self.tlm_management.clone();
         let mut in_stream = request.into_inner();
         let (tx, rx) = mpsc::channel(128);
 
@@ -74,9 +56,8 @@ impl TelemetryService for CoreTelemetryService {
                     Some(message) = in_stream.next() => {
                         match message {
                             Ok(tlm_item) => {
-                                trace!("tlm_item {:?}", tlm_item);
                                 tx
-                                    .send(Ok(TelemetryInsertResponse {}))
+                                    .send(Self::handle_new_tlm_item(&tlm_management, &tlm_item).await)
                                     .await
                                     .expect("working rx");
                             }
@@ -94,18 +75,48 @@ impl TelemetryService for CoreTelemetryService {
     }
 }
 
-pub fn setup(token: CancellationToken, telemetry_definitions: Arc<Mutex<HashMap<String, TelemetryDefinition>>>) -> Result<JoinHandle<()>, Box<dyn Error>> {
+impl CoreTelemetryService {
+    async fn handle_new_tlm_item(tlm_management: &Arc<TelemetryManagementService>, tlm_item: &TelemetryItem) -> Result<TelemetryInsertResponse, Status> {
+        trace!("CoreTelemetryService::handle_new_tlm_item {:?}", tlm_item);
+        let Some(ref uuid) = tlm_item.uuid else {
+            return Err(Status::failed_precondition("UUID Missing"));
+        };
+        let Some(tlm_data) = tlm_management.get_by_uuid(&uuid.value).await else {
+            return Err(Status::not_found("Telemetry Item Not Found"));
+        };
+
+        let Some(TelemetryValue { value: Some(value) }) = tlm_item.value else {
+            return Err(Status::failed_precondition("Value Missing"));
+        };
+
+        let expected_type = match value {
+            Value::Float32(_) => TelemetryDataType::Float32,
+            Value::Float64(_) => TelemetryDataType::Float64,
+        };
+        if expected_type != tlm_data.definition.data_type {
+            return Err(Status::failed_precondition("Data Type Mismatch"));
+        };
+
+        let _ = tlm_data.data.send_replace(Some(match value {
+            Value::Float32(x) => TelemetryDataValue::Float32(x),
+            Value::Float64(x) => TelemetryDataValue::Float64(x),
+        }));
+
+        Ok(TelemetryInsertResponse {})
+    }
+}
+
+pub fn setup(token: CancellationToken, telemetry_management_service: Arc<TelemetryManagementService>) -> Result<JoinHandle<()>, Box<dyn Error>> {
     let addr = "[::1]:50051".parse()?;
     Ok(tokio::spawn(async move {
         let tlm_service = CoreTelemetryService {
-            map: telemetry_definitions,
+            tlm_management: telemetry_management_service,
             cancellation_token: token.clone()
         };
 
         trace!("Starting gRPC Server");
         let result = Server::builder()
             .add_service(TelemetryServiceServer::new(tlm_service))
-            // .serve(addr)
             .serve_with_shutdown(addr, token.cancelled_owned())
             .await;