Transfer Panel Details Without an Extra Layer of JSON Encoding (#14)

**Rationale:**

This made it harder to snoop on the traffic in the network monitor because it was encoded as a string.

**Changes:**

- Backend now accepts & provides the panel data as a JSON object rather than as a string
- Backend now supports compression
- Minor improvements to error handling
- Some panel structures were getting saved in the JSON when they weren't supposed to be (now this no longer happens)

Reviewed-on: #14
Co-authored-by: Sergey Savelyev <sergeysav.nn@gmail.com>
Co-committed-by: Sergey Savelyev <sergeysav.nn@gmail.com>

Cargo.lock

@@ -309,6 +309,7 @@ dependencies = [
  "api-proc-macro",
  "chrono",
  "derive_more",
+ "env_logger",
  "futures-util",
  "log",
  "serde",
@@ -472,6 +473,16 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75"
 
+[[package]]
+name = "colored"
+version = "2.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "117725a109d387c937a1533ce01b450cbde6b88abceea8473c4d7a85853cda3c"
+dependencies = [
+ "lazy_static",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "concurrent-queue"
 version = "2.5.0"
@@ -734,6 +745,7 @@ version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4316185f709b23713e41e3195f90edef7fb00c3ed4adc79769cf09cc762a3b29"
 dependencies = [
+ "colored",
  "log",
 ]
 
@@ -1941,6 +1953,7 @@ dependencies = [
  "fern",
  "futures-util",
  "log",
+ "num-traits",
  "papaya",
  "serde",
  "serde_json",
@@ -2019,11 +2032,9 @@ dependencies = [
  "chrono",
  "env_logger",
  "futures-util",
- "log",
  "num-traits",
  "tokio",
  "tokio-util",
- "uuid",
 ]
 
 [[package]]

Cargo.toml

@@ -2,10 +2,17 @@
 members = ["api", "api-core", "api-proc-macro", "server", "examples/simple_producer", "examples/simple_command"]
 resolver = "2"
 
+[workspace.package]
+version = "0.1.0"
+authors = ["Sergey <me@sergeysav.com>"]
+
 [workspace.dependencies]
 actix-web = "4.12.1"
 actix-ws = "0.3.0"
 anyhow = "1.0.100"
+api = { path = "./api" }
+api-core = { path = "./api-core" }
+api-proc-macro = { path = "./api-proc-macro" }
 chrono = { version = "0.4.42" }
 derive_more = { version = "2.1.1" }
 env_logger = "0.11.8"
@@ -22,6 +29,8 @@ sqlx = "0.8.6"
 syn = "2.0.112"
 thiserror = "2.0.17"
 tokio = { version = "1.48.0" }
+tokio-stream = "0.1.17"
+tokio-test = "0.4.4"
 tokio-tungstenite = { version = "0.28.0" }
 tokio-util = "0.7.17"
 trybuild = "1.0.114"

api-core/Cargo.toml

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

api-core/src/command.rs

@@ -11,25 +11,25 @@ pub struct CommandParameterDefinition {
     pub data_type: DataType,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct CommandDefinition {
     pub name: String,
     pub parameters: Vec<CommandParameterDefinition>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct CommandHeader {
     pub timestamp: DateTime<Utc>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct Command {
     #[serde(flatten)]
     pub header: CommandHeader,
     pub parameters: HashMap<String, DataValue>,
 }
 
-#[derive(Debug, Error)]
+#[derive(Debug, PartialEq, Eq, Error)]
 pub enum IntoCommandDefinitionError {
     #[error("Parameter Missing: {0}")]
     ParameterMissing(String),
@@ -45,3 +45,16 @@ pub trait IntoCommandDefinition: Sized {
 
     fn parse(command: Command) -> Result<Self, IntoCommandDefinitionError>;
 }
+
+impl IntoCommandDefinition for () {
+    fn create(name: String) -> CommandDefinition {
+        CommandDefinition {
+            name,
+            parameters: vec![],
+        }
+    }
+
+    fn parse(_: Command) -> Result<Self, IntoCommandDefinitionError> {
+        Ok(())
+    }
+}

api-core/src/data_type.rs

@@ -1,11 +1,38 @@
 use crate::data_value::DataValue;
+use derive_more::Display;
 use serde::{Deserialize, Serialize};
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Display)]
 pub enum DataType {
     Float32,
     Float64,
     Boolean,
+    Int8,
+    Int16,
+    Int32,
+    Int64,
+    Unsigned8,
+    Unsigned16,
+    Unsigned32,
+    Unsigned64,
+}
+
+impl DataType {
+    pub fn get_data_length(self) -> u64 {
+        match self {
+            DataType::Float32 => 4,
+            DataType::Float64 => 8,
+            DataType::Boolean => 1,
+            DataType::Int8 => 1,
+            DataType::Int16 => 2,
+            DataType::Int32 => 4,
+            DataType::Int64 => 8,
+            DataType::Unsigned8 => 1,
+            DataType::Unsigned16 => 2,
+            DataType::Unsigned32 => 4,
+            DataType::Unsigned64 => 8,
+        }
+    }
 }
 
 pub trait ToDataType: Into<DataValue> {
@@ -23,3 +50,11 @@ macro_rules! impl_to_data_type {
 impl_to_data_type!(f32, DataType::Float32);
 impl_to_data_type!(f64, DataType::Float64);
 impl_to_data_type!(bool, DataType::Boolean);
+impl_to_data_type!(i8, DataType::Int8);
+impl_to_data_type!(i16, DataType::Int16);
+impl_to_data_type!(i32, DataType::Int32);
+impl_to_data_type!(i64, DataType::Int64);
+impl_to_data_type!(u8, DataType::Unsigned8);
+impl_to_data_type!(u16, DataType::Unsigned16);
+impl_to_data_type!(u32, DataType::Unsigned32);
+impl_to_data_type!(u64, DataType::Unsigned64);

api-core/src/data_value.rs

@@ -1,9 +1,36 @@
+use crate::data_type::DataType;
 use derive_more::{From, TryInto};
 use serde::{Deserialize, Serialize};
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, From, TryInto)]
+#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, From, TryInto)]
 pub enum DataValue {
     Float32(f32),
     Float64(f64),
     Boolean(bool),
+    Int8(i8),
+    Int16(i16),
+    Int32(i32),
+    Int64(i64),
+    Unsigned8(u8),
+    Unsigned16(u16),
+    Unsigned32(u32),
+    Unsigned64(u64),
+}
+
+impl DataValue {
+    pub fn to_data_type(self) -> DataType {
+        match self {
+            DataValue::Float32(_) => DataType::Float32,
+            DataValue::Float64(_) => DataType::Float64,
+            DataValue::Boolean(_) => DataType::Boolean,
+            DataValue::Int8(_) => DataType::Int8,
+            DataValue::Int16(_) => DataType::Int16,
+            DataValue::Int32(_) => DataType::Int32,
+            DataValue::Int64(_) => DataType::Int64,
+            DataValue::Unsigned8(_) => DataType::Unsigned8,
+            DataValue::Unsigned16(_) => DataType::Unsigned16,
+            DataValue::Unsigned32(_) => DataType::Unsigned32,
+            DataValue::Unsigned64(_) => DataType::Unsigned64,
+        }
+    }
 }

api-proc-macro/Cargo.toml

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

api-proc-macro/src/into_command_definition.rs

@@ -70,10 +70,7 @@ pub fn derive_into_command_definition_impl(
             });
             quote! { #(#field_entries)* }
         }
-        Fields::Unnamed(fields) => abort!(
-            fields,
-            "IntoCommandDefinition not supported for unnamed structs"
-        ),
+        Fields::Unnamed(_) => unreachable!("Already checked this"),
         Fields::Unit => quote! {},
     };
     let param_name_stream = match &data.fields {
@@ -84,10 +81,7 @@ pub fn derive_into_command_definition_impl(
             });
             quote! { #(#field_entries)* }
         }
-        Fields::Unnamed(fields) => abort!(
-            fields,
-            "IntoCommandDefinition not supported for unnamed structs"
-        ),
+        Fields::Unnamed(_) => unreachable!("Already checked this"),
         Fields::Unit => quote! {},
     };
 

api-proc-macro/tests/test_derive_macro_into_command_definition.rs

@@ -149,3 +149,22 @@ fn test_generic_command() {
     .unwrap();
     assert_eq!(result.a, true);
 }
+
+#[test]
+fn test_unit_command() {
+    #[derive(IntoCommandDefinition)]
+    struct TestStruct;
+
+    let command_definition = TestStruct::create("Test".to_string());
+
+    assert_eq!(command_definition.name, "Test");
+    assert_eq!(command_definition.parameters.capacity(), 0);
+
+    TestStruct::parse(Command {
+        header: CommandHeader {
+            timestamp: Default::default(),
+        },
+        parameters: HashMap::new(),
+    })
+    .unwrap();
+}

api/Cargo.toml

@@ -2,12 +2,12 @@
 [package]
 name = "api"
 edition = "2021"
-version = "0.1.0"
-authors = ["Sergey <me@sergeysav.com>"]
+version.workspace = true
+authors.workspace = true
 
 [dependencies]
-api-core = { path = "../api-core" }
-api-proc-macro = { path = "../api-proc-macro" }
+api-core = { workspace = true }
+api-proc-macro = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 derive_more = { workspace = true, features = ["from", "try_into"] }
 futures-util = { workspace = true }
@@ -19,3 +19,6 @@ 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

@@ -1,8 +1,9 @@
 use crate::client::Client;
 use crate::messages::command::CommandResponse;
-use api_core::command::{CommandHeader, IntoCommandDefinition};
+use api_core::command::{Command, CommandDefinition, CommandHeader, IntoCommandDefinition};
 use std::fmt::Display;
 use std::sync::Arc;
+use tokio::select;
 use tokio_util::sync::CancellationToken;
 
 pub struct CommandRegistry {
@@ -14,13 +15,13 @@ impl CommandRegistry {
         Self { client }
     }
 
-    pub fn register_handler<C: IntoCommandDefinition, F, E: Display>(
+    pub fn register_raw_handler<F, E: Display>(
         &self,
-        command_name: impl Into<String>,
+        command_definition: CommandDefinition,
         mut callback: F,
     ) -> CommandHandle
     where
-        F: FnMut(CommandHeader, C) -> Result<String, E> + Send + 'static,
+        F: FnMut(Command) -> Result<String, E> + Send + 'static,
     {
         let cancellation_token = CancellationToken::new();
         let result = CommandHandle {
@@ -28,8 +29,6 @@ impl CommandRegistry {
         };
         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
@@ -41,32 +40,52 @@ impl CommandRegistry {
                     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(),
-                            },
+                loop {
+                    // select used so that this loop gets broken if the token is cancelled
+                    select!(
+                        rx_value = rx.recv() => {
+                            if let Some((cmd, responder)) = rx_value {
+                                let response = match callback(cmd) {
+                                    Ok(response) => CommandResponse {
+                                        success: true,
+                                        response,
+                                    },
+                                    Err(err) => CommandResponse {
+                                        success: false,
+                                        response: err.to_string(),
+                                    },
+                                };
+                                // This should only err if we had an error elsewhere
+                                let _ = responder.send(response);
+                            } else {
+                                break;
+                            }
                         },
-                        Err(err) => CommandResponse {
-                            success: false,
-                            response: err.to_string(),
-                        },
-                    };
-                    // This should only err if we had an error elsewhere
-                    let _ = responder.send(response);
+                        _ = cancellation_token.cancelled() => { break; },
+                    );
                 }
             }
         });
 
         result
     }
+
+    pub fn register_handler<C: IntoCommandDefinition, F, E>(
+        &self,
+        command_name: impl Into<String>,
+        mut callback: F,
+    ) -> CommandHandle
+    where
+        F: FnMut(CommandHeader, C) -> Result<String, E> + Send + 'static,
+        E: Display,
+    {
+        self.register_raw_handler(C::create(command_name.into()), move |command: Command| {
+            let header = command.header.clone();
+            C::parse(command)
+                .map_err(|e| e.to_string())
+                .and_then(|cmd| callback(header, cmd).map_err(|err| err.to_string()))
+        })
+    }
 }
 
 pub struct CommandHandle {
@@ -78,3 +97,366 @@ impl Drop for CommandHandle {
         self.cancellation_token.cancel();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::client::command::CommandRegistry;
+    use crate::client::tests::create_test_client;
+    use crate::client::Callback;
+    use crate::messages::callback::GenericCallbackError;
+    use crate::messages::command::CommandResponse;
+    use crate::messages::payload::RequestMessagePayload;
+    use crate::messages::telemetry_definition::TelemetryDefinitionResponse;
+    use crate::messages::ResponseMessage;
+    use api_core::command::{
+        Command, CommandDefinition, CommandHeader, CommandParameterDefinition,
+        IntoCommandDefinition, IntoCommandDefinitionError,
+    };
+    use api_core::data_type::DataType;
+    use std::collections::HashMap;
+    use std::convert::Infallible;
+    use std::sync::Arc;
+    use std::time::Duration;
+    use tokio::sync::oneshot;
+    use tokio::time::timeout;
+    use uuid::Uuid;
+
+    struct CmdType {
+        #[allow(unused)]
+        param1: f32,
+    }
+
+    impl IntoCommandDefinition for CmdType {
+        fn create(name: String) -> CommandDefinition {
+            CommandDefinition {
+                name,
+                parameters: vec![CommandParameterDefinition {
+                    name: "param1".to_string(),
+                    data_type: DataType::Float32,
+                }],
+            }
+        }
+
+        fn parse(command: Command) -> Result<Self, IntoCommandDefinitionError> {
+            Ok(Self {
+                param1: (*command.parameters.get("param1").ok_or_else(|| {
+                    IntoCommandDefinitionError::ParameterMissing("param1".to_string())
+                })?)
+                .try_into()
+                .map_err(|_| IntoCommandDefinitionError::MismatchedType {
+                    parameter: "param1".to_string(),
+                    expected: DataType::Float32,
+                })?,
+            })
+        }
+    }
+
+    #[tokio::test]
+    async fn simple_handler() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let cmd_reg = CommandRegistry::new(Arc::new(client));
+
+        let _cmd_handle = cmd_reg.register_handler("cmd", |_, _: CmdType| {
+            Ok("success".to_string()) as Result<_, Infallible>
+        });
+
+        let msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let Callback::Registered(callback) = msg.callback else {
+            panic!("Incorrect Callback Type");
+        };
+
+        let mut params = HashMap::new();
+        params.insert("param1".to_string(), 0.0f32.into());
+
+        let (response_tx, response_rx) = oneshot::channel();
+        timeout(
+            Duration::from_secs(1),
+            callback.send((
+                ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: Command {
+                        header: CommandHeader {
+                            timestamp: Default::default(),
+                        },
+                        parameters: params,
+                    }
+                    .into(),
+                },
+                response_tx,
+            )),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        let response = timeout(Duration::from_secs(1), response_rx)
+            .await
+            .unwrap()
+            .unwrap();
+        let RequestMessagePayload::CommandResponse(CommandResponse { success, response }) =
+            response
+        else {
+            panic!("Unexpected Response Type");
+        };
+        assert!(success);
+        assert_eq!(response, "success");
+    }
+
+    #[tokio::test]
+    async fn handler_failed() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let cmd_reg = CommandRegistry::new(Arc::new(client));
+
+        let _cmd_handle = cmd_reg.register_handler("cmd", |_, _: CmdType| {
+            Err("failure".into()) as Result<_, Box<dyn std::error::Error>>
+        });
+
+        let msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let Callback::Registered(callback) = msg.callback else {
+            panic!("Incorrect Callback Type");
+        };
+
+        let mut params = HashMap::new();
+        params.insert("param1".to_string(), 1.0f32.into());
+
+        let (response_tx, response_rx) = oneshot::channel();
+        timeout(
+            Duration::from_secs(1),
+            callback.send((
+                ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: Command {
+                        header: CommandHeader {
+                            timestamp: Default::default(),
+                        },
+                        parameters: params,
+                    }
+                    .into(),
+                },
+                response_tx,
+            )),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        let response = timeout(Duration::from_secs(1), response_rx)
+            .await
+            .unwrap()
+            .unwrap();
+        let RequestMessagePayload::CommandResponse(CommandResponse { success, response }) =
+            response
+        else {
+            panic!("Unexpected Response Type");
+        };
+        assert!(!success);
+        assert_eq!(response, "failure");
+    }
+
+    #[tokio::test]
+    async fn parse_failed() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let cmd_reg = CommandRegistry::new(Arc::new(client));
+
+        let _cmd_handle = cmd_reg.register_handler("cmd", |_, _: CmdType| {
+            Err("failure".into()) as Result<_, Box<dyn std::error::Error>>
+        });
+
+        let msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let Callback::Registered(callback) = msg.callback else {
+            panic!("Incorrect Callback Type");
+        };
+
+        let mut params = HashMap::new();
+        params.insert("param1".to_string(), 1.0f64.into());
+
+        let (response_tx, response_rx) = oneshot::channel();
+        timeout(
+            Duration::from_secs(1),
+            callback.send((
+                ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: Command {
+                        header: CommandHeader {
+                            timestamp: Default::default(),
+                        },
+                        parameters: params,
+                    }
+                    .into(),
+                },
+                response_tx,
+            )),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        let response = timeout(Duration::from_secs(1), response_rx)
+            .await
+            .unwrap()
+            .unwrap();
+        let RequestMessagePayload::CommandResponse(CommandResponse {
+            success,
+            response: _,
+        }) = response
+        else {
+            panic!("Unexpected Response Type");
+        };
+        assert!(!success);
+    }
+
+    #[tokio::test]
+    async fn wrong_message() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let cmd_reg = CommandRegistry::new(Arc::new(client));
+
+        let _cmd_handle =
+            cmd_reg.register_handler("cmd", |_, _: CmdType| -> Result<_, Infallible> {
+                panic!("This should not happen");
+            });
+
+        let msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let Callback::Registered(callback) = msg.callback else {
+            panic!("Incorrect Callback Type");
+        };
+
+        let (response_tx, response_rx) = oneshot::channel();
+        timeout(
+            Duration::from_secs(1),
+            callback.send((
+                ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: TelemetryDefinitionResponse {
+                        uuid: Uuid::new_v4(),
+                    }
+                    .into(),
+                },
+                response_tx,
+            )),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        let response = timeout(Duration::from_secs(1), response_rx)
+            .await
+            .unwrap()
+            .unwrap();
+        let RequestMessagePayload::GenericCallbackError(err) = response else {
+            panic!("Unexpected Response Type");
+        };
+        assert_eq!(err, GenericCallbackError::MismatchedType);
+    }
+
+    #[tokio::test]
+    async fn callback_closed() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let cmd_reg = CommandRegistry::new(Arc::new(client));
+
+        let cmd_handle =
+            cmd_reg.register_handler("cmd", |_, _: CmdType| -> Result<_, Infallible> {
+                panic!("This should not happen");
+            });
+
+        let msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let Callback::Registered(callback) = msg.callback else {
+            panic!("Incorrect Callback Type");
+        };
+
+        // This should shut down the command handler
+        drop(cmd_handle);
+
+        // Send a command
+        let mut params = HashMap::new();
+        params.insert("param1".to_string(), 0.0f32.into());
+        let (response_tx, response_rx) = oneshot::channel();
+        timeout(
+            Duration::from_secs(1),
+            callback.send((
+                ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: Command {
+                        header: CommandHeader {
+                            timestamp: Default::default(),
+                        },
+                        parameters: params,
+                    }
+                    .into(),
+                },
+                response_tx,
+            )),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+
+        let response = timeout(Duration::from_secs(1), response_rx)
+            .await
+            .unwrap()
+            .unwrap();
+        let RequestMessagePayload::GenericCallbackError(err) = response else {
+            panic!("Unexpected Response Type");
+        };
+        assert_eq!(err, GenericCallbackError::CallbackClosed);
+    }
+
+    #[tokio::test]
+    async fn reconnect() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let cmd_reg = CommandRegistry::new(Arc::new(client));
+
+        let _cmd_handle =
+            cmd_reg.register_handler("cmd", |_, _: CmdType| -> Result<_, Infallible> {
+                panic!("This should not happen");
+            });
+
+        let msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let Callback::Registered(callback) = msg.callback else {
+            panic!("Incorrect Callback Type");
+        };
+
+        println!("Dropping");
+        drop(callback);
+        println!("Dropped");
+
+        // The command re-registers itself
+        let msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let Callback::Registered(_) = msg.callback else {
+            panic!("Incorrect Callback Type");
+        };
+    }
+}

