add tests to api

Cargo.toml

@@ -1,6 +1,33 @@
 [workspace]
-members = ["server", "examples/simple_producer", "examples/simple_command"]
+members = ["api", "api-core", "api-proc-macro", "server", "examples/simple_producer", "examples/simple_command"]
 resolver = "2"
 
+[workspace.dependencies]
+actix-web = "4.12.1"
+actix-ws = "0.3.0"
+anyhow = "1.0.100"
+chrono = { version = "0.4.42" }
+derive_more = { version = "2.1.1" }
+env_logger = "0.11.8"
+fern = "0.7.1"
+futures-util = "0.3.31"
+log = "0.4.29"
+num-traits = "0.2.19"
+papaya = "0.2.3"
+proc-macro-error = "1.0.4"
+quote = "1.0.42"
+serde = { version = "1.0.228" }
+serde_json = "1.0.148"
+sqlx = "0.8.6"
+syn = "2.0.112"
+thiserror = "2.0.17"
+tokio = { version = "1.48.0" }
+tokio-test = "0.4.4"
+tokio-stream = "0.1.17"
+tokio-tungstenite = { version = "0.28.0" }
+tokio-util = "0.7.17"
+trybuild = "1.0.114"
+uuid = { version = "1.19.0", features = ["v4"] }
+
 [profile.dev.package.sqlx-macros]
 opt-level = 3

api-core/Cargo.toml

@@ -0,0 +1,12 @@
+
+[package]
+name = "api-core"
+edition = "2021"
+version = "0.1.0"
+authors = ["Sergey <me@sergeysav.com>"]
+
+[dependencies]
+chrono = { workspace = true, features = ["serde"] }
+derive_more = { workspace = true, features = ["from", "try_into"] }
+serde = { workspace = true, features = ["derive"] }
+thiserror = { workspace = true }

api-core/src/command.rs

