initial http server

server/src/grpc.rs

@@ -0,0 +1,115 @@
+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::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+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 crate::core::telemetry_service_server::{TelemetryService, TelemetryServiceServer};
+use crate::core::{TelemetryDefinitionRequest, TelemetryDefinitionResponse, TelemetryInsertResponse, TelemetryItem, Uuid};
+use crate::TelemetryDefinition;
+
+pub struct CoreTelemetryService {
+    pub map: Arc<Mutex<HashMap<String, TelemetryDefinition>>>,
+    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))
+    }
+
+    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");
+
+        let cancel_token = self.cancellation_token.clone();
+        let mut in_stream = request.into_inner();
+        let (tx, rx) = mpsc::channel(128);
+
+        tokio::spawn(async move {
+            loop {
+                select! {
+                    _ = cancel_token.cancelled() => {
+                        break;
+                    },
+                    Some(message) = in_stream.next() => {
+                        match message {
+                            Ok(tlm_item) => {
+                                trace!("tlm_item {:?}", tlm_item);
+                                tx
+                                    .send(Ok(TelemetryInsertResponse {}))
+                                    .await
+                                    .expect("working rx");
+                            }
+                            Err(err) => {
+                                let _ = tx.send(Err(err)).await;
+                            }
+                        }
+                    },
+                    else => break,
+                }
+            }
+        });
+
+        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
+    }
+}
+
+pub fn setup(token: CancellationToken, telemetry_definitions: Arc<Mutex<HashMap<String, TelemetryDefinition>>>) -> Result<JoinHandle<()>, Box<dyn Error>> {
+    let addr = "[::1]:50051".parse()?;
+    Ok(tokio::spawn(async move {
+        let tlm_service = CoreTelemetryService {
+            map: telemetry_definitions,
+            cancellation_token: token.clone()
+        };
+
+        let result = Server::builder()
+            .add_service(TelemetryServiceServer::new(tlm_service))
+            // .serve(addr)
+            .serve_with_shutdown(addr, token.cancelled_owned())
+            .await;
+
+        if let Err(err) = result {
+            error!("gRPC Server Encountered An Error: {err}");
+        }
+    }))
+}