api/src/client/context.rs

@@ -4,19 +4,19 @@ 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::{SinkExt, StreamExt};
+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::net::TcpStream;
 use tokio::sync::{mpsc, oneshot, watch, RwLockWriteGuard};
 use tokio::time::sleep;
 use tokio::{select, spawn};
-use tokio_tungstenite::tungstenite::handshake::client::Request;
-use tokio_tungstenite::tungstenite::Message;
-use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
+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;
 
@@ -45,7 +45,9 @@ impl ClientContext {
                     let _ = tx.send(());
 
                     while !self.cancel.is_cancelled() {
-                        write_lock = self.run_connection(write_lock, &channel).await;
+                        write_lock = self
+                            .run_connection(write_lock, &channel, connect_async)
+                            .await;
                     }
                     drop(write_lock);
                 });
@@ -57,13 +59,19 @@ impl ClientContext {
         Ok(())
     }
 
-    async fn run_connection<'a>(
+    async fn run_connection<'a, F, W, E>(
         &mut self,
         mut write_lock: RwLockWriteGuard<'a, mpsc::Sender<OutgoingMessage>>,
         channel: &'a ClientChannel,
-    ) -> RwLockWriteGuard<'a, mpsc::Sender<OutgoingMessage>> {
+        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 connect_async(self.request.clone()).await {
+        let mut ws = match connection_fn(self.request.clone()).await {
             Ok((ws, _)) => ws,
             Err(e) => {
                 info!("Failed to Connect: {e}");
@@ -87,19 +95,24 @@ impl ClientContext {
         // the lock to use that as a signal that we have reconnected
         let _ = self.connected_state_tx.send_replace(false);
         if close_connection {
-            if let Err(e) = ws.close(None).await {
+            // 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(
+    async fn handle_connection<W>(
         &mut self,
-        ws: &mut WebSocketStream<MaybeTlsStream<TcpStream>>,
+        ws: &mut W,
         mut rx: mpsc::Receiver<OutgoingMessage>,
         channel: &ClientChannel,
-    ) -> bool {
+    ) -> 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! {
@@ -242,3 +255,340 @@ impl ClientContext {
         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

@@ -1,3 +1,4 @@
+use api_core::data_type::DataType;
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -16,6 +17,11 @@ pub enum MessageError {
     TokioTrySendError(#[from] tokio::sync::mpsc::error::TrySendError<()>),
     #[error(transparent)]
     TokioLockError(#[from] tokio::sync::TryLockError),
+    #[error("Incorrect Data Type. {expected} expected. {actual} actual.")]
+    IncorrectDataType {
+        expected: DataType,
+        actual: DataType,
+    },
 }
 
 #[derive(Error, Debug)]

api/src/client/mod.rs

@@ -24,12 +24,14 @@ 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),
 }
 
+#[derive(Debug)]
 struct OutgoingMessage {
     msg: RequestMessage,
     callback: Callback,
@@ -264,3 +266,332 @@ impl Drop for Client {
         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::pin::pin;
+    use std::time::Duration;
+    use tokio::join;
+    use tokio::time::{sleep, timeout};
+
+    pub 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 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 |_| {
+                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

@@ -75,6 +75,7 @@ impl TelemetryRegistry {
                 cancellation_token,
                 uuid: response_uuid,
                 client: stored_client,
+                data_type,
             }
         }
         inner(self.client.clone(), name.into(), data_type).await
@@ -96,6 +97,7 @@ pub struct GenericTelemetryHandle {
     cancellation_token: CancellationToken,
     uuid: Arc<RwLock<Uuid>>,
     client: Arc<Client>,
+    data_type: DataType,
 }
 
 impl GenericTelemetryHandle {
@@ -104,6 +106,12 @@ impl GenericTelemetryHandle {
         value: DataValue,
         timestamp: DateTime<Utc>,
     ) -> Result<(), MessageError> {
+        if value.to_data_type() != self.data_type {
+            return Err(MessageError::IncorrectDataType {
+                expected: self.data_type,
+                actual: value.to_data_type(),
+            });
+        }
         let Ok(lock) = self.uuid.try_read() else {
             return Ok(());
         };
@@ -163,3 +171,291 @@ impl<T: Into<DataValue>> TelemetryHandle<T> {
         self.publish(value, Utc::now()).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::client::error::MessageError;
+    use crate::client::telemetry::TelemetryRegistry;
+    use crate::client::tests::create_test_client;
+    use crate::client::Callback;
+    use crate::messages::payload::RequestMessagePayload;
+    use crate::messages::telemetry_definition::{
+        TelemetryDefinitionRequest, TelemetryDefinitionResponse,
+    };
+    use crate::messages::telemetry_entry::TelemetryEntry;
+    use crate::messages::ResponseMessage;
+    use api_core::data_type::DataType;
+    use api_core::data_value::DataValue;
+    use futures_util::FutureExt;
+    use std::sync::Arc;
+    use std::time::Duration;
+    use tokio::task::yield_now;
+    use tokio::time::timeout;
+    use tokio::try_join;
+    use uuid::Uuid;
+
+    #[tokio::test]
+    async fn generic() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let tlm = TelemetryRegistry::new(Arc::new(client));
+        let tlm_handle = tlm.register_generic("generic", DataType::Float32);
+
+        let tlm_uuid = Uuid::new_v4();
+
+        let expected_rx = async {
+            let msg = rx.recv().await.unwrap();
+            let Callback::Once(responder) = msg.callback else {
+                panic!("Expected Once Callback");
+            };
+            assert!(msg.msg.response.is_none());
+            let RequestMessagePayload::TelemetryDefinitionRequest(TelemetryDefinitionRequest {
+                name,
+                data_type,
+            }) = msg.msg.payload
+            else {
+                panic!("Expected Telemetry Definition Request")
+            };
+            assert_eq!(name, "generic".to_string());
+            assert_eq!(data_type, DataType::Float32);
+            responder
+                .send(ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: TelemetryDefinitionResponse { uuid: tlm_uuid }.into(),
+                })
+                .unwrap();
+        };
+
+        let (tlm_handle, _) = try_join!(
+            timeout(Duration::from_secs(1), tlm_handle),
+            timeout(Duration::from_secs(1), expected_rx),
+        )
+        .unwrap();
+
+        assert_eq!(*tlm_handle.uuid.try_read().unwrap(), tlm_uuid);
+
+        // This should NOT block if there is space in the queue
+        tlm_handle
+            .publish_now(0.0f32.into())
+            .now_or_never()
+            .unwrap()
+            .unwrap();
+
+        let tlm_msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(matches!(tlm_msg.callback, Callback::None));
+        match tlm_msg.msg.payload {
+            RequestMessagePayload::TelemetryEntry(TelemetryEntry { uuid, value, .. }) => {
+                assert_eq!(uuid, tlm_uuid);
+                assert_eq!(value, DataValue::Float32(0.0f32));
+            }
+            _ => panic!("Expected Telemetry Entry"),
+        }
+    }
+
+    #[tokio::test]
+    async fn mismatched_type() {
+        let (mut rx, _, client) = create_test_client();
+
+        let tlm = TelemetryRegistry::new(Arc::new(client));
+        let tlm_handle = tlm.register_generic("generic", DataType::Float32);
+
+        let tlm_uuid = Uuid::new_v4();
+
+        let expected_rx = async {
+            let msg = rx.recv().await.unwrap();
+            let Callback::Once(responder) = msg.callback else {
+                panic!("Expected Once Callback");
+            };
+            assert!(msg.msg.response.is_none());
+            let RequestMessagePayload::TelemetryDefinitionRequest(TelemetryDefinitionRequest {
+                name,
+                data_type,
+            }) = msg.msg.payload
+            else {
+                panic!("Expected Telemetry Definition Request")
+            };
+            assert_eq!(name, "generic".to_string());
+            assert_eq!(data_type, DataType::Float32);
+            responder
+                .send(ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: TelemetryDefinitionResponse { uuid: tlm_uuid }.into(),
+                })
+                .unwrap();
+        };
+
+        let (tlm_handle, _) = try_join!(
+            timeout(Duration::from_secs(1), tlm_handle),
+            timeout(Duration::from_secs(1), expected_rx),
+        )
+        .unwrap();
+
+        assert_eq!(*tlm_handle.uuid.try_read().unwrap(), tlm_uuid);
+
+        match timeout(
+            Duration::from_secs(1),
+            tlm_handle.publish_now(0.0f64.into()),
+        )
+        .await
+        .unwrap()
+        {
+            Err(MessageError::IncorrectDataType { expected, actual }) => {
+                assert_eq!(expected, DataType::Float32);
+                assert_eq!(actual, DataType::Float64);
+            }
+            _ => panic!("Error Expected"),
+        }
+    }
+
+    #[tokio::test]
+    async fn typed() {
+        // if _c drops then we are disconnected
+        let (mut rx, _c, client) = create_test_client();
+
+        let tlm = TelemetryRegistry::new(Arc::new(client));
+        let tlm_handle = tlm.register::<bool>("typed");
+
+        let tlm_uuid = Uuid::new_v4();
+
+        let expected_rx = async {
+            let msg = rx.recv().await.unwrap();
+            let Callback::Once(responder) = msg.callback else {
+                panic!("Expected Once Callback");
+            };
+            assert!(msg.msg.response.is_none());
+            let RequestMessagePayload::TelemetryDefinitionRequest(TelemetryDefinitionRequest {
+                name,
+                data_type,
+            }) = msg.msg.payload
+            else {
+                panic!("Expected Telemetry Definition Request")
+            };
+            assert_eq!(name, "typed".to_string());
+            assert_eq!(data_type, DataType::Boolean);
+            responder
+                .send(ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: TelemetryDefinitionResponse { uuid: tlm_uuid }.into(),
+                })
+                .unwrap();
+        };
+
+        let (tlm_handle, _) = try_join!(
+            timeout(Duration::from_secs(1), tlm_handle),
+            timeout(Duration::from_secs(1), expected_rx),
+        )
+        .unwrap();
+
+        assert_eq!(*tlm_handle.as_generic().uuid.try_read().unwrap(), tlm_uuid);
+
+        // This should NOT block if there is space in the queue
+        tlm_handle
+            .publish_now(true)
+            .now_or_never()
+            .unwrap()
+            .unwrap();
+        // This should block as there should not be space in the queue
+        assert!(tlm_handle.publish_now(false).now_or_never().is_none());
+
+        let tlm_msg = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(matches!(tlm_msg.callback, Callback::None));
+        match tlm_msg.msg.payload {
+            RequestMessagePayload::TelemetryEntry(TelemetryEntry { uuid, value, .. }) => {
+                assert_eq!(uuid, tlm_uuid);
+                assert_eq!(value, DataValue::Boolean(true));
+            }
+            _ => panic!("Expected Telemetry Entry"),
+        }
+
+        let _make_generic_again = tlm_handle.to_generic();
+    }
+
+    #[tokio::test]
+    async fn reconnect() {
+        // if _c drops then we are disconnected
+        let (mut rx, connected, client) = create_test_client();
+
+        let tlm = TelemetryRegistry::new(Arc::new(client));
+        let tlm_handle = tlm.register_generic("generic", DataType::Float32);
+
+        let tlm_uuid = Uuid::new_v4();
+
+        let expected_rx = async {
+            let msg = rx.recv().await.unwrap();
+            let Callback::Once(responder) = msg.callback else {
+                panic!("Expected Once Callback");
+            };
+            assert!(msg.msg.response.is_none());
+            let RequestMessagePayload::TelemetryDefinitionRequest(TelemetryDefinitionRequest {
+                name,
+                data_type,
+            }) = msg.msg.payload
+            else {
+                panic!("Expected Telemetry Definition Request")
+            };
+            assert_eq!(name, "generic".to_string());
+            assert_eq!(data_type, DataType::Float32);
+            responder
+                .send(ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: TelemetryDefinitionResponse { uuid: tlm_uuid }.into(),
+                })
+                .unwrap();
+        };
+
+        let (tlm_handle, _) = try_join!(
+            timeout(Duration::from_secs(1), tlm_handle),
+            timeout(Duration::from_secs(1), expected_rx),
+        )
+        .unwrap();
+
+        assert_eq!(*tlm_handle.uuid.try_read().unwrap(), tlm_uuid);
+
+        // Notify Disconnect
+        connected.send_replace(false);
+        // Notify Reconnect
+        connected.send_replace(true);
+
+        {
+            let new_tlm_uuid = Uuid::new_v4();
+
+            let msg = rx.recv().await.unwrap();
+            let Callback::Once(responder) = msg.callback else {
+                panic!("Expected Once Callback");
+            };
+            assert!(msg.msg.response.is_none());
+            let RequestMessagePayload::TelemetryDefinitionRequest(TelemetryDefinitionRequest {
+                name,
+                data_type,
+            }) = msg.msg.payload
+            else {
+                panic!("Expected Telemetry Definition Request")
+            };
+            assert_eq!(name, "generic".to_string());
+            assert_eq!(data_type, DataType::Float32);
+            responder
+                .send(ResponseMessage {
+                    uuid: Uuid::new_v4(),
+                    response: Some(msg.msg.uuid),
+                    payload: TelemetryDefinitionResponse { uuid: new_tlm_uuid }.into(),
+                })
+                .unwrap();
+
+            // Yield to the executor so that the UUIDs can be updated
+            yield_now().await;
+
+            assert_eq!(*tlm_handle.uuid.try_read().unwrap(), new_tlm_uuid);
+        }
+    }
+}

api/src/data_type.rs

@@ -1 +0,0 @@
-pub use api_core::data_type::*;

api/src/data_value.rs

@@ -1 +0,0 @@
-pub use api_core::data_value::*;

api/src/lib.rs

@@ -1,8 +1,15 @@
 pub mod client;
-pub mod data_type;
-pub mod data_value;
+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

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

api/src/messages/command.rs

@@ -8,7 +8,7 @@ impl RegisterCallback for CommandDefinition {
     type Response = CommandResponse;
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct CommandResponse {
     pub success: bool,
     pub response: String,

api/src/messages/mod.rs

@@ -18,7 +18,7 @@ pub struct RequestMessage {
     pub payload: RequestMessagePayload,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ResponseMessage {
     pub uuid: Uuid,
     #[serde(default)]

api/src/messages/payload.rs

@@ -7,7 +7,7 @@ use crate::messages::telemetry_entry::TelemetryEntry;
 use derive_more::{From, TryInto};
 use serde::{Deserialize, Serialize};
 
-#[derive(Debug, Clone, Serialize, Deserialize, From)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, From)]
 pub enum RequestMessagePayload {
     TelemetryDefinitionRequest(TelemetryDefinitionRequest),
     TelemetryEntry(TelemetryEntry),
@@ -16,7 +16,7 @@ pub enum RequestMessagePayload {
     CommandResponse(CommandResponse),
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, From, TryInto)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, From, TryInto)]
 pub enum ResponseMessagePayload {
     TelemetryDefinitionResponse(TelemetryDefinitionResponse),
     Command(Command),

api/src/messages/telemetry_definition.rs

@@ -3,13 +3,13 @@ use crate::messages::RequestResponse;
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct TelemetryDefinitionRequest {
     pub name: String,
     pub data_type: DataType,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct TelemetryDefinitionResponse {
     pub uuid: Uuid,
 }

api/src/messages/telemetry_entry.rs

@@ -4,7 +4,7 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct TelemetryEntry {
     pub uuid: Uuid,
     pub value: DataValue,

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

@@ -5,7 +5,7 @@ edition = "2021"
 
 [dependencies]
 anyhow = { workspace = true }
-api = { path = "../../api" }
+api = { workspace = true }
 env_logger = { workspace = true }
 log = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "signal"] }

examples/simple_producer/Cargo.toml

@@ -5,12 +5,10 @@ edition = "2021"
 
 [dependencies]
 anyhow = { workspace = true }
-api = { path = "../../api" }
+api = { workspace = true }
 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"

frontend/src/components/CommandInput.vue

@@ -3,8 +3,10 @@ import { computed, onMounted } from 'vue';
 import {
     type AnyTypeId,
     type DynamicDataType,
+    getLimits,
     isBooleanType,
     isNumericType,
+    type NumericLimits,
 } from '@/composables/dynamic.ts';
 
 const props = defineProps<{
@@ -17,23 +19,100 @@ const is_numeric = computed(() => {
     return isNumericType(props.type);
 });
 
+const numeric_limit = computed(() => {
+    if (isNumericType(props.type)) {
+        return getLimits(props.type);
+    }
+    return {
+        min: 0,
+        max: 0,
+        integer: false,
+        is_big: false,
+    } as NumericLimits;
+});
+
 const is_boolean = computed(() => {
     return isBooleanType(props.type);
 });
 
 // Initialize the parameter to some value:
 onMounted(() => {
-    if (is_numeric.value) {
-        model.value = 0.0;
-    } else if (is_boolean.value) {
-        model.value = false;
+    if (model.value === undefined) {
+        if (is_numeric.value) {
+            if (numeric_limit.value.integer) {
+                if (numeric_limit.value.is_big) {
+                    model.value = 0n;
+                } else {
+                    model.value = 0;
+                }
+            } else {
+                model.value = 0.0;
+            }
+        } else if (is_boolean.value) {
+            model.value = false;
+        }
     }
 });
+
+const value = computed({
+    get: () => {
+        if (is_numeric.value) {
+            if (numeric_limit.value.is_big) {
+                return ((model.value || 0n) as bigint).toString();
+            } else {
+                return model.value as number;
+            }
+        } else if (is_boolean.value) {
+            // Force it into a boolean
+            return !!model.value;
+        }
+        return undefined;
+    },
+    set: (value) => {
+        if (is_numeric.value) {
+            if (numeric_limit.value.is_big) {
+                let result = BigInt(value as string);
+                if (result > (numeric_limit.value.max as bigint)) {
+                    result = numeric_limit.value.max as bigint;
+                } else if (result < (numeric_limit.value.min as bigint)) {
+                    result = numeric_limit.value.min as bigint;
+                }
+                model.value = result;
+            } else {
+                let result_value = value as number;
+                if (numeric_limit.value.integer) {
+                    result_value = Math.round(value as number);
+                }
+                if (result_value > (numeric_limit.value.max as number)) {
+                    result_value = numeric_limit.value.max as number;
+                } else if (result_value < (numeric_limit.value.min as number)) {
+                    result_value = numeric_limit.value.min as number;
+                }
+                model.value = result_value;
+            }
+        } else if (is_boolean.value) {
+            model.value = value as boolean;
+        }
+    },
+});
 </script>
 
 <template>
-    <input v-if="is_numeric" type="number" v-model="model" />
-    <input v-else-if="is_boolean" type="checkbox" v-model="model" />
+    <input
+        v-if="is_numeric && !numeric_limit.is_big"
+        type="number"
+        v-model="value"
+        :max="numeric_limit.max as number"
+        :min="numeric_limit.min as number"
+        :step="numeric_limit.integer ? 1 : undefined"
+    />
+    <input
+        v-else-if="is_numeric && numeric_limit.is_big"
+        type="text"
+        v-model="value"
+        pattern="-?\d+"
+    />
+    <input v-else-if="is_boolean" type="checkbox" v-model="value" />
     <span v-else>UNKNOWN INPUT</span>
 </template>
 

frontend/src/components/CommandParameterListConfigurator.vue

@@ -2,6 +2,7 @@
 import {
     type CommandParameterData,
     type DynamicDataType,
+    getLimits,
     isBooleanType,
     isNumericType,
 } from '@/composables/dynamic.ts';
@@ -18,7 +19,9 @@ const model = defineModel<{ [key: string]: CommandParameterData }>({
     required: true,
 });
 
-const { data: command_info } = useCommand(props.command_name);
+const { data: command_info, error: command_error } = useCommand(
+    props.command_name,
+);
 
 watch([command_info], ([cmd_info]) => {
     if (cmd_info == null) {
@@ -40,7 +43,8 @@ watch([command_info], ([cmd_info]) => {
             switch (model_param_value.type) {
                 case 'constant':
                     if (
-                        typeof model_param_value.value == 'number' &&
+                        (typeof model_param_value.value == 'number' ||
+                            typeof model_param_value.value == 'bigint') &&
                         !isNumericType(param.data_type)
                     ) {
                         model_param_value = undefined;
@@ -56,10 +60,14 @@ watch([command_info], ([cmd_info]) => {
                     break;
             }
         }
-        if (!model_param_value) {
+        if (model_param_value === undefined) {
             let default_value: DynamicDataType = 0;
             if (isNumericType(param.data_type)) {
-                default_value = 0;
+                if (getLimits(param.data_type).is_big) {
+                    default_value = 0n;
+                } else {
+                    default_value = 0;
+                }
             } else if (isBooleanType(param.data_type)) {
                 default_value = false;
             }
@@ -68,6 +76,21 @@ watch([command_info], ([cmd_info]) => {
                 value: default_value,
             };
         }
+        // Perform some cleanup to remove data that shouldn't be there
+        switch (model_param_value.type) {
+            case 'constant':
+                model_param_value = {
+                    type: model_param_value.type,
+                    value: model_param_value.value,
+                };
+                break;
+            case 'input':
+                model_param_value = {
+                    type: model_param_value.type,
+                    id: model_param_value.id,
+                };
+                break;
+        }
         model_value[param.name] = model_param_value;
     }
     model.value = model_value;
@@ -91,6 +114,9 @@ watch([command_info], ([cmd_info]) => {
             ></CommandParameterDataConfigurator>
         </template>
     </template>
+    <template v-else-if="command_error">
+        <span>Error: {{ command_error }}</span>
+    </template>
     <template v-else>
         <span> Loading... </span>
     </template>

frontend/src/components/CommandSender.vue

@@ -4,6 +4,7 @@ import { ref } from 'vue';
 import CommandParameter from '@/components/CommandParameter.vue';
 import FlexDivider from '@/components/FlexDivider.vue';
 import type { DynamicDataType } from '@/composables/dynamic.ts';
+import { parseJsonString, toJsonString } from '@/composables/json.ts';
 
 const props = defineProps<{
     command: CommandDefinition | null;
@@ -27,12 +28,13 @@ async function sendCommand() {
         headers: {
             'Content-Type': 'application/json',
         },
-        body: JSON.stringify(params),
+        body: toJsonString(params),
     });
+    const text_response = await response.text();
     if (response.ok) {
-        result.value = await response.json();
+        result.value = parseJsonString(text_response);
     } else {
-        result.value = await response.text();
+        result.value = text_response;
     }
     busy.value = false;
 }

frontend/src/components/DynamicComponent.vue

@@ -7,6 +7,7 @@ import {
 } from '@/composables/dynamic.ts';
 import { computed, defineAsyncComponent, inject, type Ref, ref } from 'vue';
 import CommandParameterListConfigurator from '@/components/CommandParameterListConfigurator.vue';
+import { toJsonString } from '@/composables/json.ts';
 
 const TelemetryValue = defineAsyncComponent(
     () => import('@/components/TelemetryValue.vue'),
@@ -166,7 +167,7 @@ async function sendCommand(command: {
         headers: {
             'Content-Type': 'application/json',
         },
-        body: JSON.stringify(params),
+        body: toJsonString(params),
     });
     busy.value = false;
 }

frontend/src/components/NumericText.vue

@@ -3,7 +3,7 @@ import { computed, watch } from 'vue';
 import CopyableDynamicSpan from '@/components/CopyableDynamicSpan.vue';
 
 const props = defineProps<{
-    value: number;
+    value: number | bigint;
     max_width: number;
     copyable?: boolean;
     include_span?: boolean;
@@ -23,6 +23,9 @@ const display_value = computed(() => {
     if (props.value == 0) {
         return '0';
     }
+    if (typeof props.value === 'bigint') {
+        return props.value.toString();
+    }
     let precision = props.value.toPrecision(props.max_width - 3);
     // Chop off the last character as long as it is a 0
     while (

frontend/src/components/TelemetryLine.vue

@@ -28,6 +28,7 @@ import {
     isBooleanType,
     isNumericType,
 } from '@/composables/dynamic.ts';
+import { parseJsonString } from '@/composables/json.ts';
 
 const props = defineProps<{
     data: string;
@@ -138,7 +139,14 @@ watch(
                 const res = await fetch(
                     `/api/tlm/history/${uuid}?from=${min_x.toISOString()}&to=${max_x.toISOString()}&resolution=${data_min_sep.value}`,
                 );
-                const response = (await res.json()) as TelemetryDataItem[];
+                const text_response = await res.text();
+                if (!res.ok) {
+                    console.error(text_response);
+                    return;
+                }
+                const response = parseJsonString(
+                    text_response,
+                ) as TelemetryDataItem[];
                 for (const data_item of response) {
                     const val_t = Date.parse(data_item.timestamp);
                     const raw_item_val = data_item.value[

frontend/src/components/TelemetryValue.vue

@@ -6,6 +6,8 @@ import {
     type WebsocketHandle,
 } from '@/composables/websocket.ts';
 import NumericText from '@/components/NumericText.vue';
+import BooleanText from '@/components/BooleanText.vue';
+import { isBooleanType, isNumericType } from '@/composables/dynamic.ts';
 
 const max_update_rate = 50; // ms
 const default_update_rate = 200; // ms
@@ -33,30 +35,23 @@ const value = websocket.value.listen_to_telemetry(
 const is_data_present = computed(() => {
     return value.value != null;
 });
-
-const numeric_data = computed(() => {
-    const val = value.value;
-    if (val) {
-        const type = telemetry_data.value!.data_type;
-        const item_val = val.value[type];
-        if (typeof item_val == 'number') {
-            return item_val;
-        }
-    }
-    return null;
-});
 </script>
 
 <template>
     <span v-if="!is_data_present" v-bind="$attrs"> No Data </span>
     <template v-else>
         <NumericText
-            v-if="numeric_data"
+            v-if="isNumericType(telemetry_data!.data_type)"
             v-bind="$attrs"
-            :value="numeric_data"
+            :value="value!.value[telemetry_data!.data_type]"
             :max_width="10"
             include_span
         ></NumericText>
+        <BooleanText
+            v-else-if="isBooleanType(telemetry_data!.data_type)"
+            v-bind="$attrs"
+            :value="value!.value[telemetry_data!.data_type]"
+        ></BooleanText>
         <span v-else v-bind="$attrs">
             Cannot Display Data of Type {{ telemetry_data!.data_type }}
         </span>

frontend/src/composables/command.ts

@@ -1,6 +1,7 @@
 import { ref, toValue, watchEffect } from 'vue';
 import { type MaybeRefOrGetter } from 'vue';
 import type { AnyTypeId } from '@/composables/dynamic.ts';
+import { parseJsonString } from '@/composables/json.ts';
 
 export interface CommandParameterDefinition {
     name: string;
@@ -20,8 +21,14 @@ export function useAllCommands() {
     watchEffect(async () => {
         try {
             const res = await fetch(`/api/cmd`);
-            data.value = await res.json();
-            error.value = null;
+            const text_response = await res.text();
+            if (res.ok) {
+                data.value = parseJsonString(text_response);
+                error.value = null;
+            } else {
+                data.value = null;
+                error.value = text_response;
+            }
         } catch (e) {
             data.value = null;
             error.value = e;
@@ -41,8 +48,14 @@ export function useCommand(name: MaybeRefOrGetter<string>) {
 
         try {
             const res = await fetch(`/api/cmd/${name_value}`);
-            data.value = await res.json();
-            error.value = null;
+            const text_response = await res.text();
+            if (res.ok) {
+                data.value = parseJsonString(text_response);
+                error.value = null;
+            } else {
+                data.value = null;
+                error.value = text_response;
+            }
         } catch (e) {
             data.value = null;
             error.value = e;

frontend/src/composables/dynamic.ts

@@ -1,5 +1,74 @@
-export const NumericTypes = ['Float32', 'Float64'] as const;
+export const NumericTypes = [
+    'Float32',
+    'Float64',
+    'Int8',
+    'Int16',
+    'Int32',
+    'Int64',
+    'Unsigned8',
+    'Unsigned16',
+    'Unsigned32',
+    'Unsigned64',
+] as const;
 export type NumericTypeId = (typeof NumericTypes)[number];
+
+export type NumericLimits = {
+    min: number | bigint;
+    max: number | bigint;
+    integer: boolean;
+    is_big: boolean;
+};
+
+export function getLimits(numeric_type: NumericTypeId): NumericLimits {
+    switch (numeric_type) {
+        case 'Float32':
+            return {
+                integer: false,
+                is_big: false,
+                min: -3.40282347e38,
+                max: 3.40282347e38,
+            };
+        case 'Float64':
+            return {
+                integer: false,
+                is_big: false,
+                min: -1.7976931348623157e308,
+                max: 1.7976931348623157e308,
+            };
+        case 'Int8':
+            return { integer: true, is_big: false, min: -128, max: 127 };
+        case 'Int16':
+            return { integer: true, is_big: false, min: -32768, max: 32767 };
+        case 'Int32':
+            return {
+                integer: true,
+                is_big: false,
+                min: -2147483648,
+                max: 2147483647,
+            };
+        case 'Int64':
+            return {
+                integer: true,
+                is_big: true,
+                min: -9223372036854775808n,
+                max: 9223372036854775807n,
+            };
+        case 'Unsigned8':
+            return { integer: true, is_big: false, min: 0, max: 255 };
+        case 'Unsigned16':
+            return { integer: true, is_big: false, min: 0, max: 65535 };
+        case 'Unsigned32':
+            return { integer: true, is_big: false, min: 0, max: 4294967295 };
+        case 'Unsigned64':
+            return {
+                integer: true,
+                is_big: true,
+                min: 0,
+                max: 18446744073709551615n,
+            };
+    }
+}
+
 export const BooleanTypes = ['Boolean'] as const;
 export type BooleanTypeId = (typeof BooleanTypes)[number];
 export const AnyTypes = [...NumericTypes, ...BooleanTypes] as const;
@@ -12,7 +81,7 @@ export function isBooleanType(type: AnyTypeId): type is BooleanTypeId {
     return BooleanTypes.some((it) => it == type);
 }
 
-export type DynamicDataType = number | boolean;
+export type DynamicDataType = bigint | number | boolean;
 
 export type CommandParameterData =
     | {

frontend/src/composables/json.ts

@@ -0,0 +1,26 @@
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export function toJsonString(data: any, pretty: boolean = false): string {
+    return JSON.stringify(
+        data,
+        (_key, value) => {
+            if (typeof value == 'bigint') {
+                // @ts-expect-error TS2339
+                return JSON.rawJSON(value.toString());
+            }
+            return value;
+        },
+        pretty ? 4 : undefined,
+    );
+}
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export function parseJsonString(data: string): any {
+    // @ts-expect-error TS2345, TS7006
+    return JSON.parse(data, (key, value, context) => {
+        if (key === 'Int64' || key == 'Unsigned64') {
+            // Or use the constructor of your custom high-precision number library
+            return BigInt(context.source);
+        }
+        return value;
+    });
+}

frontend/src/composables/telemetry.ts

@@ -1,6 +1,7 @@
 import { ref, toValue, watchEffect } from 'vue';
 import { type MaybeRefOrGetter } from 'vue';
 import type { AnyTypeId } from '@/composables/dynamic.ts';
+import { parseJsonString } from '@/composables/json.ts';
 
 export interface TelemetryDefinition {
     uuid: string;
@@ -16,8 +17,14 @@ export function useAllTelemetry() {
     watchEffect(async () => {
         try {
             const res = await fetch(`/api/tlm/info`);
-            data.value = await res.json();
-            error.value = null;
+            const text_response = await res.text();
+            if (res.ok) {
+                data.value = parseJsonString(text_response);
+                error.value = null;
+            } else {
+                data.value = null;
+                error.value = text_response;
+            }
         } catch (e) {
             data.value = null;
             error.value = e;
@@ -37,8 +44,14 @@ export function useTelemetry(name: MaybeRefOrGetter<string>) {
 
         try {
             const res = await fetch(`/api/tlm/info/${name_value}`);
-            data.value = await res.json();
-            error.value = null;
+            const text_response = await res.text();
+            if (res.ok) {
+                data.value = parseJsonString(text_response);
+                error.value = null;
+            } else {
+                data.value = null;
+                error.value = text_response;
+            }
         } catch (e) {
             data.value = null;
             error.value = e;

frontend/src/composables/websocket.ts

@@ -12,6 +12,7 @@ import {
 } from 'vue';
 import type { TelemetryDefinition } from '@/composables/telemetry';
 import { onDocumentVisibilityChange } from '@/composables/document.ts';
+import { parseJsonString, toJsonString } from '@/composables/json.ts';
 
 export interface TelemetryDataItem {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -71,7 +72,7 @@ export class WebsocketHandle {
             }
         });
         this.websocket.addEventListener('message', (event) => {
-            const message = JSON.parse(event.data);
+            const message = parseJsonString(event.data);
             if (message['TlmValue']) {
                 const tlm_value = message['TlmValue'] as TlmValue;
                 const listeners = this.on_telem_value.get(tlm_value.uuid);
@@ -126,7 +127,7 @@ export class WebsocketHandle {
             ([uuid_value, connected, enabled, min_sep, live_value]) => {
                 if (connected && enabled && uuid_value && live_value) {
                     this.websocket?.send(
-                        JSON.stringify({
+                        toJsonString({
                             RegisterTlmListener: {
                                 uuid: uuid_value,
                                 minimum_separation_ms: min_sep,
@@ -142,7 +143,7 @@ export class WebsocketHandle {
                     this.on_telem_value.get(uuid_value)?.push(callback_fn);
                     onWatcherCleanup(() => {
                         this.websocket?.send(
-                            JSON.stringify({
+                            toJsonString({
                                 UnregisterTlmListener: {
                                     uuid: uuid_value,
                                 },

frontend/src/panels/panel.ts

@@ -1,5 +1,6 @@
 import type { RouteLocationRaw } from 'vue-router';
 import { ref, type Ref, watchEffect } from 'vue';
+import { parseJsonString } from '@/composables/json.ts';
 
 export enum PanelHeirarchyType {
     LEAF = 'leaf',
@@ -61,7 +62,11 @@ export function usePanelHeirarchy(): Ref<PanelHeirarchyChildren> {
     watchEffect(async () => {
         try {
             const res = await fetch(`/api/panel`);
-            const data = await res.json();
+            if (!res.ok) {
+                console.error('Failed to fetch panels');
+                return;
+            }
+            const data = parseJsonString(await res.text());
 
             const server_panels: PanelHeirarchyFolder = {
                 name: 'Server Panels',

frontend/src/views/PanelEditorView.vue

@@ -1,7 +1,8 @@
 <script setup lang="ts">
 import DynamicComponent from '@/components/DynamicComponent.vue';
 import type { OptionalDynamicComponentData } from '@/composables/dynamic.ts';
-import { ref, watchEffect } from 'vue';
+import { ref, useTemplateRef, watchEffect } from 'vue';
+import { parseJsonString, toJsonString } from '@/composables/json.ts';
 
 const data = ref<OptionalDynamicComponentData>({ type: 'none' });
 
@@ -19,7 +20,11 @@ const unselect = () => {
 
 async function reload_panel_list() {
     const res = await fetch(`/api/panel`);
-    panel_list.value = await res.json();
+    if (!res.ok) {
+        console.error('Failed to reload panel list');
+        return;
+    }
+    panel_list.value = parseJsonString(await res.text());
 }
 watchEffect(reload_panel_list);
 
@@ -28,10 +33,9 @@ async function load(id: string) {
     load_screen.value = false;
     loading.value = true;
     const panel_data = await fetch(`/api/panel/${id}`);
-    const panel_json_value = await panel_data.json();
-    data.value = JSON.parse(
-        panel_json_value['data'],
-    ) as OptionalDynamicComponentData;
+    const panel_text_value = await panel_data.text();
+    const panel_json_value = parseJsonString(panel_text_value);
+    data.value = panel_json_value['data'] as OptionalDynamicComponentData;
     panel_name.value = panel_json_value['name'];
     panel_id.value = id;
     loading.value = false;
@@ -50,28 +54,33 @@ async function save() {
     if (panel_id_value) {
         const res = await fetch(`/api/panel/${panel_id_value}`, {
             method: 'PUT',
-            body: JSON.stringify({
+            body: toJsonString({
                 name: panel_name.value,
-                data: JSON.stringify(data.value),
+                data: data.value,
             }),
             headers: {
                 'Content-Type': 'application/json',
             },
         });
-        await res.json();
+        await res.text();
+        // TODO: Handle failures
     } else {
         const res = await fetch('/api/panel', {
             method: 'POST',
-            body: JSON.stringify({
+            body: toJsonString({
                 name: panel_name.value,
-                data: JSON.stringify(data.value),
+                data: data.value,
             }),
             headers: {
                 'Content-Type': 'application/json',
             },
         });
-        const uuid = await res.json();
-        panel_id.value = uuid as string;
+        const text_response = await res.text();
+        if (res.ok) {
+            const uuid = parseJsonString(text_response);
+            panel_id.value = uuid as string;
+        }
+        // TODO: Handle failures
     }
     loading.value = false;
 }
@@ -80,6 +89,35 @@ async function showLoadScreen() {
     load_screen.value = true;
     await reload_panel_list();
 }
+
+function download() {
+    const file = new Blob([toJsonString(data.value, true)], {
+        type: 'application/json',
+    });
+    const a = document.createElement('a');
+    const url = URL.createObjectURL(file);
+    a.href = url;
+    a.download = panel_name.value.toLowerCase().replace(' ', '_') + '.json';
+    document.body.appendChild(a);
+    a.click();
+    setTimeout(function () {
+        document.body.removeChild(a);
+        window.URL.revokeObjectURL(url);
+    }, 0);
+}
+
+const upload_input = useTemplateRef<HTMLInputElement>('upload-input');
+
+function upload() {
+    upload_input.value!.click();
+}
+
+async function onUpload() {
+    const input = upload_input.value!;
+    const file = input.files![0];
+    const file_data = await file.text();
+    data.value = parseJsonString(file_data);
+}
 </script>
 
 <template>
@@ -112,6 +150,15 @@ async function showLoadScreen() {
                     {{ panel_id ? 'Save' : 'New' }}
                 </button>
                 <button @click.prevent.stop="showLoadScreen">Load</button>
+                <button @click.prevent.stop="download">Download</button>
+                <input
+                    ref="upload-input"
+                    class="file"
+                    type="file"
+                    accept="application/json"
+                    @change="onUpload"
+                />
+                <button @click.prevent.stop="upload">Upload</button>
             </div>
             <div id="inspector" class="column"></div>
         </div>
@@ -126,4 +173,8 @@ async function showLoadScreen() {
     height: 100vh;
     background: variables.$dark-background-color;
 }
+
+input.file {
+    display: none;
+}
 </style>

frontend/src/views/PanelView.vue

@@ -6,6 +6,7 @@ import type {
 } from '@/composables/dynamic.ts';
 import { computed, provide, ref, watchEffect } from 'vue';
 import { useRoute } from 'vue-router';
+import { parseJsonString } from '@/composables/json.ts';
 
 const route = useRoute();
 
@@ -21,10 +22,10 @@ provide('inputs', inputs);
 
 watchEffect(async () => {
     const panel_data = await fetch(`/api/panel/${id.value}`);
-    const panel_json_value = await panel_data.json();
-    panel.value = JSON.parse(
-        panel_json_value['data'],
-    ) as OptionalDynamicComponentData;
+    const panel_text_value = await panel_data.text();
+    panel.value = parseJsonString(panel_text_value)[
+        'data'
+    ] as OptionalDynamicComponentData;
 });
 </script>
 

server/Cargo.toml

@@ -2,19 +2,20 @@
 [package]
 name = "server"
 edition = "2021"
-version = "0.1.0"
-authors = ["Sergey <me@sergeysav.com>"]
+version.workspace = true
+authors.workspace = true
 
 [dependencies]
 actix-web = { workspace = true, features = [ ] }
 actix-ws = { workspace = true }
 anyhow = { workspace = true }
-api = { path = "../api" }
+api = { workspace = true }
 chrono = { workspace = true }
 derive_more = { workspace = true, features = ["from"] }
-fern = { workspace = true }
+fern = { workspace = true, features = ["colored"] }
 futures-util = { workspace = true }
 log = { workspace = true }
+num-traits = { workspace = true }
 papaya = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }

server/src/command/service.rs

@@ -11,6 +11,7 @@ use api::messages::command::{Command, CommandDefinition, CommandHeader, CommandR
 use api::messages::ResponseMessage;
 use chrono::Utc;
 use log::error;
+use num_traits::FromPrimitive;
 use papaya::HashMap;
 use std::collections::HashMap as StdHashMap;
 use tokio::sync::oneshot;
@@ -102,10 +103,34 @@ 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 {
-                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),
+            let Some(Some(param_value)) = (match parameter.data_type {
+                DataType::Float32 => param_value
+                    .as_f64()
+                    .map(|v| Some(DataValue::Float32(v as f32))),
+                DataType::Float64 => param_value.as_f64().map(DataValue::Float64).map(Some),
+                DataType::Boolean => param_value.as_bool().map(DataValue::Boolean).map(Some),
+                DataType::Int8 => param_value
+                    .as_i64()
+                    .map(|v| Some(DataValue::Int8(i8::from_i64(v)?))),
+                DataType::Int16 => param_value
+                    .as_i64()
+                    .map(|v| Some(DataValue::Int16(i16::from_i64(v)?))),
+                DataType::Int32 => param_value
+                    .as_i64()
+                    .map(|v| Some(DataValue::Int32(i32::from_i64(v)?))),
+                DataType::Int64 => param_value.as_i64().map(|v| Some(DataValue::Int64(v))),
+                DataType::Unsigned8 => param_value
+                    .as_u64()
+                    .map(|v| Some(DataValue::Unsigned8(u8::from_u64(v)?))),
+                DataType::Unsigned16 => param_value
+                    .as_u64()
+                    .map(|v| Some(DataValue::Unsigned16(u16::from_u64(v)?))),
+                DataType::Unsigned32 => param_value
+                    .as_u64()
+                    .map(|v| Some(DataValue::Unsigned32(u32::from_u64(v)?))),
+                DataType::Unsigned64 => {
+                    param_value.as_u64().map(|v| Some(DataValue::Unsigned64(v)))
+                }
             }) else {
                 return Err(WrongParameterType {
                     name: parameter.name.clone(),

server/src/http/api/cmd.rs

@@ -3,7 +3,7 @@ use crate::http::error::HttpServerResultError;
 use actix_web::{get, post, web, Responder};
 use std::sync::Arc;
 
-#[post("/cmd/{name:[\\w\\d/_-]+}")]
+#[post("/cmd/{name:[\\.\\w\\d/_-]+}")]
 pub(super) async fn send_command(
     command_service: web::Data<Arc<CommandManagementService>>,
     name: web::Path<String>,
@@ -23,12 +23,16 @@ pub(super) async fn get_all(
     Ok(web::Json(command_service.get_commands()?))
 }
 
-#[get("/cmd/{name:[\\w\\d/_-]+}")]
+#[get("/cmd/{name:[\\.\\w\\d/_-]+}")]
 pub(super) async fn get_one(
     command_service: web::Data<Arc<CommandManagementService>>,
     name: web::Path<String>,
 ) -> Result<impl Responder, HttpServerResultError> {
     Ok(web::Json(
-        command_service.get_command_definition(&name.to_string()),
+        command_service
+            .get_command_definition(&name.to_string())
+            .ok_or_else(|| HttpServerResultError::CmdNotFound {
+                cmd: name.to_string(),
+            })?,
     ))
 }

server/src/http/api/panels.rs

@@ -9,7 +9,7 @@ use uuid::Uuid;
 #[derive(Deserialize)]
 struct CreateParam {
     name: String,
-    data: String,
+    data: serde_json::Value,
 }
 
 #[derive(Deserialize)]

server/src/http/api/tlm.rs

@@ -9,7 +9,7 @@ use std::time::Duration;
 use tokio::time::timeout;
 use uuid::Uuid;
 
-#[get("/tlm/info/{name:[\\w\\d/_-]+}")]
+#[get("/tlm/info/{name:[\\.\\w\\d/_-]+}")]
 pub(super) async fn get_tlm_definition(
     data: web::Data<Arc<TelemetryManagementService>>,
     name: web::Path<String>,

server/src/http/error.rs

@@ -23,6 +23,8 @@ pub enum HttpServerResultError {
     PanelUuidNotFound { uuid: Uuid },
     #[error(transparent)]
     Command(#[from] crate::command::error::Error),
+    #[error("Command Not Found: {cmd}")]
+    CmdNotFound { cmd: String },
 }
 
 impl ResponseError for HttpServerResultError {
@@ -36,6 +38,7 @@ impl ResponseError for HttpServerResultError {
             HttpServerResultError::InternalError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             HttpServerResultError::PanelUuidNotFound { .. } => StatusCode::NOT_FOUND,
             HttpServerResultError::Command(inner) => inner.status_code(),
+            HttpServerResultError::CmdNotFound { .. } => StatusCode::NOT_FOUND,
         }
     }
     fn error_response(&self) -> HttpResponse {

server/src/http/mod.rs

@@ -9,7 +9,7 @@ use crate::http::backend::setup_backend;
 use crate::http::websocket::setup_websocket;
 use crate::panels::PanelService;
 use crate::telemetry::management_service::TelemetryManagementService;
-use actix_web::middleware::Logger;
+use actix_web::middleware::{Compress, Logger};
 use actix_web::{web, App, HttpServer};
 use log::info;
 use std::sync::Arc;
@@ -36,6 +36,7 @@ pub async fn setup(
             .service(web::scope("/backend").configure(setup_backend))
             .service(web::scope("/ws").configure(setup_websocket))
             .service(web::scope("/api").configure(setup_api))
+            .wrap(Compress::default())
             .wrap(Logger::default())
     })
     .bind("localhost:8080")?

server/src/main.rs

@@ -1,3 +1,4 @@
+use fern::colors::{Color, ColoredLevelConfig};
 use std::env;
 use std::str::FromStr;
 
@@ -9,14 +10,18 @@ async fn main() -> anyhow::Result<()> {
         Err(_) => log::LevelFilter::Info,
     };
 
+    let colors = ColoredLevelConfig::new()
+        .info(Color::Green)
+        .debug(Color::Blue);
+
     let mut log_config = fern::Dispatch::new()
-        .format(|out, message, record| {
+        .format(move |out, message, record| {
             out.finish(format_args!(
-                "[{}][{}][{}] {}",
+                "[{} {} {}] {}",
                 chrono::Utc::now().format("%Y-%m-%d %H:%M:%S"),
+                colors.color(record.level()),
                 record.target(),
-                record.level(),
-                message
+                message,
             ))
         })
         .level(log::LevelFilter::Warn)
@@ -27,5 +32,6 @@ async fn main() -> anyhow::Result<()> {
         log_config = log_config.chain(fern::log_file(log_file)?)
     }
     log_config.apply()?;
+
     server::setup().await
 }

server/src/panels/mod.rs

@@ -14,10 +14,12 @@ impl PanelService {
         Self { pool }
     }
 
-    pub async fn create(&self, name: &str, data: &str) -> anyhow::Result<Uuid> {
+    pub async fn create(&self, name: &str, data: &serde_json::Value) -> anyhow::Result<Uuid> {
         let id = Uuid::new_v4();
         let id_string = id.to_string();
 
+        let data = serde_json::to_string(data)?;
+
         let mut transaction = self.pool.begin().await?;
 
         let _ = sqlx::query!(

server/src/panels/panel.rs

@@ -12,11 +12,12 @@ pub struct Panel {
     #[sqlx(flatten)]
     #[serde(flatten)]
     pub header: PanelRequired,
-    pub data: String,
+    #[sqlx(json)]
+    pub data: serde_json::Value,
 }
 
 #[derive(Default, Clone, Serialize, Deserialize)]
 pub struct PanelUpdate {
     pub name: Option<String>,
-    pub data: Option<String>,
+    pub data: Option<serde_json::Value>,
 }

server/src/telemetry/history.rs

@@ -196,9 +196,17 @@ impl HistorySegmentFile {
         // Write all the values
         for value in &data.values {
             match 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)?,
+                DataValue::Float32(value) => file.write_data(*value)?,
+                DataValue::Float64(value) => file.write_data(*value)?,
+                DataValue::Boolean(value) => file.write_data(*value)?,
+                DataValue::Int8(value) => file.write_data(*value)?,
+                DataValue::Int16(value) => file.write_data(*value)?,
+                DataValue::Int32(value) => file.write_data(*value)?,
+                DataValue::Int64(value) => file.write_data(*value)?,
+                DataValue::Unsigned8(value) => file.write_data(*value)?,
+                DataValue::Unsigned16(value) => file.write_data(*value)?,
+                DataValue::Unsigned32(value) => file.write_data(*value)?,
+                DataValue::Unsigned64(value) => file.write_data(*value)?,
             }
         }
 
@@ -332,20 +340,20 @@ impl HistorySegmentFile {
     }
 
     fn read_telemetry_item(&mut self, telemetry_data_type: DataType) -> anyhow::Result<DataValue> {
-        match telemetry_data_type {
-            DataType::Float32 => {
-                self.file_position += 4;
-                Ok(DataValue::Float32(self.file.read_data::<f32>()?))
-            }
-            DataType::Float64 => {
-                self.file_position += 8;
-                Ok(DataValue::Float64(self.file.read_data::<f64>()?))
-            }
-            DataType::Boolean => {
-                self.file_position += 1;
-                Ok(DataValue::Boolean(self.file.read_data::<bool>()?))
-            }
-        }
+        self.file_position += telemetry_data_type.get_data_length() as i64;
+        Ok(match telemetry_data_type {
+            DataType::Float32 => self.file.read_data::<f32>()?.into(),
+            DataType::Float64 => self.file.read_data::<f64>()?.into(),
+            DataType::Boolean => self.file.read_data::<bool>()?.into(),
+            DataType::Int8 => self.file.read_data::<i8>()?.into(),
+            DataType::Int16 => self.file.read_data::<i16>()?.into(),
+            DataType::Int32 => self.file.read_data::<i32>()?.into(),
+            DataType::Int64 => self.file.read_data::<i64>()?.into(),
+            DataType::Unsigned8 => self.file.read_data::<u8>()?.into(),
+            DataType::Unsigned16 => self.file.read_data::<u16>()?.into(),
+            DataType::Unsigned32 => self.file.read_data::<u32>()?.into(),
+            DataType::Unsigned64 => self.file.read_data::<u64>()?.into(),
+        })
     }
 
     fn get_telemetry_item(
@@ -353,11 +361,7 @@ impl HistorySegmentFile {
         index: u64,
         telemetry_data_type: DataType,
     ) -> anyhow::Result<DataValue> {
-        let item_length = match telemetry_data_type {
-            DataType::Float32 => 4,
-            DataType::Float64 => 8,
-            DataType::Boolean => 1,
-        };
+        let item_length = telemetry_data_type.get_data_length();
         let desired_position =
             Self::HEADER_LENGTH + self.length * Self::TIMESTAMP_LENGTH + index * item_length;
         let seek_amount = desired_position as i64 - self.file_position;

server/src/telemetry/management_service.rs

@@ -3,8 +3,6 @@ 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,
 };
@@ -144,11 +142,7 @@ impl TelemetryManagementService {
             bail!("Telemetry Item Not Found");
         };
 
-        let expected_type = match &tlm_item.value {
-            DataValue::Float32(_) => DataType::Float32,
-            DataValue::Float64(_) => DataType::Float64,
-            DataValue::Boolean(_) => DataType::Boolean,
-        };
+        let expected_type = tlm_item.value.to_data_type();
         if expected_type != tlm_data.data.definition.data_type {
             bail!("Data Type Mismatch");
         };