@@ -0,0 +1,47 @@
+use crate::data_type::DataType;
+use crate::data_value::DataValue;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use thiserror::Error;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CommandParameterDefinition {
+    pub name: String,
+    pub data_type: DataType,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CommandDefinition {
+    pub name: String,
+    pub parameters: Vec<CommandParameterDefinition>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CommandHeader {
+    pub timestamp: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Command {
+    #[serde(flatten)]
+    pub header: CommandHeader,
+    pub parameters: HashMap<String, DataValue>,
+}
+
+#[derive(Debug, PartialEq, Eq, Error)]
+pub enum IntoCommandDefinitionError {
+    #[error("Parameter Missing: {0}")]
+    ParameterMissing(String),
+    #[error("Mismatched Type for {parameter}. {expected:?} expected")]
+    MismatchedType {
+        parameter: String,
+        expected: DataType,
+    },
+}
+
+pub trait IntoCommandDefinition: Sized {
+    fn create(name: String) -> CommandDefinition;
+
+    fn parse(command: Command) -> Result<Self, IntoCommandDefinitionError>;
+}

api-core/src/data_type.rs

@@ -0,0 +1,25 @@
+use crate::data_value::DataValue;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum DataType {
+    Float32,
+    Float64,
+    Boolean,
+}
+
+pub trait ToDataType: Into<DataValue> {
+    const DATA_TYPE: DataType;
+}
+
+macro_rules! impl_to_data_type {
+    ( $ty:ty, $value:expr ) => {
+        impl ToDataType for $ty {
+            const DATA_TYPE: DataType = $value;
+        }
+    };
+}
+
+impl_to_data_type!(f32, DataType::Float32);
+impl_to_data_type!(f64, DataType::Float64);
+impl_to_data_type!(bool, DataType::Boolean);

api-core/src/data_value.rs

@@ -0,0 +1,9 @@
+use derive_more::{From, TryInto};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, From, TryInto)]
+pub enum DataValue {
+    Float32(f32),
+    Float64(f64),
+    Boolean(bool),
+}

api-core/src/lib.rs

@@ -0,0 +1,3 @@
+pub mod command;
+pub mod data_type;
+pub mod data_value;

api-proc-macro/Cargo.toml

@@ -0,0 +1,19 @@
+
+[package]
+name = "api-proc-macro"
+edition = "2021"
+version = "0.1.0"
+authors = ["Sergey <me@sergeysav.com>"]
+
+[lib]
+proc-macro = true
+
+[dependencies]
+api-core = { path = "../api-core" }
+proc-macro-error = { workspace = true }
+quote = { workspace = true }
+syn = { workspace = true }
+
+[dev-dependencies]
+trybuild = { workspace = true }
+api = { path = "../api" }

api-proc-macro/src/into_command_definition.rs

@@ -0,0 +1,129 @@
+use proc_macro_error::abort;
+use quote::{quote, quote_spanned};
+use syn::spanned::Spanned;
+use syn::{parse_macro_input, parse_quote, Data, DeriveInput, Fields, GenericParam, Generics};
+
+pub fn derive_into_command_definition_impl(
+    item: proc_macro::TokenStream,
+) -> proc_macro::TokenStream {
+    let DeriveInput {
+        ident,
+        data,
+        generics,
+        ..
+    }: DeriveInput = parse_macro_input!(item as DeriveInput);
+
+    let data = match data {
+        Data::Struct(data) => data,
+        Data::Enum(data) => abort!(
+            data.enum_token,
+            "IntoCommandDefinition not supported for enum"
+        ),
+        Data::Union(data) => abort!(
+            data.union_token,
+            "IntoCommandDefinition not supported for union"
+        ),
+    };
+
+    let generics = add_trait_bounds(generics);
+    let (impl_generics, ty_generics, where_clause) = generics.split_for_impl();
+
+    let num_fields = data.fields.len();
+
+    let create_param_stream = match &data.fields {
+        Fields::Named(fields) => {
+            let field_entries = fields.named.iter().map(|field| {
+                let name = field.ident.clone().map(|id| id.to_string());
+                let field_type = &field.ty;
+                quote_spanned! { field.span() =>
+                    parameters.push(api::messages::command::CommandParameterDefinition {
+                        name: #name.to_string(),
+                        data_type: <#field_type as api::data_type::ToDataType>::DATA_TYPE,
+                    });
+                }
+            });
+            quote! { #(#field_entries)* }
+        }
+        Fields::Unnamed(fields) => abort!(
+            fields,
+            "IntoCommandDefinition not supported for unnamed structs"
+        ),
+        Fields::Unit => quote! {},
+    };
+    let parse_param_stream = match &data.fields {
+        Fields::Named(fields) => {
+            let field_entries = fields.named.iter().map(|field| {
+                let name = &field.ident;
+                let name_string = field.ident.clone().map(|id| id.to_string());
+                let field_type = &field.ty;
+                quote_spanned! { field.span() =>
+                    let #name: #field_type = (*command
+                        .parameters
+                        .get(#name_string)
+                        .ok_or_else(|| api::messages::command::IntoCommandDefinitionError::ParameterMissing(#name_string.to_string()))?)
+                        .try_into()
+                        .map_err(|_| api::messages::command::IntoCommandDefinitionError::MismatchedType {
+                            parameter: #name_string.to_string(),
+                            expected: <#field_type as api::data_type::ToDataType>::DATA_TYPE,
+                        })?;
+                }
+            });
+            quote! { #(#field_entries)* }
+        }
+        Fields::Unnamed(fields) => abort!(
+            fields,
+            "IntoCommandDefinition not supported for unnamed structs"
+        ),
+        Fields::Unit => quote! {},
+    };
+    let param_name_stream = match &data.fields {
+        Fields::Named(fields) => {
+            let field_entries = fields.named.iter().map(|field| {
+                let name = &field.ident;
+                quote_spanned! { field.span() => #name, }
+            });
+            quote! { #(#field_entries)* }
+        }
+        Fields::Unnamed(fields) => abort!(
+            fields,
+            "IntoCommandDefinition not supported for unnamed structs"
+        ),
+        Fields::Unit => quote! {},
+    };
+
+    let result = quote! {
+        impl #impl_generics api::messages::command::IntoCommandDefinition for #ident #ty_generics #where_clause {
+            fn create(name: std::string::String) -> api::messages::command::CommandDefinition {
+                let mut parameters = std::vec::Vec::with_capacity( #num_fields );
+                #create_param_stream
+                api::messages::command::CommandDefinition {
+                    name: name,
+                    parameters: parameters,
+                }
+            }
+
+            fn parse(command: api::messages::command::Command) -> core::result::Result<Self, api::messages::command::IntoCommandDefinitionError> {
+                #parse_param_stream
+                Ok(Self {
+                    #param_name_stream
+                })
+            }
+        }
+    };
+
+    result.into()
+}
+
+fn add_trait_bounds(mut generics: Generics) -> Generics {
+    for param in &mut generics.params {
+        if let GenericParam::Type(ref mut type_param) = *param {
+            type_param
+                .bounds
+                .push(parse_quote!(api::data_type::ToDataType));
+            type_param.bounds.push(parse_quote!(
+                core::convert::TryFrom<api::data_value::DataValue>
+            ));
+        }
+    }
+    generics
+}

api-proc-macro/src/lib.rs

@@ -0,0 +1,11 @@
+extern crate proc_macro;
+
+use proc_macro_error::proc_macro_error;
+
+mod into_command_definition;
+
+#[proc_macro_error]
+#[proc_macro_derive(IntoCommandDefinition)]
+pub fn derive_into_command_definition(item: proc_macro::TokenStream) -> proc_macro::TokenStream {
+    into_command_definition::derive_into_command_definition_impl(item)
+}

api-proc-macro/tests/into_command_definition/enum_fails.rs

@@ -0,0 +1,10 @@
+use api_proc_macro::IntoCommandDefinition;
+
+#[derive(IntoCommandDefinition)]
+enum TestEnum {
+    Variant
+}
+
+fn main() {
+
+}

api-proc-macro/tests/into_command_definition/enum_fails.stderr

@@ -0,0 +1,5 @@
+error: IntoCommandDefinition not supported for enum
+ --> tests/into_command_definition/enum_fails.rs:4:1
+  |
+4 | enum TestEnum {
+  | ^^^^

api-proc-macro/tests/into_command_definition/union_fails.rs

@@ -0,0 +1,12 @@
+use api_proc_macro::IntoCommandDefinition;
+
+#[derive(IntoCommandDefinition)]
+#[repr(C)]
+union TestUnion {
+    f1: u32,
+    f2: f32,
+}
+
+fn main() {
+
+}

api-proc-macro/tests/into_command_definition/union_fails.stderr

@@ -0,0 +1,5 @@
+error: IntoCommandDefinition not supported for union
+ --> tests/into_command_definition/union_fails.rs:5:1
+  |
+5 | union TestUnion {
+  | ^^^^^

api-proc-macro/tests/into_command_definition/unnamed_struct_fails.rs

@@ -0,0 +1,8 @@
+use api_proc_macro::IntoCommandDefinition;
+
+#[derive(IntoCommandDefinition)]
+struct TestUnnamedStruct(f32, f64, bool);
+
+fn main() {
+
+}

api-proc-macro/tests/into_command_definition/unnamed_struct_fails.stderr

@@ -0,0 +1,5 @@
+error: IntoCommandDefinition not supported for unnamed structs
+ --> tests/into_command_definition/unnamed_struct_fails.rs:4:25
+  |
+4 | struct TestUnnamedStruct(f32, f64, bool);
+  |                         ^^^^^^^^^^^^^^^^

api-proc-macro/tests/test_derive_macro_into_command_definition.rs

@@ -0,0 +1,151 @@
+use api_core::command::{
+    Command, CommandHeader, CommandParameterDefinition, IntoCommandDefinition,
+};
+use api_core::data_type::DataType;
+use api_core::data_value::DataValue;
+use api_proc_macro::IntoCommandDefinition;
+use std::collections::HashMap;
+
+#[test]
+fn test_enum_fails() {
+    let t = trybuild::TestCases::new();
+    t.compile_fail("tests/into_command_definition/enum_fails.rs");
+}
+
+#[test]
+fn test_union_fails() {
+    let t = trybuild::TestCases::new();
+    t.compile_fail("tests/into_command_definition/union_fails.rs");
+}
+
+#[test]
+fn test_unnamed_struct_fails() {
+    let t = trybuild::TestCases::new();
+    t.compile_fail("tests/into_command_definition/unnamed_struct_fails.rs");
+}
+
+#[test]
+fn test_basic_command() {
+    #[derive(IntoCommandDefinition)]
+    struct TestStruct {
+        #[allow(unused)]
+        a: f32,
+        #[allow(unused)]
+        b: f64,
+        #[allow(unused)]
+        c: bool,
+    }
+
+    let command_definition = TestStruct::create("Test".to_string());
+
+    assert_eq!(command_definition.name, "Test");
+    assert_eq!(command_definition.parameters.capacity(), 3);
+    assert_eq!(
+        command_definition.parameters[0],
+        CommandParameterDefinition {
+            name: "a".to_string(),
+            data_type: DataType::Float32,
+        }
+    );
+    assert_eq!(
+        command_definition.parameters[1],
+        CommandParameterDefinition {
+            name: "b".to_string(),
+            data_type: DataType::Float64,
+        }
+    );
+    assert_eq!(
+        command_definition.parameters[2],
+        CommandParameterDefinition {
+            name: "c".to_string(),
+            data_type: DataType::Boolean,
+        }
+    );
+
+    let mut parameters = HashMap::new();
+    parameters.insert("a".to_string(), DataValue::Float32(1.0));
+    parameters.insert("b".to_string(), DataValue::Float64(2.0));
+    parameters.insert("c".to_string(), DataValue::Boolean(true));
+    let result = TestStruct::parse(Command {
+        header: CommandHeader {
+            timestamp: Default::default(),
+        },
+        parameters,
+    })
+    .unwrap();
+    assert_eq!(result.a, 1.0f32);
+    assert_eq!(result.b, 2.0f64);
+    assert_eq!(result.c, true);
+}
+
+#[test]
+fn test_generic_command() {
+    #[derive(IntoCommandDefinition)]
+    struct TestStruct<T> {
+        #[allow(unused)]
+        a: T,
+    }
+
+    let command_definition = TestStruct::<f32>::create("Test".to_string());
+    assert_eq!(command_definition.name, "Test");
+    assert_eq!(command_definition.parameters.capacity(), 1);
+    assert_eq!(
+        command_definition.parameters[0],
+        CommandParameterDefinition {
+            name: "a".to_string(),
+            data_type: DataType::Float32,
+        }
+    );
+    let mut parameters = HashMap::new();
+    parameters.insert("a".to_string(), DataValue::Float32(1.0));
+    let result = TestStruct::<f32>::parse(Command {
+        header: CommandHeader {
+            timestamp: Default::default(),
+        },
+        parameters,
+    })
+    .unwrap();
+    assert_eq!(result.a, 1.0f32);
+
+    let command_definition = TestStruct::<f64>::create("Test2".to_string());
+    assert_eq!(command_definition.name, "Test2");
+    assert_eq!(command_definition.parameters.capacity(), 1);
+    assert_eq!(
+        command_definition.parameters[0],
+        CommandParameterDefinition {
+            name: "a".to_string(),
+            data_type: DataType::Float64,
+        }
+    );
+    let mut parameters = HashMap::new();
+    parameters.insert("a".to_string(), DataValue::Float64(2.0));
+    let result = TestStruct::<f64>::parse(Command {
+        header: CommandHeader {
+            timestamp: Default::default(),
+        },
+        parameters,
+    })
+    .unwrap();
+    assert_eq!(result.a, 2.0f64);
+
+    let command_definition = TestStruct::<bool>::create("Test3".to_string());
+    assert_eq!(command_definition.name, "Test3");
+    assert_eq!(command_definition.parameters.capacity(), 1);
+    assert_eq!(
+        command_definition.parameters[0],
+        CommandParameterDefinition {
+            name: "a".to_string(),
+            data_type: DataType::Boolean,
+        }
+    );
+    let mut parameters = HashMap::new();
+    parameters.insert("a".to_string(), DataValue::Boolean(true));
+    let result = TestStruct::<bool>::parse(Command {
+        header: CommandHeader {
+            timestamp: Default::default(),
+        },
+        parameters,
+    })
+    .unwrap();
+    assert_eq!(result.a, true);
+}

api/Cargo.toml

@@ -0,0 +1,24 @@
+
+[package]
+name = "api"
+edition = "2021"
+version = "0.1.0"
+authors = ["Sergey <me@sergeysav.com>"]
+
+[dependencies]
+api-core = { path = "../api-core" }
+api-proc-macro = { path = "../api-proc-macro" }
+chrono = { workspace = true, features = ["serde"] }
+derive_more = { workspace = true, features = ["from", "try_into"] }
+futures-util = { workspace = true }
+log = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["rt", "macros", "time"] }
+tokio-tungstenite = { workspace = true, features = ["rustls-tls-native-roots"] }
+tokio-util = { workspace = true }
+uuid = { workspace = true, features = ["serde"] }
+
+[dev-dependencies]
+env_logger = { workspace = true }

api/src/client/command.rs

@@ -0,0 +1,80 @@
+use crate::client::Client;
+use crate::messages::command::CommandResponse;
+use api_core::command::{CommandHeader, IntoCommandDefinition};
+use std::fmt::Display;
+use std::sync::Arc;
+use tokio_util::sync::CancellationToken;
+
+pub struct CommandRegistry {
+    client: Arc<Client>,
+}
+
+impl CommandRegistry {
+    pub fn new(client: Arc<Client>) -> Self {
+        Self { client }
+    }
+
+    pub fn register_handler<C: IntoCommandDefinition, F, E: Display>(
+        &self,
+        command_name: impl Into<String>,
+        mut callback: F,
+    ) -> CommandHandle
+    where
+        F: FnMut(CommandHeader, C) -> Result<String, E> + Send + 'static,
+    {
+        let cancellation_token = CancellationToken::new();
+        let result = CommandHandle {
+            cancellation_token: cancellation_token.clone(),
+        };
+        let client = self.client.clone();
+
+        let command_definition = C::create(command_name.into());
+
+        tokio::spawn(async move {
+            while !cancellation_token.is_cancelled() {
+                // This would only fail if the sender closed while trying to insert data
+                // It would wait until space is made
+                let Ok(mut rx) = client
+                    .register_callback_channel(command_definition.clone())
+                    .await
+                else {
+                    continue;
+                };
+
+                while let Some((cmd, responder)) = rx.recv().await {
+                    let header = cmd.header.clone();
+                    let response = match C::parse(cmd) {
+                        Ok(cmd) => match callback(header, cmd) {
+                            Ok(response) => CommandResponse {
+                                success: true,
+                                response,
+                            },
+                            Err(err) => CommandResponse {
+                                success: false,
+                                response: err.to_string(),
+                            },
+                        },
+                        Err(err) => CommandResponse {
+                            success: false,
+                            response: err.to_string(),
+                        },
+                    };
+                    // This should only err if we had an error elsewhere
+                    let _ = responder.send(response);
+                }
+            }
+        });
+
+        result
+    }
+}
+
+pub struct CommandHandle {
+    cancellation_token: CancellationToken,
+}
+
+impl Drop for CommandHandle {
+    fn drop(&mut self) {
+        self.cancellation_token.cancel();
+    }
+}

api/src/client/config.rs

@@ -0,0 +1,11 @@
+pub struct ClientConfiguration {
+    pub send_buffer_size: usize,
+}
+
+impl Default for ClientConfiguration {
+    fn default() -> Self {
+        Self {
+            send_buffer_size: 128,
+        }
+    }
+}

api/src/client/context.rs

@@ -0,0 +1,594 @@
+use crate::client::config::ClientConfiguration;
+use crate::client::error::{ConnectError, MessageError};
+use crate::client::{Callback, ClientChannel, OutgoingMessage, RegisteredCallback};
+use crate::messages::callback::GenericCallbackError;
+use crate::messages::payload::RequestMessagePayload;
+use crate::messages::{RequestMessage, ResponseMessage};
+use futures_util::{Sink, SinkExt, Stream, StreamExt};
+use log::{debug, error, info, trace, warn};
+use std::collections::HashMap;
+use std::fmt::Display;
+use std::sync::mpsc::sync_channel;
+use std::thread;
+use std::time::Duration;
+use tokio::sync::{mpsc, oneshot, watch, RwLockWriteGuard};
+use tokio::time::sleep;
+use tokio::{select, spawn};
+use tokio_tungstenite::connect_async;
+use tokio_tungstenite::tungstenite::handshake::client::{Request, Response as TungResponse};
+use tokio_tungstenite::tungstenite::{Error as TungError, Message};
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+pub struct ClientContext {
+    pub cancel: CancellationToken,
+    pub request: Request,
+    pub connected_state_tx: watch::Sender<bool>,
+    pub client_configuration: ClientConfiguration,
+}
+
+impl ClientContext {
+    pub fn start(mut self, channel: ClientChannel) -> Result<(), ConnectError> {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()?;
+
+        let (tx, rx) = sync_channel::<()>(1);
+
+        let _detached = thread::Builder::new()
+            .name("tlm-client".to_string())
+            .spawn(move || {
+                runtime.block_on(async {
+                    let mut write_lock = channel.write().await;
+
+                    // This cannot fail
+                    let _ = tx.send(());
+
+                    while !self.cancel.is_cancelled() {
+                        write_lock = self
+                            .run_connection(write_lock, &channel, connect_async)
+                            .await;
+                    }
+                    drop(write_lock);
+                });
+            })?;
+
+        // This cannot fail
+        let _ = rx.recv();
+
+        Ok(())
+    }
+
+    async fn run_connection<'a, F, W, E>(
+        &mut self,
+        mut write_lock: RwLockWriteGuard<'a, mpsc::Sender<OutgoingMessage>>,
+        channel: &'a ClientChannel,
+        mut connection_fn: F,
+    ) -> RwLockWriteGuard<'a, mpsc::Sender<OutgoingMessage>>
+    where
+        F: AsyncFnMut(Request) -> Result<(W, TungResponse), TungError>,
+        W: Stream<Item = Result<Message, TungError>> + Sink<Message, Error = E> + Unpin,
+        E: Display,
+    {
+        debug!("Attempting to Connect to {}", self.request.uri());
+        let mut ws = match connection_fn(self.request.clone()).await {
+            Ok((ws, _)) => ws,
+            Err(e) => {
+                info!("Failed to Connect: {e}");
+                sleep(Duration::from_secs(1)).await;
+                return write_lock;
+            }
+        };
+        info!("Connected to {}", self.request.uri());
+
+        let (tx, rx) = mpsc::channel(self.client_configuration.send_buffer_size);
+        *write_lock = tx;
+        drop(write_lock);
+
+        // Don't care about the previous value
+        let _ = self.connected_state_tx.send_replace(true);
+
+        let close_connection = self.handle_connection(&mut ws, rx, channel).await;
+
+        let write_lock = channel.write().await;
+        // Send this after grabbing the lock - to prevent extra contention when others try to grab
+        // the lock to use that as a signal that we have reconnected
+        let _ = self.connected_state_tx.send_replace(false);
+        if close_connection {
+            // Manually close to allow the impl trait to be used
+            if let Err(e) = ws.send(Message::Close(None)).await {
+                error!("Failed to Close the Connection: {e}");
+            }
+        }
+        write_lock
+    }
+
+    async fn handle_connection<W>(
+        &mut self,
+        ws: &mut W,
+        mut rx: mpsc::Receiver<OutgoingMessage>,
+        channel: &ClientChannel,
+    ) -> bool
+    where
+        W: Stream<Item = Result<Message, TungError>> + Sink<Message> + Unpin,
+        <W as Sink<Message>>::Error: Display,
+    {
+        let mut callbacks = HashMap::<Uuid, Callback>::new();
+        loop {
+            select! {
+                _ = self.cancel.cancelled() => { break; },
+                Some(msg) = ws.next() => {
+                    match msg {
+                        Ok(msg) => {
+                            match msg {
+                                Message::Text(msg) => {
+                                    trace!("Incoming: {msg}");
+                                    let msg: ResponseMessage = match serde_json::from_str(&msg) {
+                                        Ok(m) => m,
+                                        Err(e) => {
+                                            error!("Failed to deserialize {e}");
+                                            break;
+                                        }
+                                    };
+                                    self.handle_incoming(msg, &mut callbacks, channel).await;
+                                }
+                                Message::Binary(_) => unimplemented!("Binary Data Not Implemented"),
+                                Message::Ping(data) => {
+                                    if let Err(e) = ws.send(Message::Pong(data)).await {
+                                        error!("Failed to send Pong {e}");
+                                        break;
+                                    }
+                                }
+                                Message::Pong(_) => {
+                                    // Intentionally Left Empty
+                                }
+                                Message::Close(_) => {
+                                    debug!("Websocket Closed");
+                                    return false;
+                                }
+                                Message::Frame(_) => unreachable!("Not Possible"),
+                            }
+                        }
+                        Err(e) => {
+                            error!("Receive Error {e}");
+                            break;
+                        }
+                    }
+                }
+                Some(msg) = rx.recv() => {
+                    // Insert a callback if it isn't a None callback
+                    if !matches!(msg.callback, Callback::None) {
+                        callbacks.insert(msg.msg.uuid, msg.callback);
+                    }
+                    let msg = match serde_json::to_string(&msg.msg) {
+                        Ok(m) => m,
+                        Err(e) => {
+                            error!("Encode Error {e}");
+                            break;
+                        }
+                    };
+                    trace!("Outgoing: {msg}");
+                    if let Err(e) = ws.send(Message::Text(msg.into())).await {
+                        error!("Send Error {e}");
+                        break;
+                    }
+                }
+                else => { break; },
+            }
+        }
+        true
+    }
+
+    async fn handle_incoming(
+        &mut self,
+        msg: ResponseMessage,
+        callbacks: &mut HashMap<Uuid, Callback>,
+        channel: &ClientChannel,
+    ) {
+        if let Some(response_uuid) = msg.response {
+            match callbacks.get(&response_uuid) {
+                Some(Callback::None) => {
+                    callbacks.remove(&response_uuid);
+                    unreachable!("We skip registering callbacks of None type");
+                }
+                Some(Callback::Once(_)) => {
+                    let Some(Callback::Once(callback)) = callbacks.remove(&response_uuid) else {
+                        return;
+                    };
+                    let _ = callback.send(msg);
+                }
+                Some(Callback::Registered(callback)) => {
+                    let callback = callback.clone();
+                    spawn(Self::handle_registered_callback(
+                        callback,
+                        msg,
+                        channel.clone(),
+                    ));
+                }
+                None => {
+                    warn!("No Callback Registered for {response_uuid}");
+                }
+            }
+        }
+    }
+
+    async fn handle_registered_callback(
+        callback: RegisteredCallback,
+        msg: ResponseMessage,
+        channel: ClientChannel,
+    ) {
+        let (tx, rx) = oneshot::channel();
+
+        let uuid = msg.uuid;
+
+        let response = match callback.send((msg, tx)).await {
+            Err(_) => GenericCallbackError::CallbackClosed.into(),
+            Ok(()) => rx
+                .await
+                .unwrap_or_else(|_| GenericCallbackError::CallbackClosed.into()),
+        };
+
+        if let Err(e) = Self::send_response(channel, response, uuid).await {
+            error!("Failed to send response {e}");
+        }
+    }
+
+    async fn send_response(
+        channel: ClientChannel,
+        payload: RequestMessagePayload,
+        response_uuid: Uuid,
+    ) -> Result<(), MessageError> {
+        // If this failed that means we're in the middle of reconnecting, so our callbacks
+        // are all being cleaned up as-is. No response needed.
+        let sender = channel.try_read()?;
+        let data = sender.reserve().await?;
+
+        data.send(OutgoingMessage {
+            msg: RequestMessage {
+                uuid: Uuid::new_v4(),
+                response: Some(response_uuid),
+                payload,
+            },
+            callback: Callback::None,
+        });
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::messages::telemetry_definition::{
+        TelemetryDefinitionRequest, TelemetryDefinitionResponse,
+    };
+    use crate::test::mock_stream_sink::{create_mock_stream_sink, MockStreamSinkControl};
+    use api_core::data_type::DataType;
+    use log::LevelFilter;
+    use std::future::Future;
+    use std::ops::Deref;
+    use tokio::sync::mpsc::Sender;
+    use tokio::sync::RwLock;
+    use tokio::time::timeout;
+    use tokio::try_join;
+    use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+    use tokio_util::bytes::Bytes;
+
+    async fn assert_client_interaction<F, R>(future: F)
+    where
+        F: Send
+            + FnOnce(
+                Sender<OutgoingMessage>,
+                MockStreamSinkControl<Result<Message, TungError>, Message>,
+                CancellationToken,
+            ) -> R
+            + 'static,
+        R: Future<Output = ()> + Send,
+    {
+        let (control, stream_sink) =
+            create_mock_stream_sink::<Result<Message, TungError>, Message>();
+
+        let cancel_token = CancellationToken::new();
+        let inner_cancel_token = cancel_token.clone();
+        let (connected_state_tx, _connected_state_rx) = watch::channel(false);
+
+        let mut context = ClientContext {
+            cancel: cancel_token,
+            request: "mock".into_client_request().unwrap(),
+            connected_state_tx,
+            client_configuration: Default::default(),
+        };
+
+        let (tx, _rx) = mpsc::channel(1);
+        let channel = ClientChannel::new(RwLock::new(tx));
+        let used_channel = channel.clone();
+
+        let write_lock = used_channel.write().await;
+
+        let handle = spawn(async move {
+            let channel = channel;
+            let read = channel.read().await;
+            let sender = read.deref().clone();
+            drop(read);
+            future(sender, control, inner_cancel_token).await;
+        });
+
+        let mut stream_sink = Some(stream_sink);
+
+        let connection_fn = async |_: Request| {
+            let stream_sink = stream_sink.take().ok_or(TungError::ConnectionClosed)?;
+
+            Ok((stream_sink, TungResponse::default())) as Result<(_, _), TungError>
+        };
+
+        let context_result = async {
+            drop(
+                context
+                    .run_connection(write_lock, &used_channel, connection_fn)
+                    .await,
+            );
+            Ok(())
+        };
+
+        try_join!(context_result, timeout(Duration::from_secs(1), handle),)
+            .unwrap()
+            .1
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn connection_closes_when_websocket_closes() {
+        let _ = env_logger::builder()
+            .is_test(true)
+            .filter_level(LevelFilter::Trace)
+            .try_init();
+
+        assert_client_interaction(|sender, mut control, _| async move {
+            let msg = Uuid::new_v4();
+            sender
+                .send(OutgoingMessage {
+                    msg: RequestMessage {
+                        uuid: msg,
+                        response: None,
+                        payload: TelemetryDefinitionRequest {
+                            name: "".to_string(),
+                            data_type: DataType::Float32,
+                        }
+                        .into(),
+                    },
+                    callback: Callback::None,
+                })
+                .await
+                .unwrap();
+            // We expect an outgoing message
+            assert!(matches!(
+                control.outgoing.recv().await.unwrap(),
+                Message::Text(_)
+            ));
+            // We receive an incoming close message
+            control
+                .incoming
+                .send(Ok(Message::Close(None)))
+                .await
+                .unwrap();
+            // Then we expect the outgoing to close with no message
+            assert!(control.outgoing.recv().await.is_none());
+            assert!(control.incoming.is_closed());
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn connection_closes_when_cancelled() {
+        let _ = env_logger::builder()
+            .is_test(true)
+            .filter_level(LevelFilter::Trace)
+            .try_init();
+
+        assert_client_interaction(|_, mut control, cancel| async move {
+            cancel.cancel();
+            // We expect an outgoing cancel message
+            assert!(matches!(
+                control.outgoing.recv().await.unwrap(),
+                Message::Close(_)
+            ));
+            // Then we expect to close with no message
+            assert!(control.outgoing.recv().await.is_none());
+            assert!(control.incoming.is_closed());
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn callback_request() {
+        let _ = env_logger::builder()
+            .is_test(true)
+            .filter_level(LevelFilter::Trace)
+            .try_init();
+
+        assert_client_interaction(|sender, mut control, _| async move {
+            let (callback_tx, callback_rx) = oneshot::channel();
+            let msg = Uuid::new_v4();
+            sender
+                .send(OutgoingMessage {
+                    msg: RequestMessage {
+                        uuid: msg,
+                        response: None,
+                        payload: TelemetryDefinitionRequest {
+                            name: "".to_string(),
+                            data_type: DataType::Float32,
+                        }
+                        .into(),
+                    },
+                    callback: Callback::Once(callback_tx),
+                })
+                .await
+                .unwrap();
+
+            // We expect an outgoing message
+            assert!(matches!(
+                control.outgoing.recv().await.unwrap(),
+                Message::Text(_)
+            ));
+
+            // Then we get an incoming message for this callback
+            let response_message = ResponseMessage {
+                uuid: Uuid::new_v4(),
+                response: Some(msg),
+                payload: TelemetryDefinitionResponse {
+                    uuid: Uuid::new_v4(),
+                }
+                .into(),
+            };
+            control
+                .incoming
+                .send(Ok(Message::Text(
+                    serde_json::to_string(&response_message).unwrap().into(),
+                )))
+                .await
+                .unwrap();
+
+            // We expect the callback to run
+            let message = callback_rx.await.unwrap();
+            // And give us the message we provided it
+            assert_eq!(message, response_message);
+
+            // We receive an incoming close message
+            control
+                .incoming
+                .send(Ok(Message::Close(None)))
+                .await
+                .unwrap();
+            // Then we expect the outgoing to close with no message
+            assert!(control.outgoing.recv().await.is_none());
+            assert!(control.incoming.is_closed());
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn callback_registered() {
+        let _ = env_logger::builder()
+            .is_test(true)
+            .filter_level(LevelFilter::Trace)
+            .try_init();
+
+        assert_client_interaction(|sender, mut control, _| async move {
+            let (callback_tx, mut callback_rx) = mpsc::channel(1);
+            let msg = Uuid::new_v4();
+            sender
+                .send(OutgoingMessage {
+                    msg: RequestMessage {
+                        uuid: msg,
+                        response: None,
+                        payload: TelemetryDefinitionRequest {
+                            name: "".to_string(),
+                            data_type: DataType::Float32,
+                        }
+                        .into(),
+                    },
+                    callback: Callback::Registered(callback_tx),
+                })
+                .await
+                .unwrap();
+
+            // We expect an outgoing message
+            assert!(matches!(
+                control.outgoing.recv().await.unwrap(),
+                Message::Text(_)
+            ));
+
+            // We handle the callback a few times
+            for _ in 0..5 {
+                // Then we get an incoming message for this callback
+                let response_message = ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg),
+                    payload: TelemetryDefinitionResponse {
+                        uuid: Uuid::new_v4(),
+                    }
+                    .into(),
+                };
+                control
+                    .incoming
+                    .send(Ok(Message::Text(
+                        serde_json::to_string(&response_message).unwrap().into(),
+                    )))
+                    .await
+                    .unwrap();
+
+                // We expect the response
+                let (rx, responder) = callback_rx.recv().await.unwrap();
+                // And give us the message we provided it
+                assert_eq!(rx, response_message);
+                // Then the response gets sent out
+                responder
+                    .send(
+                        TelemetryDefinitionRequest {
+                            name: "".to_string(),
+                            data_type: DataType::Float32,
+                        }
+                        .into(),
+                    )
+                    .unwrap();
+
+                // We expect an outgoing message
+                assert!(matches!(
+                    control.outgoing.recv().await.unwrap(),
+                    Message::Text(_)
+                ));
+            }
+
+            // We receive an incoming close message
+            control
+                .incoming
+                .send(Ok(Message::Close(None)))
+                .await
+                .unwrap();
+            // Then we expect the outgoing to close with no message
+            assert!(control.outgoing.recv().await.is_none());
+            assert!(control.incoming.is_closed());
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn ping_pong() {
+        let _ = env_logger::builder()
+            .is_test(true)
+            .filter_level(LevelFilter::Trace)
+            .try_init();
+
+        assert_client_interaction(|_, mut control, _| async move {
+            // Expect a pong in response to a ping
+            let bytes = Bytes::from_owner(Uuid::new_v4().into_bytes());
+            control
+                .incoming
+                .send(Ok(Message::Ping(bytes.clone())))
+                .await
+                .unwrap();
+            let Some(Message::Pong(pong_bytes)) = control.outgoing.recv().await else {
+                panic!("Expected Pong Response");
+            };
+            assert_eq!(bytes, pong_bytes);
+
+            // Nothing should happen
+            control
+                .incoming
+                .send(Ok(Message::Pong(bytes.clone())))
+                .await
+                .unwrap();
+
+            // We receive an incoming close message
+            control
+                .incoming
+                .send(Ok(Message::Close(None)))
+                .await
+                .unwrap();
+            // Then we expect the outgoing to close with no message
+            assert!(control.outgoing.recv().await.is_none());
+            assert!(control.incoming.is_closed());
+        })
+        .await;
+    }
+}

api/src/client/error.rs

@@ -0,0 +1,31 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum ConnectError {
+    #[error(transparent)]
+    TungsteniteError(#[from] tokio_tungstenite::tungstenite::Error),
+    #[error(transparent)]
+    IoError(#[from] std::io::Error),
+}
+
+#[derive(Error, Debug)]
+pub enum MessageError {
+    #[error(transparent)]
+    TokioSendError(#[from] tokio::sync::mpsc::error::SendError<()>),
+    #[error(transparent)]
+    TokioTrySendError(#[from] tokio::sync::mpsc::error::TrySendError<()>),
+    #[error(transparent)]
+    TokioLockError(#[from] tokio::sync::TryLockError),
+}
+
+#[derive(Error, Debug)]
+pub enum RequestError<E> {
+    #[error(transparent)]
+    TokioSendError(#[from] tokio::sync::mpsc::error::SendError<()>),
+    #[error(transparent)]
+    TokioLockError(#[from] tokio::sync::TryLockError),
+    #[error(transparent)]
+    RecvError(#[from] tokio::sync::oneshot::error::RecvError),
+    #[error(transparent)]
+    Inner(E),
+}

api/src/client/mod.rs

@@ -0,0 +1,598 @@
+pub mod command;
+mod config;
+mod context;
+pub mod error;
+pub mod telemetry;
+
+use crate::client::config::ClientConfiguration;
+use crate::client::error::{MessageError, RequestError};
+use crate::messages::callback::GenericCallbackError;
+use crate::messages::payload::RequestMessagePayload;
+use crate::messages::payload::ResponseMessagePayload;
+use crate::messages::{
+    ClientMessage, RegisterCallback, RequestMessage, RequestResponse, ResponseMessage,
+};
+use context::ClientContext;
+use error::ConnectError;
+use std::sync::Arc;
+use tokio::spawn;
+use tokio::sync::{mpsc, oneshot, watch, RwLock};
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+type RegisteredCallback = mpsc::Sender<(ResponseMessage, oneshot::Sender<RequestMessagePayload>)>;
+type ClientChannel = Arc<RwLock<mpsc::Sender<OutgoingMessage>>>;
+
+#[derive(Debug)]
+enum Callback {
+    None,
+    Once(oneshot::Sender<ResponseMessage>),
+    Registered(RegisteredCallback),
+}
+
+struct OutgoingMessage {
+    msg: RequestMessage,
+    callback: Callback,
+}
+
+pub struct Client {
+    cancel: CancellationToken,
+    channel: ClientChannel,
+    connected_state_rx: watch::Receiver<bool>,
+}
+
+impl Client {
+    pub fn connect<R>(request: R) -> Result<Self, ConnectError>
+    where
+        R: IntoClientRequest,
+    {
+        Self::connect_with_config(request, ClientConfiguration::default())
+    }
+
+    pub fn connect_with_config<R>(
+        request: R,
+        config: ClientConfiguration,
+    ) -> Result<Self, ConnectError>
+    where
+        R: IntoClientRequest,
+    {
+        let (tx, _rx) = mpsc::channel(1);
+        let cancel = CancellationToken::new();
+        let channel = Arc::new(RwLock::new(tx));
+        let (connected_state_tx, connected_state_rx) = watch::channel(false);
+        let context = ClientContext {
+            cancel: cancel.clone(),
+            request: request.into_client_request()?,
+            connected_state_tx,
+            client_configuration: config,
+        };
+
+        context.start(channel.clone())?;
+
+        Ok(Self {
+            cancel,
+            channel,
+            connected_state_rx,
+        })
+    }
+
+    pub async fn send_message<M: ClientMessage>(&self, msg: M) -> Result<(), MessageError> {
+        let sender = self.channel.read().await;
+        let data = sender.reserve().await?;
+        data.send(OutgoingMessage {
+            msg: RequestMessage {
+                uuid: Uuid::new_v4(),
+                response: None,
+                payload: msg.into(),
+            },
+            callback: Callback::None,
+        });
+        Ok(())
+    }
+
+    pub async fn send_message_if_connected<M: ClientMessage>(
+        &self,
+        msg: M,
+    ) -> Result<(), MessageError> {
+        let sender = self.channel.try_read()?;
+        let data = sender.reserve().await?;
+        data.send(OutgoingMessage {
+            msg: RequestMessage {
+                uuid: Uuid::new_v4(),
+                response: None,
+                payload: msg.into(),
+            },
+            callback: Callback::None,
+        });
+        Ok(())
+    }
+
+    pub fn try_send_message<M: ClientMessage>(&self, msg: M) -> Result<(), MessageError> {
+        let sender = self.channel.try_read()?;
+        let data = sender.try_reserve()?;
+        data.send(OutgoingMessage {
+            msg: RequestMessage {
+                uuid: Uuid::new_v4(),
+                response: None,
+                payload: msg.into(),
+            },
+            callback: Callback::None,
+        });
+        Ok(())
+    }
+
+    pub async fn send_request<M: RequestResponse>(
+        &self,
+        msg: M,
+    ) -> Result<M::Response, RequestError<<M::Response as TryFrom<ResponseMessagePayload>>::Error>>
+    {
+        let sender = self.channel.read().await;
+        let data = sender.reserve().await?;
+
+        let (tx, rx) = oneshot::channel();
+        data.send(OutgoingMessage {
+            msg: RequestMessage {
+                uuid: Uuid::new_v4(),
+                response: None,
+                payload: msg.into(),
+            },
+            callback: Callback::Once(tx),
+        });
+
+        let response = rx.await?;
+        let response = M::Response::try_from(response.payload).map_err(RequestError::Inner)?;
+
+        Ok(response)
+    }
+
+    pub async fn register_callback_channel<M: RegisterCallback>(
+        &self,
+        msg: M,
+    ) -> Result<mpsc::Receiver<(M::Callback, oneshot::Sender<M::Response>)>, MessageError>
+    where
+        <M as RegisterCallback>::Callback: Send + 'static,
+        <M as RegisterCallback>::Response: Send + 'static,
+        <<M as RegisterCallback>::Callback as TryFrom<ResponseMessagePayload>>::Error: Send,
+    {
+        let sender = self.channel.read().await;
+        let data = sender.reserve().await?;
+
+        let (inner_tx, mut inner_rx) = mpsc::channel(16);
+        let (outer_tx, outer_rx) = mpsc::channel(1);
+
+        data.send(OutgoingMessage {
+            msg: RequestMessage {
+                uuid: Uuid::new_v4(),
+                response: None,
+                payload: msg.into(),
+            },
+            callback: Callback::Registered(inner_tx),
+        });
+
+        spawn(async move {
+            // If the handler was unregistered we can stop
+            while let Some((msg, responder)) = inner_rx.recv().await {
+                let response: RequestMessagePayload = match M::Callback::try_from(msg.payload) {
+                    Err(_) => GenericCallbackError::MismatchedType.into(),
+                    Ok(o) => {
+                        let (response_tx, response_rx) = oneshot::channel::<M::Response>();
+                        match outer_tx.send((o, response_tx)).await {
+                            Err(_) => GenericCallbackError::CallbackClosed.into(),
+                            Ok(()) => response_rx
+                                .await
+                                .map(M::Response::into)
+                                .unwrap_or_else(|_| GenericCallbackError::CallbackClosed.into()),
+                        }
+                    }
+                };
+
+                if responder.send(response).is_err() {
+                    // If the callback was unregistered we can stop
+                    break;
+                }
+            }
+        });
+
+        Ok(outer_rx)
+    }
+
+    pub async fn register_callback_fn<M: RegisterCallback, F>(
+        &self,
+        msg: M,
+        mut f: F,
+    ) -> Result<(), MessageError>
+    where
+        F: FnMut(M::Callback) -> M::Response + Send + 'static,
+    {
+        let sender = self.channel.read().await;
+        let data = sender.reserve().await?;
+
+        let (inner_tx, mut inner_rx) = mpsc::channel(16);
+
+        data.send(OutgoingMessage {
+            msg: RequestMessage {
+                uuid: Uuid::new_v4(),
+                response: None,
+                payload: msg.into(),
+            },
+            callback: Callback::Registered(inner_tx),
+        });
+
+        spawn(async move {
+            // If the handler was unregistered we can stop
+            while let Some((msg, responder)) = inner_rx.recv().await {
+                let response: RequestMessagePayload = match M::Callback::try_from(msg.payload) {
+                    Err(_) => GenericCallbackError::MismatchedType.into(),
+                    Ok(o) => f(o).into(),
+                };
+
+                if responder.send(response).is_err() {
+                    // If the callback was unregistered we can stop
+                    break;
+                }
+            }
+        });
+
+        Ok(())
+    }
+
+    pub async fn wait_connected(&self) {
+        let mut connected_rx = self.connected_state_rx.clone();
+
+        // If we aren't currently connected
+        if !*connected_rx.borrow_and_update() {
+            // Wait for a change notification
+            // If the channel is closed there is nothing we can do
+            let _ = connected_rx.changed().await;
+        }
+    }
+
+    pub async fn wait_disconnected(&self) {
+        let mut connected_rx = self.connected_state_rx.clone();
+
+        // If we are currently connected
+        if *connected_rx.borrow_and_update() {
+            // Wait for a change notification
+            // If the channel is closed there is nothing we can do
+            let _ = connected_rx.changed().await;
+        }
+    }
+}
+
+impl Drop for Client {
+    fn drop(&mut self) {
+        self.cancel.cancel();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::messages::command::CommandResponse;
+    use crate::messages::telemetry_definition::{
+        TelemetryDefinitionRequest, TelemetryDefinitionResponse,
+    };
+    use crate::messages::telemetry_entry::TelemetryEntry;
+    use api_core::command::{Command, CommandDefinition, CommandHeader};
+    use api_core::data_type::DataType;
+    use chrono::Utc;
+    use futures_util::future::{select, Either};
+    use futures_util::FutureExt;
+    use std::future::Future;
+    use std::pin::{pin, Pin};
+    use std::time::Duration;
+    use tokio::join;
+    use tokio::time::{sleep, timeout};
+
+    fn create_test_client() -> (mpsc::Receiver<OutgoingMessage>, watch::Sender<bool>, Client) {
+        let cancel = CancellationToken::new();
+        let (tx, rx) = mpsc::channel(1);
+        let channel = Arc::new(RwLock::new(tx));
+        let (connected_state_tx, connected_state_rx) = watch::channel(true);
+        let client = Client {
+            cancel,
+            channel,
+            connected_state_rx,
+        };
+        (rx, connected_state_tx, client)
+    }
+
+    #[tokio::test]
+    async fn send_message() {
+        let (mut rx, _, client) = create_test_client();
+
+        let msg_to_send = TelemetryEntry {
+            uuid: Uuid::new_v4(),
+            value: 0.0f32.into(),
+            timestamp: Utc::now(),
+        };
+        let msg_send = timeout(
+            Duration::from_secs(1),
+            client.send_message(msg_to_send.clone()),
+        );
+        let msg_recv = timeout(Duration::from_secs(1), rx.recv());
+
+        let (send, recv) = join!(msg_send, msg_recv);
+        send.unwrap().unwrap();
+        let recv = recv.unwrap().unwrap();
+
+        assert!(matches!(recv.callback, Callback::None));
+        assert!(recv.msg.response.is_none());
+        // uuid should be random
+
+        let RequestMessagePayload::TelemetryEntry(recv) = recv.msg.payload else {
+            panic!("Wrong Message Received")
+        };
+
+        assert_eq!(recv, msg_to_send);
+    }
+
+    #[tokio::test]
+    async fn send_message_if_connected() {
+        let (mut rx, _, client) = create_test_client();
+
+        let msg_to_send = TelemetryEntry {
+            uuid: Uuid::new_v4(),
+            value: 0.0f32.into(),
+            timestamp: Utc::now(),
+        };
+        let msg_send = timeout(
+            Duration::from_secs(1),
+            client.send_message_if_connected(msg_to_send.clone()),
+        );
+        let msg_recv = timeout(Duration::from_secs(1), rx.recv());
+
+        let (send, recv) = join!(msg_send, msg_recv);
+        send.unwrap().unwrap();
+        let recv = recv.unwrap().unwrap();
+
+        assert!(matches!(recv.callback, Callback::None));
+        assert!(recv.msg.response.is_none());
+        // uuid should be random
+
+        let RequestMessagePayload::TelemetryEntry(recv) = recv.msg.payload else {
+            panic!("Wrong Message Received")
+        };
+
+        assert_eq!(recv, msg_to_send);
+    }
+
+    #[tokio::test]
+    async fn send_message_if_connected_not_connected() {
+        let (_, connected_state_tx, client) = create_test_client();
+
+        let lock = client.channel.write().await;
+        connected_state_tx.send_replace(false);
+
+        let msg_to_send = TelemetryEntry {
+            uuid: Uuid::new_v4(),
+            value: 0.0f32.into(),
+            timestamp: Utc::now(),
+        };
+        let msg_send = timeout(
+            Duration::from_secs(1),
+            client.send_message_if_connected(msg_to_send.clone()),
+        );
+
+        let Err(MessageError::TokioLockError(_)) = msg_send.await.unwrap() else {
+            panic!("Expected to Err due to lock being unavailable")
+        };
+    }
+
+    #[tokio::test]
+    async fn try_send_message() {
+        let (_tx, _, client) = create_test_client();
+
+        let msg_to_send = TelemetryEntry {
+            uuid: Uuid::new_v4(),
+            value: 0.0f32.into(),
+            timestamp: Utc::now(),
+        };
+        client.try_send_message(msg_to_send.clone()).unwrap();
+        let Err(MessageError::TokioTrySendError(_)) = client.try_send_message(msg_to_send.clone())
+        else {
+            panic!("Expected the buffer to be full");
+        };
+    }
+
+    #[tokio::test]
+    async fn send_request() {
+        let (mut tx, _, client) = create_test_client();
+
+        let msg_to_send = TelemetryDefinitionRequest {
+            name: "".to_string(),
+            data_type: DataType::Float32,
+        };
+        let response = timeout(
+            Duration::from_secs(1),
+            client.send_request(msg_to_send.clone()),
+        );
+
+        let response_uuid = Uuid::new_v4();
+        let outgoing_rx = timeout(Duration::from_secs(1), async {
+            let msg = tx.recv().await.unwrap();
+            let Callback::Once(cb) = msg.callback else {
+                panic!("Wrong Callback Type")
+            };
+            cb.send(ResponseMessage {
+                uuid: Uuid::new_v4(),
+                response: Some(msg.msg.uuid),
+                payload: TelemetryDefinitionResponse {
+                    uuid: response_uuid,
+                }
+                .into(),
+            })
+            .unwrap();
+        });
+
+        let (response, outgoing_rx) = join!(response, outgoing_rx);
+        let response = response.unwrap().unwrap();
+        outgoing_rx.unwrap();
+
+        assert_eq!(response.uuid, response_uuid);
+    }
+
+    #[tokio::test]
+    async fn register_callback_channel() {
+        let (mut tx, _, client) = create_test_client();
+
+        let msg_to_send = CommandDefinition {
+            name: "".to_string(),
+            parameters: vec![],
+        };
+        let mut response = timeout(
+            Duration::from_secs(1),
+            client.register_callback_channel(msg_to_send),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+
+        let response_uuid = Uuid::new_v4();
+        let outgoing_rx = timeout(Duration::from_secs(1), async {
+            let msg = tx.recv().await.unwrap();
+            let Callback::Registered(cb) = msg.callback else {
+                panic!("Wrong Callback Type")
+            };
+
+            // Check that we get responses to the callback the expected number of times
+            for i in 0..5 {
+                let (tx, rx) = oneshot::channel();
+                cb.send((
+                    ResponseMessage {
+                        uuid: Uuid::new_v4(),
+                        response: Some(msg.msg.uuid),
+                        payload: Command {
+                            header: CommandHeader {
+                                timestamp: Utc::now(),
+                            },
+                            parameters: Default::default(),
+                        }
+                        .into(),
+                    },
+                    tx,
+                ))
+                .await
+                .unwrap();
+                let RequestMessagePayload::CommandResponse(response) = rx.await.unwrap() else {
+                    panic!("Unexpected Response Type");
+                };
+                assert_eq!(response.response, format!("{i}"));
+            }
+        });
+
+        let responder = timeout(Duration::from_secs(1), async {
+            for i in 0..5 {
+                let (cmd, responder) = response.recv().await.unwrap();
+                responder
+                    .send(CommandResponse {
+                        success: false,
+                        response: format!("{i}"),
+                    })
+                    .unwrap();
+            }
+        });
+
+        let (response, outgoing_rx) = join!(responder, outgoing_rx);
+        response.unwrap();
+        outgoing_rx.unwrap();
+    }
+
+    #[tokio::test]
+    async fn register_callback_fn() {
+        let (mut tx, _, client) = create_test_client();
+
+        let msg_to_send = CommandDefinition {
+            name: "".to_string(),
+            parameters: vec![],
+        };
+        let mut index = 0usize;
+        timeout(
+            Duration::from_secs(1),
+            client.register_callback_fn(msg_to_send, move |cmd| {
+                index += 1;
+                CommandResponse {
+                    success: false,
+                    response: format!("{}", index - 1),
+                }
+            }),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+
+        timeout(Duration::from_secs(1), async {
+            let msg = tx.recv().await.unwrap();
+            let Callback::Registered(cb) = msg.callback else {
+                panic!("Wrong Callback Type")
+            };
+
+            // Check that we get responses to the callback the expected number of times
+            for i in 0..3 {
+                let (tx, rx) = oneshot::channel();
+                cb.send((
+                    ResponseMessage {
+                        uuid: Uuid::new_v4(),
+                        response: Some(msg.msg.uuid),
+                        payload: Command {
+                            header: CommandHeader {
+                                timestamp: Utc::now(),
+                            },
+                            parameters: Default::default(),
+                        }
+                        .into(),
+                    },
+                    tx,
+                ))
+                .await
+                .unwrap();
+                let RequestMessagePayload::CommandResponse(response) = rx.await.unwrap() else {
+                    panic!("Unexpected Response Type");
+                };
+                assert_eq!(response.response, format!("{i}"));
+            }
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn connected_disconnected() {
+        let (_, connected, client) = create_test_client();
+
+        // When we're connected we should return immediately
+        connected.send_replace(true);
+        client.wait_connected().now_or_never().unwrap();
+
+        // When we're disconnected we should return immediately
+        connected.send_replace(false);
+        client.wait_disconnected().now_or_never().unwrap();
+
+        let c2 = connected.clone();
+        // When we're disconnected, we should not return immediately
+        let f1 = pin!(client.wait_connected());
+        let f2 = pin!(async move {
+            sleep(Duration::from_millis(1)).await;
+            c2.send_replace(true);
+        });
+        let r = select(f1, f2).await;
+        match r {
+            Either::Left(_) => panic!("Wait Connected Finished Before Connection Changed"),
+            Either::Right((_, other)) => timeout(Duration::from_secs(1), other).await.unwrap(),
+        }
+
+        let c2 = connected.clone();
+        // When we're disconnected, we should not return immediately
+        let f1 = pin!(client.wait_disconnected());
+        let f2 = pin!(async move {
+            sleep(Duration::from_millis(1)).await;
+            c2.send_replace(false);
+        });
+        let r = select(f1, f2).await;
+        match r {
+            Either::Left(_) => panic!("Wait Disconnected Finished Before Connection Changed"),
+            Either::Right((_, other)) => timeout(Duration::from_secs(1), other).await.unwrap(),
+        }
+    }
+}

api/src/client/telemetry.rs

@@ -0,0 +1,168 @@
+use crate::client::error::MessageError;
+use crate::client::Client;
+use crate::data_value::DataValue;
+use crate::messages::telemetry_definition::TelemetryDefinitionRequest;
+use crate::messages::telemetry_entry::TelemetryEntry;
+use api_core::data_type::{DataType, ToDataType};
+use chrono::{DateTime, Utc};
+use std::marker::PhantomData;
+use std::sync::Arc;
+use tokio::sync::{oneshot, RwLock};
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+pub struct TelemetryRegistry {
+    client: Arc<Client>,
+}
+
+impl TelemetryRegistry {
+    pub fn new(client: Arc<Client>) -> Self {
+        Self { client }
+    }
+
+    #[inline]
+    pub async fn register_generic(
+        &self,
+        name: impl Into<String>,
+        data_type: DataType,
+    ) -> GenericTelemetryHandle {
+        // inner for compilation performance
+        async fn inner(
+            client: Arc<Client>,
+            name: String,
+            data_type: DataType,
+        ) -> GenericTelemetryHandle {
+            let cancellation_token = CancellationToken::new();
+            let cancel_token = cancellation_token.clone();
+            let stored_client = client.clone();
+
+            let response_uuid = Arc::new(RwLock::new(Uuid::nil()));
+
+            let response_uuid_inner = response_uuid.clone();
+            let (tx, rx) = oneshot::channel();
+
+            tokio::spawn(async move {
+                let mut write_lock = Some(response_uuid_inner.write().await);
+                let _ = tx.send(());
+                while !cancel_token.is_cancelled() {
+                    if let Ok(response) = client
+                        .send_request(TelemetryDefinitionRequest {
+                            name: name.clone(),
+                            data_type,
+                        })
+                        .await
+                    {
+                        let mut lock = match write_lock {
+                            None => response_uuid_inner.write().await,
+                            Some(lock) => lock,
+                        };
+                        // Update the value in the lock
+                        *lock = response.uuid;
+                        // Set this value so the loop works
+                        write_lock = None;
+                    }
+
+                    client.wait_disconnected().await;
+                }
+            });
+
+            // Wait until the write lock is acquired
+            let _ = rx.await;
+            // Wait until the write lock is released for the first time
+            drop(response_uuid.read().await);
+
+            GenericTelemetryHandle {
+                cancellation_token,
+                uuid: response_uuid,
+                client: stored_client,
+            }
+        }
+        inner(self.client.clone(), name.into(), data_type).await
+    }
+
+    #[inline]
+    pub async fn register<T: ToDataType>(&self, name: impl Into<String>) -> TelemetryHandle<T> {
+        self.register_generic(name, T::DATA_TYPE).await.coerce()
+    }
+}
+
+impl Drop for GenericTelemetryHandle {
+    fn drop(&mut self) {
+        self.cancellation_token.cancel();
+    }
+}
+
+pub struct GenericTelemetryHandle {
+    cancellation_token: CancellationToken,
+    uuid: Arc<RwLock<Uuid>>,
+    client: Arc<Client>,
+}
+
+impl GenericTelemetryHandle {
+    pub async fn publish(
+        &self,
+        value: DataValue,
+        timestamp: DateTime<Utc>,
+    ) -> Result<(), MessageError> {
+        let Ok(lock) = self.uuid.try_read() else {
+            return Ok(());
+        };
+        let uuid = *lock;
+        drop(lock);
+
+        self.client
+            .send_message_if_connected(TelemetryEntry {
+                uuid,
+                value,
+                timestamp,
+            })
+            .await
+            .or_else(|e| match e {
+                MessageError::TokioLockError(_) => Ok(()),
+                e => Err(e),
+            })?;
+
+        Ok(())
+    }
+
+    #[inline]
+    pub async fn publish_now(&self, value: DataValue) -> Result<(), MessageError> {
+        self.publish(value, Utc::now()).await
+    }
+
+    fn coerce<T: Into<DataValue>>(self) -> TelemetryHandle<T> {
+        TelemetryHandle::<T> {
+            generic_handle: self,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+pub struct TelemetryHandle<T> {
+    generic_handle: GenericTelemetryHandle,
+    _phantom: PhantomData<T>,
+}
+
+impl<T> TelemetryHandle<T> {
+    pub fn to_generic(self) -> GenericTelemetryHandle {
+        self.generic_handle
+    }
+    pub fn as_generic(&self) -> &GenericTelemetryHandle {
+        &self.generic_handle
+    }
+}
+
+impl<T: Into<DataValue>> TelemetryHandle<T> {
+    #[inline]
+    pub async fn publish(&self, value: T, timestamp: DateTime<Utc>) -> Result<(), MessageError> {
+        self.as_generic().publish(value.into(), timestamp).await
+    }
+
+    #[inline]
+    pub async fn publish_now(&self, value: T) -> Result<(), MessageError> {
+        self.publish(value, Utc::now()).await
+    }
+}
+
+#[cfg(test)]
+mod tests {}

api/src/lib.rs

@@ -0,0 +1,15 @@
+pub mod client;
+pub mod data_type {
+    pub use api_core::data_type::*;
+}
+pub mod data_value {
+    pub use api_core::data_value::*;
+}
+pub mod messages;
+
+pub mod macros {
+    pub use api_proc_macro::IntoCommandDefinition;
+}
+
+#[cfg(test)]
+pub mod test;

api/src/messages/callback.rs

@@ -0,0 +1,7 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum GenericCallbackError {
+    CallbackClosed,
+    MismatchedType,
+}

api/src/messages/command.rs

@@ -0,0 +1,15 @@
+use crate::messages::RegisterCallback;
+use serde::{Deserialize, Serialize};
+
+pub use api_core::command::*;
+
+impl RegisterCallback for CommandDefinition {
+    type Callback = Command;
+    type Response = CommandResponse;
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CommandResponse {
+    pub success: bool,
+    pub response: String,
+}

api/src/messages/mod.rs

@@ -0,0 +1,40 @@
+pub mod callback;
+pub mod command;
+pub mod payload;
+pub mod telemetry_definition;
+pub mod telemetry_entry;
+
+use payload::{RequestMessagePayload, ResponseMessagePayload};
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RequestMessage {
+    pub uuid: Uuid,
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub response: Option<Uuid>,
+    #[serde(flatten)]
+    pub payload: RequestMessagePayload,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ResponseMessage {
+    pub uuid: Uuid,
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub response: Option<Uuid>,
+    #[serde(flatten)]
+    pub payload: ResponseMessagePayload,
+}
+
+pub trait ClientMessage: Into<RequestMessagePayload> {}
+
+pub trait RequestResponse: Into<RequestMessagePayload> {
+    type Response: TryFrom<ResponseMessagePayload>;
+}
+
+pub trait RegisterCallback: Into<RequestMessagePayload> {
+    type Callback: TryFrom<ResponseMessagePayload>;
+    type Response: Into<RequestMessagePayload>;
+}

api/src/messages/payload.rs

@@ -0,0 +1,23 @@
+use crate::messages::callback::GenericCallbackError;
+use crate::messages::command::{Command, CommandDefinition, CommandResponse};
+use crate::messages::telemetry_definition::{
+    TelemetryDefinitionRequest, TelemetryDefinitionResponse,
+};
+use crate::messages::telemetry_entry::TelemetryEntry;
+use derive_more::{From, TryInto};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, From)]
+pub enum RequestMessagePayload {
+    TelemetryDefinitionRequest(TelemetryDefinitionRequest),
+    TelemetryEntry(TelemetryEntry),
+    GenericCallbackError(GenericCallbackError),
+    CommandDefinition(CommandDefinition),
+    CommandResponse(CommandResponse),
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, From, TryInto)]
+pub enum ResponseMessagePayload {
+    TelemetryDefinitionResponse(TelemetryDefinitionResponse),
+    Command(Command),
+}

api/src/messages/telemetry_definition.rs

@@ -0,0 +1,19 @@
+use crate::data_type::DataType;
+use crate::messages::RequestResponse;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TelemetryDefinitionRequest {
+    pub name: String,
+    pub data_type: DataType,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TelemetryDefinitionResponse {
+    pub uuid: Uuid,
+}
+
+impl RequestResponse for TelemetryDefinitionRequest {
+    type Response = TelemetryDefinitionResponse;
+}

api/src/messages/telemetry_entry.rs

@@ -0,0 +1,14 @@
+use crate::data_value::DataValue;
+use crate::messages::ClientMessage;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct TelemetryEntry {
+    pub uuid: Uuid,
+    pub value: DataValue,
+    pub timestamp: DateTime<Utc>,
+}
+
+impl ClientMessage for TelemetryEntry {}

api/src/test/mock_stream_sink.rs

@@ -0,0 +1,82 @@
+use futures_util::sink::{unfold, Unfold};
+use futures_util::{Sink, SinkExt, Stream};
+use std::fmt::Display;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use tokio::sync::mpsc;
+use tokio::sync::mpsc::error::SendError;
+use tokio::sync::mpsc::{Receiver, Sender};
+
+pub struct MockStreamSinkControl<T, R> {
+    pub incoming: Sender<T>,
+    pub outgoing: Receiver<R>,
+}
+
+pub struct MockStreamSink<T, U1, U2> {
+    stream_rx: Receiver<T>,
+    sink_tx: Pin<Box<Unfold<u32, U1, U2>>>,
+}
+
+impl<T, U1, U2> Stream for MockStreamSink<T, U1, U2>
+where
+    Self: Unpin,
+{
+    type Item = T;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.stream_rx.poll_recv(cx)
+    }
+}
+
+impl<T, R, U1, U2, E> Sink<R> for MockStreamSink<T, U1, U2>
+where
+    U1: FnMut(u32, R) -> U2,
+    U2: Future<Output = Result<u32, E>>,
+{
+    type Error = E;
+
+    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.sink_tx.poll_ready_unpin(cx)
+    }
+
+    fn start_send(mut self: Pin<&mut Self>, item: R) -> Result<(), Self::Error> {
+        self.sink_tx.start_send_unpin(item)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.sink_tx.poll_flush_unpin(cx)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.sink_tx.poll_close_unpin(cx)
+    }
+}
+
+pub fn create_mock_stream_sink<T: Send, R: Send + 'static>() -> (
+    MockStreamSinkControl<T, R>,
+    impl Stream<Item = T> + Sink<R, Error = impl Display>,
+) {
+    let (stream_tx, stream_rx) = mpsc::channel::<T>(1);
+    let (sink_tx, sink_rx) = mpsc::channel::<R>(1);
+
+    let sink_tx = Arc::new(sink_tx);
+
+    (
+        MockStreamSinkControl {
+            incoming: stream_tx,
+            outgoing: sink_rx,
+        },
+        MockStreamSink::<T, _, _> {
+            stream_rx,
+            sink_tx: Box::pin(unfold(0u32, move |_, item| {
+                let sink_tx = sink_tx.clone();
+                async move {
+                    sink_tx.send(item).await?;
+                    Ok(0u32) as Result<_, SendError<R>>
+                }
+            })),
+        },
+    )
+}

api/src/test/mod.rs

@@ -0,0 +1 @@
+pub mod mock_stream_sink;

examples/simple_command/Cargo.toml

@@ -4,11 +4,9 @@ name = "simple_command"
 edition = "2021"
 
 [dependencies]
-server = { path = "../../server" }
-tonic = "0.12.3"
-tokio = { version = "1.43.0", features = ["rt-multi-thread", "signal"] }
-chrono = "0.4.39"
-tokio-util = "0.7.13"
-num-traits = "0.2.19"
-log = "0.4.29"
-anyhow = "1.0.100"
+anyhow = { workspace = true }
+api = { path = "../../api" }
+env_logger = { workspace = true }
+log = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "signal"] }
+tokio-util = { workspace = true }

examples/simple_command/src/main.rs

@@ -1,82 +1,16 @@
-use chrono::DateTime;
-use server::core::client_side_command::Inner;
-use server::core::command_service_client::CommandServiceClient;
-use server::core::telemetry_value::Value;
-use server::core::{
-    ClientSideCommand, Command, CommandDefinitionRequest, CommandParameterDefinition,
-    CommandResponse, TelemetryDataType,
-};
+use api::client::command::CommandRegistry;
+use api::client::Client;
+use api::macros::IntoCommandDefinition;
+use api::messages::command::CommandHeader;
+use log::info;
 use std::error::Error;
-use tokio::select;
-use tokio::sync::mpsc;
-use tokio::task::JoinHandle;
+use std::sync::Arc;
 use tokio_util::sync::CancellationToken;
-use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
-use tonic::codegen::tokio_stream::StreamExt;
-use tonic::transport::Channel;
-
-struct CommandHandler {
-    handle: JoinHandle<()>,
-}
-
-impl CommandHandler {
-    pub async fn new<F: Fn(Command) -> CommandResponse + Send + 'static>(
-        cancellation_token: CancellationToken,
-        client: &mut CommandServiceClient<Channel>,
-        command_definition_request: CommandDefinitionRequest,
-        handler: F,
-    ) -> anyhow::Result<Self> {
-        let (tx, rx) = mpsc::channel(4);
-
-        // The buffer size of 4 means this is safe to send immediately
-        tx.send(ClientSideCommand {
-            inner: Some(Inner::Request(command_definition_request)),
-        })
-        .await?;
-        let response = client.new_command(ReceiverStream::new(rx)).await?;
-        let mut cmd_stream = response.into_inner();
-
-        let handle = tokio::spawn(async move {
-            loop {
-                select! {
-                    _ = cancellation_token.cancelled() => break,
-                    Some(msg) = cmd_stream.next() => {
-                        match msg {
-                            Ok(cmd) => {
-                                let uuid = cmd.uuid.clone();
-                                let mut response = handler(cmd);
-                                response.uuid = uuid;
-                                match tx.send(ClientSideCommand {
-                                    inner: Some(Inner::Response(response))
-                                }).await {
-                                    Ok(()) => {},
-                                    Err(e) => {
-                                        println!("SendError: {e}");
-                                        break;
-                                    }
-                                }
-                            }
-                            Err(e) => {
-                                println!("Error: {e}");
-                                break;
-                            }
-                        }
-                    },
-                    else => break,
-                }
-            }
-        });
-
-        Ok(Self { handle })
-    }
-
-    pub async fn join(self) -> anyhow::Result<()> {
-        Ok(self.handle.await?)
-    }
-}
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
+    env_logger::init();
+
     let cancellation_token = CancellationToken::new();
     {
         let cancellation_token = cancellation_token.clone();
@@ -86,56 +20,34 @@ async fn main() -> Result<(), Box<dyn Error>> {
         });
     }
 
-    let mut client = CommandServiceClient::connect("http://[::1]:50051").await?;
+    let client = Arc::new(Client::connect("ws://localhost:8080/backend")?);
+    let cmd = CommandRegistry::new(client);
 
-    let cmd_handler = CommandHandler::new(
-        cancellation_token,
-        &mut client,
-        CommandDefinitionRequest {
-            name: "simple_command/a".to_string(),
-            parameters: vec![
-                CommandParameterDefinition {
-                    name: "a".to_string(),
-                    data_type: TelemetryDataType::Float32.into(),
-                },
-                CommandParameterDefinition {
-                    name: "b".to_string(),
-                    data_type: TelemetryDataType::Float64.into(),
-                },
-                CommandParameterDefinition {
-                    name: "c".to_string(),
-                    data_type: TelemetryDataType::Boolean.into(),
-                },
-            ],
-        },
-        |command| {
-            let timestamp = command.timestamp.expect("Missing Timestamp");
-            let timestamp = DateTime::from_timestamp(timestamp.secs, timestamp.nanos as u32)
-                .expect("Could not construct date time");
-            let Value::Float32(a) = command.parameters["a"].value.expect("Missing Value a") else {
-                panic!("Wrong Type a");
-            };
-            let Value::Float64(b) = command.parameters["b"].value.expect("Missing Value b") else {
-                panic!("Wrong Type b");
-            };
-            let Value::Boolean(c) = command.parameters["c"].value.expect("Missing Value c") else {
-                panic!("Wrong Type c");
-            };
+    let handle = cmd.register_handler("simple_command/a", handle_command);
 
-            println!("Command Received:\n  timestamp: {timestamp}\n  a: {a}\n  b: {b}\n  c: {c}");
+    cancellation_token.cancelled().await;
 
-            CommandResponse {
-                uuid: command.uuid.clone(),
-                success: true,
-                response: format!(
-                    "Successfully Received Command! timestamp: {timestamp} a: {a} b: {b} c: {c}"
-                ),
-            }
-        },
-    )
-    .await?;
-
-    cmd_handler.join().await?;
+    // This will automatically drop when we return
+    drop(handle);
 
     Ok(())
 }
+
+#[derive(IntoCommandDefinition)]
+struct SimpleCommandA {
+    a: f32,
+    b: f64,
+    c: bool,
+}
+
+fn handle_command(header: CommandHeader, command: SimpleCommandA) -> anyhow::Result<String> {
+    let timestamp = header.timestamp;
+    let SimpleCommandA { a, b, c } = command;
+
+    info!("Command Received:\n  timestamp: {timestamp}\n  a: {a}\n  b: {b}\n  c: {c}");
+
+    // This gets sent back to the source of the command
+    Ok(format!(
+        "Successfully Received Command! timestamp: {timestamp} a: {a} b: {b} c: {c}"
+    ))
+}

examples/simple_producer/Cargo.toml

@@ -4,9 +4,13 @@ name = "simple_producer"
 edition = "2021"
 
 [dependencies]
-server = { path = "../../server" }
-tonic = "0.12.3"
-tokio = { version = "1.43.0", features = ["rt-multi-thread", "signal"] }
-chrono = "0.4.39"
-tokio-util = "0.7.13"
-num-traits = "0.2.19"
+anyhow = { workspace = true }
+api = { path = "../../api" }
+chrono = { workspace = true }
+env_logger = { workspace = true }
+futures-util = { workspace = true }
+num-traits = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "signal", "time", "macros"] }
+tokio-util = { workspace = true }
+uuid = { workspace = true }
+log = "0.4.29"

examples/simple_producer/src/main.rs

@@ -1,213 +1,44 @@
-use chrono::DateTime;
-use num_traits::float::FloatConst;
-use server::core::telemetry_service_client::TelemetryServiceClient;
-use server::core::telemetry_value::Value;
-use server::core::{
-    TelemetryDataType, TelemetryDefinitionRequest, TelemetryItem, TelemetryValue, Timestamp, Uuid,
-};
-use std::error::Error;
+use api::client::telemetry::TelemetryRegistry;
+use api::client::Client;
+use chrono::{TimeDelta, Utc};
+use futures_util::future::join_all;
+use num_traits::FloatConst;
+use std::f64;
+use std::sync::Arc;
 use std::time::Duration;
-use tokio::select;
-use tokio::sync::mpsc;
-use tokio::sync::mpsc::Sender;
-use tokio::time::Instant;
+use tokio::time::{sleep_until, Instant};
 use tokio_util::sync::CancellationToken;
-use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
-use tonic::codegen::tokio_stream::StreamExt;
-use tonic::codegen::StdError;
-use tonic::transport::Channel;
-use tonic::Request;
-
-struct Telemetry {
-    client: TelemetryServiceClient<Channel>,
-    tx: Sender<TelemetryItem>,
-    cancel: CancellationToken,
-}
-
-struct TelemetryItemHandle {
-    uuid: String,
-    tx: Sender<TelemetryItem>,
-}
-
-impl Telemetry {
-    pub async fn new<D>(dst: D) -> Result<Self, Box<dyn Error>>
-    where
-        D: TryInto<tonic::transport::Endpoint>,
-        D::Error: Into<StdError>,
-    {
-        let mut client = TelemetryServiceClient::connect(dst).await?;
-        let client_stored = client.clone();
-        let cancel = CancellationToken::new();
-        let cancel_stored = cancel.clone();
-        let (local_tx, mut local_rx) = mpsc::channel(16);
-
-        tokio::spawn(async move {
-            while !cancel.is_cancelled() {
-                let (server_tx, server_rx) = mpsc::channel(16);
-                let response_stream = client
-                    .insert_telemetry(ReceiverStream::new(server_rx))
-                    .await;
-                if let Ok(response_stream) = response_stream {
-                    let mut response_stream = response_stream.into_inner();
-                    loop {
-                        select! {
-                            _ = cancel.cancelled() => {
-                                break;
-                            },
-                            Some(item) = local_rx.recv() => {
-                                match server_tx.send(item).await {
-                                    Ok(_) => {}
-                                    Err(_) => break,
-                                }
-                            },
-                            Some(response) = response_stream.next() => {
-                                match response {
-                                    Ok(_) => {}
-                                    Err(_) => {
-                                        break;
-                                    }
-                                }
-                            },
-                            else => break,
-                        }
-                    }
-                } else {
-                    tokio::time::sleep(Duration::from_secs(1)).await;
-                }
-            }
-        });
-
-        Ok(Self {
-            client: client_stored,
-            tx: local_tx,
-            cancel: cancel_stored,
-        })
-    }
-
-    pub async fn register(
-        &mut self,
-        name: String,
-        data_type: TelemetryDataType,
-    ) -> Result<TelemetryItemHandle, Box<dyn Error>> {
-        let response = self
-            .client
-            .new_telemetry(Request::new(TelemetryDefinitionRequest {
-                name,
-                data_type: data_type.into(),
-            }))
-            .await?
-            .into_inner();
-
-        let Some(uuid) = response.uuid else {
-            return Err("UUID Missing".into());
-        };
-
-        Ok(TelemetryItemHandle {
-            uuid: uuid.value,
-            tx: self.tx.clone(),
-        })
-    }
-}
-
-impl Drop for Telemetry {
-    fn drop(&mut self) {
-        self.cancel.cancel();
-    }
-}
-
-impl TelemetryItemHandle {
-    pub async fn publish(
-        &self,
-        value: Value,
-        timestamp: DateTime<chrono::Utc>,
-    ) -> Result<(), Box<dyn Error>> {
-        let offset_from_unix_epoch =
-            timestamp - DateTime::from_timestamp(0, 0).expect("Could not get Unix epoch");
-        self.tx
-            .send(TelemetryItem {
-                uuid: Some(Uuid {
-                    value: self.uuid.clone(),
-                }),
-                value: Some(TelemetryValue { value: Some(value) }),
-                timestamp: Some(Timestamp {
-                    secs: offset_from_unix_epoch.num_seconds(),
-                    nanos: offset_from_unix_epoch.subsec_nanos(),
-                }),
-            })
-            .await?;
-        Ok(())
-    }
-}
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    let mut tlm = Telemetry::new("http://[::1]:50051").await?;
+async fn main() -> anyhow::Result<()> {
+    env_logger::init();
 
-    let index_handle = tlm
-        .register(
-            "simple_producer/time_offset".into(),
-            TelemetryDataType::Float64,
-        )
-        .await?;
+    let client = Arc::new(Client::connect("ws://localhost:8080/backend")?);
+    let tlm = TelemetryRegistry::new(client);
 
-    let publish_offset = tlm
-        .register(
-            "simple_producer/publish_offset".into(),
-            TelemetryDataType::Float64,
-        )
-        .await?;
+    let time_offset = tlm.register::<f64>("simple_producer/time_offset").await;
 
-    let await_offset = tlm
-        .register(
-            "simple_producer/await_offset".into(),
-            TelemetryDataType::Float64,
-        )
-        .await?;
+    let publish_offset = tlm.register::<f64>("simple_producer/publish_offset").await;
 
-    let sin_tlm_handle = tlm
-        .register("simple_producer/sin".into(), TelemetryDataType::Float32)
-        .await?;
-    let cos_tlm_handle = tlm
-        .register("simple_producer/cos".into(), TelemetryDataType::Float64)
-        .await?;
-    let bool_tlm_handle = tlm
-        .register("simple_producer/bool".into(), TelemetryDataType::Boolean)
-        .await?;
+    let await_offset = tlm.register::<f64>("simple_producer/await_offset").await;
 
-    let sin2_tlm_handle = tlm
-        .register("simple_producer/sin2".into(), TelemetryDataType::Float32)
-        .await?;
-    let cos2_tlm_handle = tlm
-        .register("simple_producer/cos2".into(), TelemetryDataType::Float64)
-        .await?;
+    let bool_tlm_handle = tlm.register::<bool>("simple_producer/bool").await;
 
-    let sin3_tlm_handle = tlm
-        .register("simple_producer/sin3".into(), TelemetryDataType::Float32)
-        .await?;
-    let cos3_tlm_handle = tlm
-        .register("simple_producer/cos3".into(), TelemetryDataType::Float64)
-        .await?;
+    let sin_handles = join_all((1..=6).map(|i| {
+        tlm.register::<f32>(format!(
+            "simple_producer/sin{}",
+            if i == 1 { "".into() } else { i.to_string() }
+        ))
+    }))
+    .await;
 
-    let sin4_tlm_handle = tlm
-        .register("simple_producer/sin4".into(), TelemetryDataType::Float32)
-        .await?;
-    let cos4_tlm_handle = tlm
-        .register("simple_producer/cos4".into(), TelemetryDataType::Float64)
-        .await?;
-
-    let sin5_tlm_handle = tlm
-        .register("simple_producer/sin5".into(), TelemetryDataType::Float32)
-        .await?;
-    let cos5_tlm_handle = tlm
-        .register("simple_producer/cos5".into(), TelemetryDataType::Float64)
-        .await?;
-
-    let sin6_tlm_handle = tlm
-        .register("simple_producer/sin6".into(), TelemetryDataType::Float32)
-        .await?;
-    let cos6_tlm_handle = tlm
-        .register("simple_producer/cos6".into(), TelemetryDataType::Float64)
-        .await?;
+    let cos_handles = join_all((1..=6).map(|i| {
+        tlm.register::<f64>(format!(
+            "simple_producer/cos{}",
+            if i == 1 { "".into() } else { i.to_string() }
+        ))
+    }))
+    .await;
 
     let cancellation_token = CancellationToken::new();
     {
@@ -218,85 +49,48 @@ async fn main() -> Result<(), Box<dyn Error>> {
         });
     }
 
-    let start_time = chrono::Utc::now();
+    let start_time = Utc::now();
     let start_instant = Instant::now();
     let mut next_time = start_instant;
     let mut index = 0;
-    let mut tasks = vec![];
     while !cancellation_token.is_cancelled() {
         next_time += Duration::from_millis(10);
         index += 1;
-        tokio::time::sleep_until(next_time).await;
-        let publish_time =
-            start_time + chrono::TimeDelta::from_std(next_time - start_instant).unwrap();
+        sleep_until(next_time).await;
+        let publish_time = start_time + TimeDelta::from_std(next_time - start_instant).unwrap();
         let actual_time = Instant::now();
-        tasks.push(index_handle.publish(
-            Value::Float64((actual_time - next_time).as_secs_f64()),
-            chrono::Utc::now(),
-        ));
-        tasks.push(sin_tlm_handle.publish(
-            Value::Float32((f32::TAU() * (index as f32) / (1000.0_f32)).sin()),
-            publish_time,
-        ));
-        tasks.push(cos_tlm_handle.publish(
-            Value::Float64((f64::TAU() * (index as f64) / (1000.0_f64)).cos()),
-            publish_time,
-        ));
-        tasks.push(bool_tlm_handle.publish(Value::Boolean(index % 1000 > 500), publish_time));
-        tasks.push(sin2_tlm_handle.publish(
-            Value::Float32((f32::TAU() * (index as f32) / (500.0_f32)).sin()),
-            publish_time,
-        ));
-        tasks.push(cos2_tlm_handle.publish(
-            Value::Float64((f64::TAU() * (index as f64) / (500.0_f64)).cos()),
-            publish_time,
-        ));
-        tasks.push(sin3_tlm_handle.publish(
-            Value::Float32((f32::TAU() * (index as f32) / (333.0_f32)).sin()),
-            publish_time,
-        ));
-        tasks.push(cos3_tlm_handle.publish(
-            Value::Float64((f64::TAU() * (index as f64) / (333.0_f64)).cos()),
-            publish_time,
-        ));
-        tasks.push(sin4_tlm_handle.publish(
-            Value::Float32((f32::TAU() * (index as f32) / (250.0_f32)).sin()),
-            publish_time,
-        ));
-        tasks.push(cos4_tlm_handle.publish(
-            Value::Float64((f64::TAU() * (index as f64) / (250.0_f64)).cos()),
-            publish_time,
-        ));
-        tasks.push(sin5_tlm_handle.publish(
-            Value::Float32((f32::TAU() * (index as f32) / (200.0_f32)).sin()),
-            publish_time,
-        ));
-        tasks.push(cos5_tlm_handle.publish(
-            Value::Float64((f64::TAU() * (index as f64) / (200.0_f64)).cos()),
-            publish_time,
-        ));
-        tasks.push(sin6_tlm_handle.publish(
-            Value::Float32((f32::TAU() * (index as f32) / (166.0_f32)).sin()),
-            publish_time,
-        ));
-        tasks.push(cos6_tlm_handle.publish(
-            Value::Float64((f64::TAU() * (index as f64) / (166.0_f64)).cos()),
-            publish_time,
-        ));
 
-        tasks.push(publish_offset.publish(
-            Value::Float64((Instant::now() - actual_time).as_secs_f64()),
-            chrono::Utc::now(),
-        ));
+        // Due to how telemetry handles are implemented, unless the send buffer is full awaiting
+        // these will return immediately
+        time_offset
+            .publish_now((actual_time - next_time).as_secs_f64())
+            .await?;
+        bool_tlm_handle
+            .publish(index % 1000 > 500, publish_time)
+            .await?;
 
-        for task in tasks.drain(..) {
-            task.await?;
+        for (i, sin) in sin_handles.iter().enumerate() {
+            sin.publish(
+                (f32::TAU() * (index as f32) / (1000.0_f32 / (i + 1) as f32)).sin(),
+                publish_time,
+            )
+            .await?;
+        }
+        for (i, cos) in cos_handles.iter().enumerate() {
+            cos.publish(
+                (f64::TAU() * (index as f64) / (1000.0_f64 / (i + 1) as f64)).cos(),
+                publish_time,
+            )
+            .await?;
         }
 
-        tasks.push(await_offset.publish(
-            Value::Float64((Instant::now() - actual_time).as_secs_f64()),
-            chrono::Utc::now(),
-        ));
+        publish_offset
+            .publish((Instant::now() - actual_time).as_secs_f64(), Utc::now())
+            .await?;
+
+        await_offset
+            .publish((Instant::now() - actual_time).as_secs_f64(), Utc::now())
+            .await?;
     }
 
     Ok(())

server/Cargo.toml

@@ -6,25 +6,20 @@ version = "0.1.0"
 authors = ["Sergey <me@sergeysav.com>"]
 
 [dependencies]
-fern = "0.7.1"
-log = "0.4.29"
-prost = "0.13.5"
-rand = "0.9.0"
-tonic = { version = "0.12.3" }
-tokio = { version = "1.43.0", features = ["rt-multi-thread", "signal", "fs"] }
-chrono = "0.4.42"
-actix-web = { version = "4.12.1", features = [ ] }
-actix-ws = "0.3.0"
-tokio-util = "0.7.17"
-serde = { version = "1.0.228", features = ["derive"] }
-serde_json = "1.0.145"
-hex = "0.4.3"
-papaya = "0.2.3"
-thiserror = "2.0.17"
-derive_more = { version = "2.1.0", features = ["from"] }
-anyhow = "1.0.100"
-sqlx = { version = "0.8.6", features = [ "runtime-tokio", "tls-native-tls", "sqlite" ] }
-uuid = { version = "1.19.0", features = ["v4"] }
-
-[build-dependencies]
-tonic-build = "0.12.3"
+actix-web = { workspace = true, features = [ ] }
+actix-ws = { workspace = true }
+anyhow = { workspace = true }
+api = { path = "../api" }
+chrono = { workspace = true }
+derive_more = { workspace = true, features = ["from"] }
+fern = { workspace = true }
+futures-util = { workspace = true }
+log = { workspace = true }
+papaya = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+sqlx = { workspace = true, features = [ "runtime-tokio", "tls-rustls-ring-native-roots", "sqlite" ] }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "signal", "fs"] }
+tokio-util = { workspace = true }
+uuid = { workspace = true }

server/build.rs

@@ -1,5 +1,4 @@
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("cargo:rerun-if-changed=migrations");
-    tonic_build::compile_protos("proto/core.proto")?;
     Ok(())
 }

server/proto/core.proto

@@ -1,83 +0,0 @@
-
-syntax = "proto3";
-package core;
-
-enum TelemetryDataType {
-    Float32 = 0;
-    Float64 = 1;
-    Boolean = 2;
-}
-
-message TelemetryValue {
-    oneof value {
-        float float_32 = 1;
-        double float_64 = 2;
-        bool boolean = 3;
-    }
-}
-
-message UUID {
-    string value = 1;
-}
-
-// UTC since UNIX
-message Timestamp {
-    sfixed64 secs = 1;
-    sfixed32 nanos = 2;
-}
-
-message TelemetryDefinitionRequest {
-    string name = 1;
-    TelemetryDataType data_type = 2;
-}
-
-message TelemetryDefinitionResponse {
-    UUID uuid = 1;
-}
-
-message TelemetryItem {
-    UUID uuid = 1;
-    TelemetryValue value = 2;
-    Timestamp timestamp = 3;
-}
-
-message TelemetryInsertResponse {
-}
-
-service TelemetryService {
-    rpc NewTelemetry (TelemetryDefinitionRequest) returns (TelemetryDefinitionResponse);
-    rpc InsertTelemetry (stream TelemetryItem) returns (stream TelemetryInsertResponse);
-}
-
-message CommandParameterDefinition {
-    string name = 1;
-    TelemetryDataType data_type = 2;
-}
-
-message CommandDefinitionRequest {
-    string name = 1;
-    repeated CommandParameterDefinition parameters = 2;
-}
-
-message Command {
-    UUID uuid = 1;
-    Timestamp timestamp = 2;
-    map<string, TelemetryValue> parameters = 3;
-}
-
-message CommandResponse {
-    UUID uuid = 1;
-    bool success = 2;
-    string response = 3;
-}
-
-message ClientSideCommand {
-    oneof inner {
-        CommandDefinitionRequest request = 1;
-        CommandResponse response = 2;
-    }
-}
-
-service CommandService {
-    rpc NewCommand (stream ClientSideCommand) returns (stream Command);
-}

server/src/command/command_handle.rs

@@ -0,0 +1,20 @@
+use uuid::Uuid;
+
+pub struct CommandHandle {
+    name: String,
+    uuid: Uuid,
+}
+
+impl CommandHandle {
+    pub fn new(name: String, uuid: Uuid) -> Self {
+        Self { name, uuid }
+    }
+
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    pub fn uuid(&self) -> &Uuid {
+        &self.uuid
+    }
+}

server/src/command/definition.rs

@@ -1,22 +1,5 @@
 use crate::command::service::RegisteredCommand;
-use crate::core::TelemetryDataType;
-use crate::telemetry::data_type::tlm_data_type_deserializer;
-use crate::telemetry::data_type::tlm_data_type_serializer;
-use serde::{Deserialize, Serialize};
-
-#[derive(Clone, Debug, Serialize, Deserialize)]
-pub struct CommandParameterDefinition {
-    pub name: String,
-    #[serde(serialize_with = "tlm_data_type_serializer")]
-    #[serde(deserialize_with = "tlm_data_type_deserializer")]
-    pub data_type: TelemetryDataType,
-}
-
-#[derive(Clone, Debug, Serialize, Deserialize)]
-pub struct CommandDefinition {
-    pub name: String,
-    pub parameters: Vec<CommandParameterDefinition>,
-}
+use api::messages::command::{CommandDefinition, CommandParameterDefinition};
 
 impl From<RegisteredCommand> for CommandDefinition {
     fn from(value: RegisteredCommand) -> Self {
@@ -27,7 +10,7 @@ impl From<RegisteredCommand> for CommandDefinition {
                 .parameters
                 .into_iter()
                 .map(|param| CommandParameterDefinition {
-                    data_type: param.data_type(),
+                    data_type: param.data_type,
                     name: param.name,
                 })
                 .collect(),

server/src/command/error.rs

@@ -1,6 +1,6 @@
-use crate::core::TelemetryDataType;
 use actix_web::http::StatusCode;
 use actix_web::ResponseError;
+use api::data_type::DataType;
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -14,7 +14,7 @@ pub enum Error {
     #[error("Incorrect Parameter Type for {name}. {expected_type:?} expected.")]
     WrongParameterType {
         name: String,
-        expected_type: TelemetryDataType,
+        expected_type: DataType,
     },
     #[error("No Command Receiver")]
     NoCommandReceiver,

server/src/command/mod.rs

@@ -1,3 +1,4 @@
+pub mod command_handle;
 mod definition;
 pub mod error;
 pub mod service;

server/src/command/service.rs

@@ -1,36 +1,40 @@
-use crate::command::definition::CommandDefinition;
+use crate::command::command_handle::CommandHandle;
 use crate::command::error::Error as CmdError;
 use crate::command::error::Error::{
     CommandFailure, CommandNotFound, FailedToReceiveResponse, FailedToSend,
     IncorrectParameterCount, MisingParameter, NoCommandReceiver, WrongParameterType,
 };
-use crate::core::telemetry_value::Value;
-use crate::core::{
-    Command, CommandDefinitionRequest, CommandResponse, TelemetryDataType, TelemetryValue,
-    Timestamp, Uuid,
-};
-use chrono::{DateTime, Utc};
+use anyhow::bail;
+use api::data_type::DataType;
+use api::data_value::DataValue;
+use api::messages::command::{Command, CommandDefinition, CommandHeader, CommandResponse};
+use api::messages::ResponseMessage;
+use chrono::Utc;
 use log::error;
 use papaya::HashMap;
 use std::collections::HashMap as StdHashMap;
-use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio::sync::{mpsc, RwLock};
+use uuid::Uuid;
 
 #[derive(Clone)]
 pub(super) struct RegisteredCommand {
     pub(super) name: String,
-    pub(super) definition: CommandDefinitionRequest,
-    tx: mpsc::Sender<Option<(Command, oneshot::Sender<CommandResponse>)>>,
+    pub(super) definition: CommandDefinition,
+    response_uuid: Uuid,
+    tx: mpsc::Sender<ResponseMessage>,
 }
 
 pub struct CommandManagementService {
     registered_commands: HashMap<String, RegisteredCommand>,
+    outstanding_responses: RwLock<StdHashMap<Uuid, oneshot::Sender<CommandResponse>>>,
 }
 
 impl CommandManagementService {
     pub fn new() -> Self {
         Self {
             registered_commands: HashMap::new(),
+            outstanding_responses: RwLock::new(StdHashMap::new()),
         }
     }
 
@@ -52,26 +56,26 @@ impl CommandManagementService {
             .map(|registration| registration.clone().into())
     }
 
-    pub async fn register_command(
+    pub fn register_command(
         &self,
-        command: CommandDefinitionRequest,
-    ) -> anyhow::Result<mpsc::Receiver<Option<(Command, oneshot::Sender<CommandResponse>)>>> {
-        let (tx, rx) = mpsc::channel(1);
-
-        let registered_commands = self.registered_commands.pin_owned();
-        if let Some(previous) = registered_commands.insert(
-            command.name.clone(),
+        uuid: Uuid,
+        command: CommandDefinition,
+        tx: mpsc::Sender<ResponseMessage>,
+    ) -> anyhow::Result<CommandHandle> {
+        let registered_commands = self.registered_commands.pin();
+        // We don't care about the previously registered command
+        let name = command.name.clone();
+        let _ = registered_commands.insert(
+            name.clone(),
             RegisteredCommand {
-                name: command.name.clone(),
+                response_uuid: uuid,
+                name: name.clone(),
                 definition: command,
                 tx,
             },
-        ) {
-            // If the receiver was already closed, we don't care (ignore error)
-            let _ = previous.tx.send(None).await;
-        }
+        );
 
-        Ok(rx)
+        Ok(CommandHandle::new(name, uuid))
     }
 
     pub async fn send_command(
@@ -80,8 +84,6 @@ impl CommandManagementService {
         parameters: serde_json::Map<String, serde_json::Value>,
     ) -> Result<String, CmdError> {
         let timestamp = Utc::now();
-        let offset_from_unix_epoch =
-            timestamp - DateTime::from_timestamp(0, 0).expect("Could not get Unix epoch");
 
         let name = name.into();
         let registered_commands = self.registered_commands.pin();
@@ -100,27 +102,21 @@ impl CommandManagementService {
             let Some(param_value) = parameters.get(&parameter.name) else {
                 return Err(MisingParameter(parameter.name.clone()));
             };
-            let Some(param_value) = (match parameter.data_type() {
-                TelemetryDataType::Float32 => {
-                    param_value.as_f64().map(|v| Value::Float32(v as f32))
-                }
-                TelemetryDataType::Float64 => param_value.as_f64().map(Value::Float64),
-                TelemetryDataType::Boolean => param_value.as_bool().map(Value::Boolean),
+            let Some(param_value) = (match parameter.data_type {
+                DataType::Float32 => param_value.as_f64().map(|v| DataValue::Float32(v as f32)),
+                DataType::Float64 => param_value.as_f64().map(DataValue::Float64),
+                DataType::Boolean => param_value.as_bool().map(DataValue::Boolean),
             }) else {
                 return Err(WrongParameterType {
                     name: parameter.name.clone(),
-                    expected_type: parameter.data_type(),
+                    expected_type: parameter.data_type,
                 });
             };
-            result_parameters.insert(
-                parameter.name.clone(),
-                TelemetryValue {
-                    value: Some(param_value),
-                },
-            );
+            result_parameters.insert(parameter.name.clone(), param_value);
         }
 
         // Clone & Drop lets us use a standard pin instead of an owned pin
+        let response_uuid = registration.response_uuid;
         let tx = registration.tx.clone();
         drop(registered_commands);
 
@@ -128,23 +124,27 @@ impl CommandManagementService {
             return Err(NoCommandReceiver);
         }
 
-        let uuid = Uuid::random();
+        let uuid = Uuid::new_v4();
         let (response_tx, response_rx) = oneshot::channel();
+
+        {
+            let mut outstanding_responses = self.outstanding_responses.write().await;
+            outstanding_responses.insert(uuid, response_tx);
+        }
+
         if let Err(e) = tx
-            .send(Some((
-                Command {
-                    uuid: Some(uuid),
-                    timestamp: Some(Timestamp {
-                        secs: offset_from_unix_epoch.num_seconds(),
-                        nanos: offset_from_unix_epoch.subsec_nanos(),
-                    }),
+            .send(ResponseMessage {
+                uuid,
+                response: Some(response_uuid),
+                payload: Command {
+                    header: CommandHeader { timestamp },
                     parameters: result_parameters,
-                },
-                response_tx,
-            )))
+                }
+                .into(),
+            })
             .await
         {
-            error!("Failed to Send Command: {e}");
+            error!("Failed to Send Command {e}");
             return Err(FailedToSend);
         }
 
@@ -162,4 +162,33 @@ impl CommandManagementService {
                 }
             })
     }
+
+    pub async fn handle_command_response(
+        &self,
+        uuid: Uuid,
+        response: CommandResponse,
+    ) -> anyhow::Result<()> {
+        let responder = {
+            let mut outstanding_responses = self.outstanding_responses.write().await;
+            outstanding_responses.remove(&uuid)
+        };
+        match responder {
+            None => bail!("Unexpected Command Response for Command {uuid}"),
+            Some(response_tx) => {
+                if let Err(e) = response_tx.send(response) {
+                    bail!("Failed to send Command Response {e:?}");
+                }
+            }
+        };
+
+        Ok(())
+    }
+
+    pub fn unregister(&self, command_handle: CommandHandle) {
+        let registered_commands = self.registered_commands.pin();
+        // We don't care if this succeeded
+        let _ = registered_commands.remove_if(command_handle.name(), |_, registration| {
+            registration.response_uuid == *command_handle.uuid()
+        });
+    }
 }

server/src/grpc/cmd.rs

@@ -135,11 +135,14 @@ impl CommandService for CoreCommandService {
                 }
             }
             for (key, sender) in in_progress.drain() {
-                if sender.send(CommandResponse {
-                    uuid: Some(Uuid::from(key)),
-                    success: false,
-                    response: "Command Handler Shut Down".to_string(),
-                }).is_err() {
+                if sender
+                    .send(CommandResponse {
+                        uuid: Some(Uuid::from(key)),
+                        success: false,
+                        response: "Command Handler Shut Down".to_string(),
+                    })
+                    .is_err()
+                {
                     error!("Failed to send command response on shutdown");
                 }
             }

server/src/grpc/mod.rs

@@ -1,44 +0,0 @@
-mod cmd;
-mod tlm;
-
-use crate::command::service::CommandManagementService;
-use crate::core::command_service_server::CommandServiceServer;
-use crate::core::telemetry_service_server::TelemetryServiceServer;
-use crate::grpc::cmd::CoreCommandService;
-use crate::grpc::tlm::CoreTelemetryService;
-use crate::telemetry::management_service::TelemetryManagementService;
-use log::{error, info};
-use std::sync::Arc;
-use tokio::task::JoinHandle;
-use tokio_util::sync::CancellationToken;
-use tonic::transport::Server;
-
-pub fn setup(
-    token: CancellationToken,
-    telemetry_management_service: Arc<TelemetryManagementService>,
-    command_service: Arc<CommandManagementService>,
-) -> anyhow::Result<JoinHandle<()>> {
-    let addr = "[::1]:50051".parse()?;
-    Ok(tokio::spawn(async move {
-        let tlm_service = CoreTelemetryService {
-            tlm_management: telemetry_management_service,
-            cancellation_token: token.clone(),
-        };
-
-        let cmd_service = CoreCommandService {
-            command_service,
-            cancellation_token: token.clone(),
-        };
-
-        info!("Starting gRPC Server");
-        let result = Server::builder()
-            .add_service(TelemetryServiceServer::new(tlm_service))
-            .add_service(CommandServiceServer::new(cmd_service))
-            .serve_with_shutdown(addr, token.cancelled_owned())
-            .await;
-
-        if let Err(err) = result {
-            error!("gRPC Server Encountered An Error: {err}");
-        }
-    }))
-}

server/src/grpc/tlm.rs

@@ -1,141 +0,0 @@
-use crate::core::telemetry_service_server::TelemetryService;
-use crate::core::telemetry_value::Value;
-use crate::core::{
-    TelemetryDataType, TelemetryDefinitionRequest, TelemetryDefinitionResponse,
-    TelemetryInsertResponse, TelemetryItem, TelemetryValue, Uuid,
-};
-use crate::telemetry::data_item::TelemetryDataItem;
-use crate::telemetry::data_value::TelemetryDataValue;
-use crate::telemetry::history::TelemetryHistory;
-use crate::telemetry::management_service::TelemetryManagementService;
-use chrono::{DateTime, SecondsFormat};
-use log::trace;
-use std::pin::Pin;
-use std::sync::Arc;
-use tokio::select;
-use tokio::sync::mpsc;
-use tokio_util::sync::CancellationToken;
-use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
-use tonic::codegen::tokio_stream::{Stream, StreamExt};
-use tonic::{Request, Response, Status, Streaming};
-
-pub struct CoreTelemetryService {
-    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!("CoreTelemetryService::new_telemetry");
-        self.tlm_management
-            .register(request.into_inner())
-            .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!("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);
-
-        tokio::spawn(async move {
-            loop {
-                select! {
-                    _ = cancel_token.cancelled() => {
-                        break;
-                    },
-                    Some(message) = in_stream.next() => {
-                        match message {
-                            Ok(tlm_item) => {
-                                tx
-                                    .send(Self::handle_new_tlm_item(&tlm_management, &tlm_item))
-                                    .await
-                                    .expect("working rx");
-                            }
-                            Err(err) => {
-                                let _ = tx.send(Err(err)).await;
-                            }
-                        }
-                    },
-                    else => break,
-                }
-            }
-        });
-
-        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
-    }
-}
-
-impl CoreTelemetryService {
-    #[allow(clippy::result_large_err)]
-    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 tlm_management_pin = tlm_management.pin();
-        let Some(tlm_data) = tlm_management_pin.get_by_uuid(&uuid.value) 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 Some(timestamp) = tlm_item.timestamp else {
-            return Err(Status::failed_precondition("Timestamp Missing"));
-        };
-
-        let expected_type = match value {
-            Value::Float32(_) => TelemetryDataType::Float32,
-            Value::Float64(_) => TelemetryDataType::Float64,
-            Value::Boolean(_) => TelemetryDataType::Boolean,
-        };
-        if expected_type != tlm_data.data.definition.data_type {
-            return Err(Status::failed_precondition("Data Type Mismatch"));
-        };
-
-        let Some(timestamp) = DateTime::from_timestamp(timestamp.secs, timestamp.nanos as u32)
-        else {
-            return Err(Status::invalid_argument("Failed to construct UTC DateTime"));
-        };
-
-        let value = match value {
-            Value::Float32(x) => TelemetryDataValue::Float32(x),
-            Value::Float64(x) => TelemetryDataValue::Float64(x),
-            Value::Boolean(x) => TelemetryDataValue::Boolean(x),
-        };
-        let _ = tlm_data.data.data.send_replace(Some(TelemetryDataItem {
-            value: value.clone(),
-            timestamp: timestamp.to_rfc3339_opts(SecondsFormat::Millis, true),
-        }));
-        TelemetryHistory::insert_sync(
-            tlm_data.clone(),
-            tlm_management.history_service(),
-            value,
-            timestamp,
-        );
-
-        Ok(TelemetryInsertResponse {})
-    }
-}

server/src/http/api/panels.rs

@@ -4,6 +4,7 @@ use crate::panels::PanelService;
 use actix_web::{delete, get, post, put, web, Responder};
 use serde::Deserialize;
 use std::sync::Arc;
+use uuid::Uuid;
 
 #[derive(Deserialize)]
 struct CreateParam {
@@ -13,7 +14,7 @@ struct CreateParam {
 
 #[derive(Deserialize)]
 struct IdParam {
-    id: String,
+    id: Uuid,
 }
 
 #[post("/panel")]
@@ -22,7 +23,7 @@ pub(super) async fn new(
     data: web::Json<CreateParam>,
 ) -> Result<impl Responder, HttpServerResultError> {
     let uuid = panels.create(&data.name, &data.data).await?;
-    Ok(web::Json(uuid.value))
+    Ok(web::Json(uuid))
 }
 
 #[get("/panel")]
@@ -38,12 +39,10 @@ pub(super) async fn get_one(
     panels: web::Data<Arc<PanelService>>,
     path: web::Path<IdParam>,
 ) -> Result<impl Responder, HttpServerResultError> {
-    let result = panels.read(path.id.clone().into()).await?;
+    let result = panels.read(path.id).await?;
     match result {
         Some(result) => Ok(web::Json(result)),
-        None => Err(HttpServerResultError::PanelUuidNotFound {
-            uuid: path.id.clone(),
-        }),
+        None => Err(HttpServerResultError::PanelUuidNotFound { uuid: path.id }),
     }
 }
 
@@ -53,7 +52,7 @@ pub(super) async fn set(
     path: web::Path<IdParam>,
     data: web::Json<PanelUpdate>,
 ) -> Result<impl Responder, HttpServerResultError> {
-    panels.update(path.id.clone().into(), data.0).await?;
+    panels.update(path.id, data.0).await?;
     Ok(web::Json(()))
 }
 
@@ -62,6 +61,6 @@ pub(super) async fn delete(
     panels: web::Data<Arc<PanelService>>,
     path: web::Path<IdParam>,
 ) -> Result<impl Responder, HttpServerResultError> {
-    panels.delete(path.id.clone().into()).await?;
+    panels.delete(path.id).await?;
     Ok(web::Json(()))
 }

server/src/http/api/tlm.rs

@@ -7,6 +7,7 @@ use serde::Deserialize;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::timeout;
+use uuid::Uuid;
 
 #[get("/tlm/info/{name:[\\w\\d/_-]+}")]
 pub(super) async fn get_tlm_definition(
@@ -36,13 +37,17 @@ struct HistoryQuery {
     resolution: i64,
 }
 
-#[get("/tlm/history/{uuid:[0-9a-f]+}")]
+#[get("/tlm/history/{uuid:[0-9a-f-]+}")]
 pub(super) async fn get_tlm_history(
     data_arc: web::Data<Arc<TelemetryManagementService>>,
     uuid: web::Path<String>,
     info: web::Query<HistoryQuery>,
 ) -> Result<impl Responder, HttpServerResultError> {
-    let uuid = uuid.to_string();
+    let Ok(uuid) = Uuid::parse_str(&uuid) else {
+        return Err(HttpServerResultError::InvalidUuid {
+            uuid: uuid.to_string(),
+        });
+    };
     trace!(
         "get_tlm_history {} from {} to {} resolution {}",
         uuid,

server/src/http/backend/connection.rs

@@ -0,0 +1,117 @@
+use crate::command::command_handle::CommandHandle;
+use crate::command::service::CommandManagementService;
+use crate::telemetry::management_service::TelemetryManagementService;
+use actix_ws::{AggregatedMessage, ProtocolError, Session};
+use anyhow::bail;
+use api::messages::payload::RequestMessagePayload;
+use api::messages::{RequestMessage, ResponseMessage};
+use std::sync::Arc;
+use tokio::sync::mpsc::{Receiver, Sender};
+use uuid::Uuid;
+
+pub(super) struct BackendConnection {
+    session: Session,
+    tlm_management: Arc<TelemetryManagementService>,
+    cmd_management: Arc<CommandManagementService>,
+    tx: Sender<ResponseMessage>,
+    commands: Vec<CommandHandle>,
+    pub rx: Receiver<ResponseMessage>,
+    pub should_close: bool,
+}
+
+impl BackendConnection {
+    pub fn new(
+        session: Session,
+        tlm_management: Arc<TelemetryManagementService>,
+        cmd_management: Arc<CommandManagementService>,
+    ) -> Self {
+        let (tx, rx) = tokio::sync::mpsc::channel::<ResponseMessage>(128);
+        Self {
+            session,
+            tlm_management,
+            cmd_management,
+            tx,
+            commands: vec![],
+            rx,
+            should_close: false,
+        }
+    }
+
+    async fn handle_request(&mut self, msg: RequestMessage) -> anyhow::Result<()> {
+        match msg.payload {
+            RequestMessagePayload::TelemetryDefinitionRequest(tlm_def) => {
+                self.tx
+                    .send(ResponseMessage {
+                        uuid: Uuid::new_v4(),
+                        response: Some(msg.uuid),
+                        payload: self.tlm_management.register(tlm_def)?.into(),
+                    })
+                    .await?;
+            }
+            RequestMessagePayload::TelemetryEntry(tlm_entry) => {
+                self.tlm_management.add_tlm_item(tlm_entry)?;
+            }
+            RequestMessagePayload::GenericCallbackError(_) => todo!(),
+            RequestMessagePayload::CommandDefinition(def) => {
+                let cmd = self
+                    .cmd_management
+                    .register_command(msg.uuid, def, self.tx.clone())?;
+                self.commands.push(cmd);
+            }
+            RequestMessagePayload::CommandResponse(response) => match msg.response {
+                None => bail!("Command Response Payload Must Respond to a Command"),
+                Some(uuid) => {
+                    self.cmd_management
+                        .handle_command_response(uuid, response)
+                        .await?;
+                }
+            },
+        }
+        Ok(())
+    }
+
+    pub async fn handle_request_message(
+        &mut self,
+        msg: Result<AggregatedMessage, ProtocolError>,
+    ) -> anyhow::Result<()> {
+        let msg = msg?;
+        match msg {
+            AggregatedMessage::Text(data) => {
+                self.handle_request(serde_json::from_str(&data)?).await?;
+            }
+            AggregatedMessage::Binary(_) => {
+                bail!("Binary Messages Unsupported");
+            }
+            AggregatedMessage::Ping(bytes) => {
+                self.session.pong(&bytes).await?;
+            }
+            AggregatedMessage::Pong(_) => {
+                // Intentionally Ignore
+            }
+            AggregatedMessage::Close(_) => {
+                self.should_close = true;
+            }
+        }
+        Ok(())
+    }
+
+    pub async fn handle_response(&mut self, msg: ResponseMessage) -> anyhow::Result<()> {
+        let msg_json = serde_json::to_string(&msg)?;
+        self.session.text(msg_json).await?;
+        Ok(())
+    }
+
+    pub async fn cleanup(mut self) {
+        self.rx.close();
+        // Clone here to prevent conflict with the Drop trait
+        let _ = self.session.clone().close(None).await;
+    }
+}
+
+impl Drop for BackendConnection {
+    fn drop(&mut self) {
+        for command in self.commands.drain(..) {
+            self.cmd_management.unregister(command);
+        }
+    }
+}

server/src/http/backend/mod.rs

@@ -0,0 +1,60 @@
+use futures_util::stream::StreamExt;
+mod connection;
+
+use crate::command::service::CommandManagementService;
+use crate::http::backend::connection::BackendConnection;
+use crate::telemetry::management_service::TelemetryManagementService;
+use actix_web::{rt, web, HttpRequest, HttpResponse};
+use log::{error, trace};
+use std::sync::Arc;
+use tokio::select;
+use tokio_util::sync::CancellationToken;
+
+async fn backend_connect(
+    req: HttpRequest,
+    stream: web::Payload,
+    cancel_token: web::Data<CancellationToken>,
+    telemetry_management_service: web::Data<Arc<TelemetryManagementService>>,
+    command_management_service: web::Data<Arc<CommandManagementService>>,
+) -> Result<HttpResponse, actix_web::Error> {
+    trace!("backend_connect");
+    let (res, session, stream) = actix_ws::handle(&req, stream)?;
+
+    let mut stream = stream
+        .aggregate_continuations()
+        // up to 1 MiB
+        .max_continuation_size(2_usize.pow(20));
+
+    let cancel_token = cancel_token.get_ref().clone();
+    let tlm_management = telemetry_management_service.get_ref().clone();
+    let cmd_management = command_management_service.get_ref().clone();
+
+    rt::spawn(async move {
+        let mut connection = BackendConnection::new(session, tlm_management, cmd_management);
+        while !connection.should_close {
+            let result = select! {
+                _ = cancel_token.cancelled() => {
+                    connection.should_close = true;
+                    Ok(())
+                },
+                Some(msg) = connection.rx.recv() => connection.handle_response(msg).await,
+                Some(msg) = stream.next() => connection.handle_request_message(msg).await,
+                else => {
+                    connection.should_close = true;
+                    Ok(())
+                },
+            };
+            if let Err(e) = result {
+                error!("backend socket error: {e}");
+                connection.should_close = true;
+            }
+        }
+        connection.cleanup().await;
+    });
+
+    Ok(res)
+}
+
+pub fn setup_backend(cfg: &mut web::ServiceConfig) {
+    cfg.route("", web::get().to(backend_connect));
+}

server/src/http/error.rs

@@ -3,13 +3,16 @@ use actix_web::http::header::ContentType;
 use actix_web::http::StatusCode;
 use actix_web::HttpResponse;
 use thiserror::Error;
+use uuid::Uuid;
 
 #[derive(Error, Debug)]
 pub enum HttpServerResultError {
     #[error("Telemetry Name Not Found: {tlm}")]
     TlmNameNotFound { tlm: String },
+    #[error("Invalid Uuid: {uuid}")]
+    InvalidUuid { uuid: String },
     #[error("Telemetry Uuid Not Found: {uuid}")]
-    TlmUuidNotFound { uuid: String },
+    TlmUuidNotFound { uuid: Uuid },
     #[error("DateTime Parsing Error: {date_time}")]
     InvalidDateTime { date_time: String },
     #[error("Timed out")]
@@ -17,7 +20,7 @@ pub enum HttpServerResultError {
     #[error("Internal Error")]
     InternalError(#[from] anyhow::Error),
     #[error("Panel Uuid Not Found: {uuid}")]
-    PanelUuidNotFound { uuid: String },
+    PanelUuidNotFound { uuid: Uuid },
     #[error(transparent)]
     Command(#[from] crate::command::error::Error),
 }
@@ -26,6 +29,7 @@ impl ResponseError for HttpServerResultError {
     fn status_code(&self) -> StatusCode {
         match self {
             HttpServerResultError::TlmNameNotFound { .. } => StatusCode::NOT_FOUND,
+            HttpServerResultError::InvalidUuid { .. } => StatusCode::BAD_REQUEST,
             HttpServerResultError::TlmUuidNotFound { .. } => StatusCode::NOT_FOUND,
             HttpServerResultError::InvalidDateTime { .. } => StatusCode::BAD_REQUEST,
             HttpServerResultError::Timeout => StatusCode::GATEWAY_TIMEOUT,

server/src/http/mod.rs

@@ -1,9 +1,11 @@
 mod api;
+mod backend;
 mod error;
 mod websocket;
 
 use crate::command::service::CommandManagementService;
 use crate::http::api::setup_api;
+use crate::http::backend::setup_backend;
 use crate::http::websocket::setup_websocket;
 use crate::panels::PanelService;
 use crate::telemetry::management_service::TelemetryManagementService;
@@ -31,6 +33,7 @@ pub async fn setup(
             .app_data(cancel_token.clone())
             .app_data(panel_service.clone())
             .app_data(command_service.clone())
+            .service(web::scope("/backend").configure(setup_backend))
             .service(web::scope("/ws").configure(setup_websocket))
             .service(web::scope("/api").configure(setup_api))
             .wrap(Logger::default())

server/src/http/websocket/mod.rs

@@ -6,6 +6,7 @@ use crate::telemetry::management_service::TelemetryManagementService;
 use actix_web::{rt, web, HttpRequest, HttpResponse};
 use actix_ws::{AggregatedMessage, ProtocolError, Session};
 use anyhow::anyhow;
+use futures_util::StreamExt;
 use log::{error, trace};
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -14,7 +15,7 @@ use tokio::select;
 use tokio::sync::mpsc::Sender;
 use tokio::time::{sleep_until, Instant};
 use tokio_util::sync::CancellationToken;
-use tonic::codegen::tokio_stream::StreamExt;
+use uuid::Uuid;
 
 pub mod request;
 pub mod response;
@@ -23,11 +24,11 @@ fn handle_register_tlm_listener(
     data: &Arc<TelemetryManagementService>,
     request: RegisterTlmListenerRequest,
     tx: &Sender<WebsocketResponse>,
-    tlm_listeners: &mut HashMap<String, CancellationToken>,
+    tlm_listeners: &mut HashMap<Uuid, CancellationToken>,
 ) {
     if let Some(tlm_data) = data.get_by_uuid(&request.uuid) {
         let token = CancellationToken::new();
-        if let Some(token) = tlm_listeners.insert(tlm_data.definition.uuid.clone(), token.clone()) {
+        if let Some(token) = tlm_listeners.insert(tlm_data.definition.uuid, token.clone()) {
             token.cancel();
         }
         let minimum_separation = Duration::from_millis(request.minimum_separation_ms as u64);
@@ -46,7 +47,7 @@ fn handle_register_tlm_listener(
                             ref_val.clone()
                         };
                         let _ = tx.send(TlmValueResponse {
-                            uuid: request.uuid.clone(),
+                            uuid: request.uuid,
                             value,
                         }.into()).await;
                         now
@@ -65,7 +66,7 @@ fn handle_register_tlm_listener(
 
 fn handle_unregister_tlm_listener(
     request: UnregisterTlmListenerRequest,
-    tlm_listeners: &mut HashMap<String, CancellationToken>,
+    tlm_listeners: &mut HashMap<Uuid, CancellationToken>,
 ) {
     if let Some(token) = tlm_listeners.remove(&request.uuid) {
         token.cancel();
@@ -76,7 +77,7 @@ async fn handle_websocket_message(
     data: &Arc<TelemetryManagementService>,
     request: WebsocketRequest,
     tx: &Sender<WebsocketResponse>,
-    tlm_listeners: &mut HashMap<String, CancellationToken>,
+    tlm_listeners: &mut HashMap<Uuid, CancellationToken>,
 ) {
     match request {
         WebsocketRequest::RegisterTlmListener(request) => {
@@ -110,7 +111,7 @@ async fn handle_websocket_incoming(
     data: &Arc<TelemetryManagementService>,
     session: &mut Session,
     tx: &Sender<WebsocketResponse>,
-    tlm_listeners: &mut HashMap<String, CancellationToken>,
+    tlm_listeners: &mut HashMap<Uuid, CancellationToken>,
 ) -> anyhow::Result<bool> {
     match msg {
         Ok(AggregatedMessage::Close(_)) => Ok(false),
@@ -130,7 +131,7 @@ async fn handle_websocket_incoming(
     }
 }
 
-pub async fn websocket_connect(
+async fn websocket_connect(
     req: HttpRequest,
     stream: web::Payload,
     data: web::Data<Arc<TelemetryManagementService>>,

server/src/http/websocket/request.rs

@@ -1,15 +1,16 @@
 use derive_more::From;
 use serde::{Deserialize, Serialize};
+use uuid::Uuid;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RegisterTlmListenerRequest {
-    pub uuid: String,
+    pub uuid: Uuid,
     pub minimum_separation_ms: u32,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct UnregisterTlmListenerRequest {
-    pub uuid: String,
+    pub uuid: Uuid,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, From)]

server/src/http/websocket/response.rs

@@ -1,10 +1,11 @@
 use crate::telemetry::data_item::TelemetryDataItem;
 use derive_more::From;
 use serde::{Deserialize, Serialize};
+use uuid::Uuid;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TlmValueResponse {
-    pub uuid: String,
+    pub uuid: Uuid,
     pub value: Option<TelemetryDataItem>,
 }
 

server/src/lib.rs

@@ -1,14 +1,8 @@
 mod command;
-mod grpc;
 mod http;
 mod panels;
 mod serialization;
 mod telemetry;
-mod uuid;
-
-pub mod core {
-    tonic::include_proto!("core");
-}
 
 use crate::command::service::CommandManagementService;
 use crate::panels::PanelService;
@@ -53,14 +47,11 @@ pub async fn setup() -> anyhow::Result<()> {
 
     let cmd = Arc::new(CommandManagementService::new());
 
-    let grpc_server = grpc::setup(cancellation_token.clone(), tlm.clone(), cmd.clone())?;
-
     let panel_service = PanelService::new(sqlite.clone());
 
     let result = http::setup(cancellation_token.clone(), tlm.clone(), panel_service, cmd).await;
     cancellation_token.cancel();
     result?; // result is dropped
-    grpc_server.await?; //grpc server is dropped
     drop(cancellation_token); // All cancellation tokens are now dropped
 
     sqlite.close().await;

server/src/panels/mod.rs

@@ -1,9 +1,9 @@
 pub mod panel;
 
-use crate::core::Uuid;
 use crate::panels::panel::{PanelRequired, PanelUpdate};
 use panel::Panel;
 use sqlx::SqlitePool;
+use uuid::Uuid;
 
 pub struct PanelService {
     pool: SqlitePool,
@@ -15,7 +15,8 @@ impl PanelService {
     }
 
     pub async fn create(&self, name: &str, data: &str) -> anyhow::Result<Uuid> {
-        let id = Uuid::random();
+        let id = Uuid::new_v4();
+        let id_string = id.to_string();
 
         let mut transaction = self.pool.begin().await?;
 
@@ -24,7 +25,7 @@ impl PanelService {
                 INSERT INTO PANELS (id, name, data, deleted)
                 VALUES ($1, $2, $3, FALSE);
             "#,
-            id.value,
+            id_string,
             name,
             data
         )
@@ -65,7 +66,7 @@ impl PanelService {
                     WHERE id = $1 AND deleted = FALSE
                 "#,
         )
-        .bind(id.value)
+        .bind(id.to_string())
         .fetch_optional(&mut *transaction)
         .await?;
 
@@ -75,6 +76,7 @@ impl PanelService {
     }
 
     pub async fn update(&self, id: Uuid, data: PanelUpdate) -> anyhow::Result<()> {
+        let id = id.to_string();
         let mut transaction = self.pool.begin().await?;
 
         if let Some(name) = data.name {
@@ -84,7 +86,7 @@ impl PanelService {
                         SET name = $2
                         WHERE id = $1;
                     "#,
-                id.value,
+                id,
                 name
             )
             .execute(&mut *transaction)
@@ -97,7 +99,7 @@ impl PanelService {
                         SET data = $2
                         WHERE id = $1;
                     "#,
-                id.value,
+                id,
                 data
             )
             .execute(&mut *transaction)
@@ -110,6 +112,7 @@ impl PanelService {
     }
 
     pub async fn delete(&self, id: Uuid) -> anyhow::Result<()> {
+        let id = id.to_string();
         let mut transaction = self.pool.begin().await?;
 
         let _ = sqlx::query!(
@@ -118,7 +121,7 @@ impl PanelService {
                     SET deleted = TRUE
                     WHERE id = $1;
                 "#,
-            id.value,
+            id,
         )
         .execute(&mut *transaction)
         .await?;

server/src/telemetry/data_item.rs

@@ -1,8 +1,8 @@
-use crate::telemetry::data_value::TelemetryDataValue;
+use api::data_value::DataValue;
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TelemetryDataItem {
-    pub value: TelemetryDataValue,
+    pub value: DataValue,
     pub timestamp: String,
 }

server/src/telemetry/data_type.rs

@@ -1,38 +0,0 @@
-use crate::core::TelemetryDataType;
-use serde::de::Visitor;
-use serde::{Deserializer, Serializer};
-use std::fmt::Formatter;
-
-pub fn tlm_data_type_serializer<S>(
-    tlm_data_type: &TelemetryDataType,
-    serializer: S,
-) -> Result<S::Ok, S::Error>
-where
-    S: Serializer,
-{
-    serializer.serialize_str(tlm_data_type.as_str_name())
-}
-
-struct TlmDataTypeVisitor;
-
-impl Visitor<'_> for TlmDataTypeVisitor {
-    type Value = TelemetryDataType;
-
-    fn expecting(&self, formatter: &mut Formatter) -> std::fmt::Result {
-        formatter.write_str("A &str")
-    }
-
-    fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
-    where
-        E: serde::de::Error,
-    {
-        TelemetryDataType::from_str_name(v).ok_or(E::custom("Invalid TelemetryDataType"))
-    }
-}
-
-pub fn tlm_data_type_deserializer<'de, D>(deserializer: D) -> Result<TelemetryDataType, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    deserializer.deserialize_str(TlmDataTypeVisitor)
-}

server/src/telemetry/data_value.rs

@@ -1,8 +0,0 @@
-use serde::{Deserialize, Serialize};
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub enum TelemetryDataValue {
-    Float32(f32),
-    Float64(f64),
-    Boolean(bool),
-}

server/src/telemetry/definition.rs

@@ -1,13 +1,10 @@
-use crate::core::TelemetryDataType;
-use crate::telemetry::data_type::tlm_data_type_deserializer;
-use crate::telemetry::data_type::tlm_data_type_serializer;
+use api::data_type::DataType;
 use serde::{Deserialize, Serialize};
+use uuid::Uuid;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TelemetryDefinition {
-    pub uuid: String,
+    pub uuid: Uuid,
     pub name: String,
-    #[serde(serialize_with = "tlm_data_type_serializer")]
-    #[serde(deserialize_with = "tlm_data_type_deserializer")]
-    pub data_type: TelemetryDataType,
+    pub data_type: DataType,
 }

server/src/telemetry/history.rs

@@ -1,10 +1,10 @@
-use crate::core::TelemetryDataType;
 use crate::serialization::file_ext::{ReadExt, WriteExt};
 use crate::telemetry::data::TelemetryData;
 use crate::telemetry::data_item::TelemetryDataItem;
-use crate::telemetry::data_value::TelemetryDataValue;
 use crate::telemetry::definition::TelemetryDefinition;
 use anyhow::{anyhow, ensure, Context};
+use api::data_type::DataType;
+use api::data_value::DataValue;
 use chrono::{DateTime, DurationRound, SecondsFormat, TimeDelta, Utc};
 use log::{error, info};
 use std::cmp::min;
@@ -44,7 +44,7 @@ fn update_next_from(
 }
 
 struct SegmentData {
-    values: Vec<TelemetryDataValue>,
+    values: Vec<DataValue>,
     timestamps: Vec<DateTime<Utc>>,
 }
 
@@ -66,7 +66,7 @@ impl HistorySegmentRam {
         }
     }
 
-    fn insert(&self, value: TelemetryDataValue, timestamp: DateTime<Utc>) {
+    fn insert(&self, value: DataValue, timestamp: DateTime<Utc>) {
         if timestamp < self.start || timestamp >= self.end {
             return;
         }
@@ -121,7 +121,7 @@ impl HistorySegmentRam {
                             next_from,
                         );
                         result.push(TelemetryDataItem {
-                            value: data.values[i].clone(),
+                            value: data.values[i],
                             timestamp: t.to_rfc3339_opts(SecondsFormat::Millis, true),
                         });
                     }
@@ -196,9 +196,9 @@ impl HistorySegmentFile {
         // Write all the values
         for value in &data.values {
             match value {
-                TelemetryDataValue::Float32(value) => file.write_data::<f32>(*value)?,
-                TelemetryDataValue::Float64(value) => file.write_data::<f64>(*value)?,
-                TelemetryDataValue::Boolean(value) => file.write_data::<bool>(*value)?,
+                DataValue::Float32(value) => file.write_data::<f32>(*value)?,
+                DataValue::Float64(value) => file.write_data::<f64>(*value)?,
+                DataValue::Boolean(value) => file.write_data::<bool>(*value)?,
             }
         }
 
@@ -215,10 +215,7 @@ impl HistorySegmentFile {
         })
     }
 
-    fn load_to_ram(
-        mut self,
-        telemetry_data_type: TelemetryDataType,
-    ) -> anyhow::Result<HistorySegmentRam> {
+    fn load_to_ram(mut self, telemetry_data_type: DataType) -> anyhow::Result<HistorySegmentRam> {
         let mut segment_data = SegmentData {
             values: Vec::with_capacity(self.length as usize),
             timestamps: Vec::with_capacity(self.length as usize),
@@ -281,7 +278,7 @@ impl HistorySegmentFile {
         from: DateTime<Utc>,
         to: DateTime<Utc>,
         maximum_resolution: TimeDelta,
-        telemetry_data_type: TelemetryDataType,
+        telemetry_data_type: DataType,
     ) -> anyhow::Result<(DateTime<Utc>, Vec<TelemetryDataItem>)> {
         self.file_position = 0;
         self.file.seek(SeekFrom::Start(0))?;
@@ -334,22 +331,19 @@ impl HistorySegmentFile {
         self.read_date_time()
     }
 
-    fn read_telemetry_item(
-        &mut self,
-        telemetry_data_type: TelemetryDataType,
-    ) -> anyhow::Result<TelemetryDataValue> {
+    fn read_telemetry_item(&mut self, telemetry_data_type: DataType) -> anyhow::Result<DataValue> {
         match telemetry_data_type {
-            TelemetryDataType::Float32 => {
+            DataType::Float32 => {
                 self.file_position += 4;
-                Ok(TelemetryDataValue::Float32(self.file.read_data::<f32>()?))
+                Ok(DataValue::Float32(self.file.read_data::<f32>()?))
             }
-            TelemetryDataType::Float64 => {
+            DataType::Float64 => {
                 self.file_position += 8;
-                Ok(TelemetryDataValue::Float64(self.file.read_data::<f64>()?))
+                Ok(DataValue::Float64(self.file.read_data::<f64>()?))
             }
-            TelemetryDataType::Boolean => {
+            DataType::Boolean => {
                 self.file_position += 1;
-                Ok(TelemetryDataValue::Boolean(self.file.read_data::<bool>()?))
+                Ok(DataValue::Boolean(self.file.read_data::<bool>()?))
             }
         }
     }
@@ -357,12 +351,12 @@ impl HistorySegmentFile {
     fn get_telemetry_item(
         &mut self,
         index: u64,
-        telemetry_data_type: TelemetryDataType,
-    ) -> anyhow::Result<TelemetryDataValue> {
+        telemetry_data_type: DataType,
+    ) -> anyhow::Result<DataValue> {
         let item_length = match telemetry_data_type {
-            TelemetryDataType::Float32 => 4,
-            TelemetryDataType::Float64 => 8,
-            TelemetryDataType::Boolean => 1,
+            DataType::Float32 => 4,
+            DataType::Float64 => 8,
+            DataType::Boolean => 1,
         };
         let desired_position =
             Self::HEADER_LENGTH + self.length * Self::TIMESTAMP_LENGTH + index * item_length;
@@ -429,7 +423,7 @@ impl TelemetryHistory {
         history_segment_ram: HistorySegmentRam,
     ) -> JoinHandle<()> {
         let mut path = service.data_root_folder.clone();
-        path.push(&self.data.definition.uuid);
+        path.push(self.data.definition.uuid.as_hyphenated().to_string());
         spawn_blocking(move || {
             match HistorySegmentFile::save_to_disk(path, history_segment_ram) {
                 // Immediately drop the segment - now that we've saved it to disk we don't need to keep it in memory
@@ -450,7 +444,7 @@ impl TelemetryHistory {
         start: DateTime<Utc>,
     ) -> JoinHandle<anyhow::Result<HistorySegmentFile>> {
         let mut path = service.data_root_folder.clone();
-        path.push(&self.data.definition.uuid);
+        path.push(self.data.definition.uuid.as_hyphenated().to_string());
         spawn_blocking(move || HistorySegmentFile::open(path, start))
     }
 
@@ -458,7 +452,7 @@ impl TelemetryHistory {
         &self,
         start: DateTime<Utc>,
         service: &TelemetryHistoryService,
-        telemetry_data_type: TelemetryDataType,
+        telemetry_data_type: DataType,
     ) -> HistorySegmentRam {
         let ram = self
             .get_disk_segment(service, start)
@@ -480,7 +474,7 @@ impl TelemetryHistory {
     pub async fn insert(
         &self,
         service: &TelemetryHistoryService,
-        value: TelemetryDataValue,
+        value: DataValue,
         timestamp: DateTime<Utc>,
     ) {
         let segments = self.segments.read().await;
@@ -531,7 +525,7 @@ impl TelemetryHistory {
     pub fn insert_sync(
         history: Arc<Self>,
         service: Arc<TelemetryHistoryService>,
-        value: TelemetryDataValue,
+        value: DataValue,
         timestamp: DateTime<Utc>,
     ) {
         tokio::spawn(async move {
@@ -579,7 +573,7 @@ impl TelemetryHistory {
                 .unwrap();
 
             let mut path = telemetry_history_service.data_root_folder.clone();
-            path.push(&self.data.definition.uuid);
+            path.push(self.data.definition.uuid.as_hyphenated().to_string());
 
             let mut start = start;
             while start < end {

server/src/telemetry/management_service.rs

@@ -1,7 +1,15 @@
-use crate::core::{TelemetryDefinitionRequest, Uuid};
 use crate::telemetry::data::TelemetryData;
+use crate::telemetry::data_item::TelemetryDataItem;
 use crate::telemetry::definition::TelemetryDefinition;
 use crate::telemetry::history::{TelemetryHistory, TelemetryHistoryService};
+use anyhow::bail;
+use api::data_type::DataType;
+use api::data_value::DataValue;
+use api::messages::telemetry_definition::{
+    TelemetryDefinitionRequest, TelemetryDefinitionResponse,
+};
+use api::messages::telemetry_entry::TelemetryEntry;
+use chrono::SecondsFormat;
 use log::{error, info, warn};
 use papaya::{HashMap, HashMapRef, LocalGuard};
 use std::fs;
@@ -12,12 +20,13 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
+use uuid::Uuid;
 
 const RELEASED_ATTEMPTS: usize = 5;
 
 pub struct TelemetryManagementService {
-    uuid_index: HashMap<String, String>,
-    tlm_data: HashMap<String, Arc<TelemetryHistory>>,
+    uuid_index: HashMap<String, Uuid>,
+    tlm_data: HashMap<Uuid, Arc<TelemetryHistory>>,
     telemetry_history_service: Arc<TelemetryHistoryService>,
     metadata_file: Arc<Mutex<File>>,
 }
@@ -49,8 +58,8 @@ impl TelemetryManagementService {
                     // Skip invalid entries
                     match serde_json::from_str::<TelemetryDefinition>(line) {
                         Ok(tlm_def) => {
-                            let _ = uuid_index.insert(tlm_def.name.clone(), tlm_def.uuid.clone());
-                            let _ = tlm_data.insert(tlm_def.uuid.clone(), Arc::new(tlm_def.into()));
+                            let _ = uuid_index.insert(tlm_def.name.clone(), tlm_def.uuid);
+                            let _ = tlm_data.insert(tlm_def.uuid, Arc::new(tlm_def.into()));
                         }
                         Err(err) => {
                             error!("Failed to parse metadata entry {err}");
@@ -79,23 +88,20 @@ impl TelemetryManagementService {
     pub fn register(
         &self,
         telemetry_definition_request: TelemetryDefinitionRequest,
-    ) -> anyhow::Result<String> {
+    ) -> anyhow::Result<TelemetryDefinitionResponse> {
         let uuid_index = self.uuid_index.pin();
         let tlm_data = self.tlm_data.pin();
 
-        let uuid = uuid_index
-            .get_or_insert_with(telemetry_definition_request.name.clone(), || {
-                Uuid::random().value
-            })
-            .clone();
+        let uuid =
+            *uuid_index.get_or_insert_with(telemetry_definition_request.name.clone(), Uuid::new_v4);
 
         let inserted = tlm_data.try_insert(
-            uuid.clone(),
+            uuid,
             Arc::new(
                 TelemetryDefinition {
-                    uuid: uuid.clone(),
+                    uuid,
                     name: telemetry_definition_request.name.clone(),
-                    data_type: telemetry_definition_request.data_type(),
+                    data_type: telemetry_definition_request.data_type,
                 }
                 .into(),
             ),
@@ -129,7 +135,38 @@ impl TelemetryManagementService {
             });
         }
 
-        Ok(uuid)
+        Ok(TelemetryDefinitionResponse { uuid })
+    }
+
+    pub fn add_tlm_item(&self, tlm_item: TelemetryEntry) -> anyhow::Result<()> {
+        let tlm_management_pin = self.pin();
+        let Some(tlm_data) = tlm_management_pin.get_by_uuid(&tlm_item.uuid) else {
+            bail!("Telemetry Item Not Found");
+        };
+
+        let expected_type = match &tlm_item.value {
+            DataValue::Float32(_) => DataType::Float32,
+            DataValue::Float64(_) => DataType::Float64,
+            DataValue::Boolean(_) => DataType::Boolean,
+        };
+        if expected_type != tlm_data.data.definition.data_type {
+            bail!("Data Type Mismatch");
+        };
+
+        let _ = tlm_data.data.data.send_replace(Some(TelemetryDataItem {
+            value: tlm_item.value,
+            timestamp: tlm_item
+                .timestamp
+                .to_rfc3339_opts(SecondsFormat::Millis, true),
+        }));
+        TelemetryHistory::insert_sync(
+            tlm_data.clone(),
+            self.history_service(),
+            tlm_item.value,
+            tlm_item.timestamp,
+        );
+
+        Ok(())
     }
 
     pub fn get_by_name(&self, name: &String) -> Option<TelemetryData> {
@@ -138,7 +175,7 @@ impl TelemetryManagementService {
         self.get_by_uuid(uuid)
     }
 
-    pub fn get_by_uuid(&self, uuid: &String) -> Option<TelemetryData> {
+    pub fn get_by_uuid(&self, uuid: &Uuid) -> Option<TelemetryData> {
         let tlm_data = self.tlm_data.pin();
         tlm_data
             .get(uuid)
@@ -200,11 +237,11 @@ impl TelemetryManagementService {
 }
 
 pub struct TelemetryManagementServicePin<'a> {
-    tlm_data: HashMapRef<'a, String, Arc<TelemetryHistory>, RandomState, LocalGuard<'a>>,
+    tlm_data: HashMapRef<'a, Uuid, Arc<TelemetryHistory>, RandomState, LocalGuard<'a>>,
 }
 
 impl<'a> TelemetryManagementServicePin<'a> {
-    pub fn get_by_uuid(&'a self, uuid: &String) -> Option<&'a Arc<TelemetryHistory>> {
+    pub fn get_by_uuid(&'a self, uuid: &Uuid) -> Option<&'a Arc<TelemetryHistory>> {
         self.tlm_data.get(uuid)
     }
 }

server/src/telemetry/mod.rs

@@ -1,7 +1,5 @@
 pub mod data;
 pub mod data_item;
-pub mod data_type;
-pub mod data_value;
 pub mod definition;
 pub mod history;
 pub mod management_service;

server/src/uuid.rs

@@ -1,18 +0,0 @@
-use crate::core::Uuid;
-use rand::RngCore;
-
-impl Uuid {
-    pub fn random() -> Self {
-        let mut uuid = [0u8; 16];
-        rand::rng().fill_bytes(&mut uuid);
-        Self {
-            value: hex::encode(uuid),
-        }
-    }
-}
-
-impl From<String> for Uuid {
-    fn from(value: String) -> Self {
-        Self { value }
-    }
-}