stores telemetry definitions

server/src/lib.rs

@@ -4,28 +4,61 @@ pub mod core {
     tonic::include_proto!("core");
 }
 
+use std::collections::HashMap;
 use std::error::Error;
 use std::pin::Pin;
 use log::{trace};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, Mutex};
 use tonic::{Request, Response, Status, Streaming};
 use tonic::codegen::tokio_stream::{Stream, StreamExt};
 use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
 use tonic::transport::Server;
 use core::telemetry_service_server::TelemetryService;
-use crate::core::{TelemetryDefinitionRequest, TelemetryDefinitionResponse, TelemetryInsertResponse, TelemetryItem, Uuid};
+use crate::core::{TelemetryDataType, TelemetryDefinitionRequest, TelemetryDefinitionResponse, TelemetryInsertResponse, TelemetryItem, Uuid};
 use crate::core::telemetry_service_server::TelemetryServiceServer;
 
-#[derive(Debug, Default)]
-pub struct CoreTelemetryService {}
+#[derive(Debug)]
+struct TelemetryDefinition {
+    uuid: Vec<u8>,
+    // name: String,
+    data_type: TelemetryDataType,
+}
+
+pub struct CoreTelemetryService {
+    map: Mutex<HashMap<String, TelemetryDefinition>>
+}
 
 #[tonic::async_trait]
 impl TelemetryService for CoreTelemetryService {
     async fn new_telemetry(&self, request: Request<TelemetryDefinitionRequest>) -> Result<Response<TelemetryDefinitionResponse>, Status> {
-        trace!("TelemetryService::new_telemetry {:?}", request);
+        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::random()),
+            uuid: Some(Uuid {
+                value: uuid
+            }),
         };
 
         Ok(Response::new(reply))
@@ -34,7 +67,7 @@ impl TelemetryService for CoreTelemetryService {
     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 {:?}", request);
+        trace!("TelemetryService::insert_telemetry");
 
         let mut in_stream = request.into_inner();
         let (tx, rx) = mpsc::channel(128);
@@ -62,7 +95,9 @@ impl TelemetryService for CoreTelemetryService {
 
 pub async fn setup() -> Result<(), Box<dyn Error>> {
     let addr = "[::1]:50051".parse()?;
-    let tlm_service = CoreTelemetryService::default();
+    let tlm_service = CoreTelemetryService {
+        map: Mutex::new(HashMap::new()),
+    };
 
     Server::builder()
         .add_service(TelemetryServiceServer::new(tlm_service))