Compare commits

..

5 Commits

Author SHA1 Message Date
44862f65d2 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>
2026-01-03 18:24:40 -08:00
167e9d0a01 Implement Integral Data Types (#13)
**Rationale:**

Integral Types were missing and are needed for Project Nautilus.

**Changes:**

- Implements Integral Data Types
  - u64 and i64 implemented through bigint

Reviewed-on: #13
Co-authored-by: Sergey Savelyev <sergeysav.nn@gmail.com>
Co-committed-by: Sergey Savelyev <sergeysav.nn@gmail.com>
2026-01-03 08:37:26 -08:00
458c94c2ad Implement Fixes for Project Nautilus Integration (#12)
**Rationale:**

Some bugs were discovered in Project Nautilus integration.

**Changes:**

- Command and Telemetry paths now support `.`
- Telemetry Values now work with Boolean types in the frontend
- Constant Command Parameters no longer reset in the panel editor when you open the inspector

Reviewed-on: #12
Co-authored-by: Sergey Savelyev <sergeysav.nn@gmail.com>
Co-committed-by: Sergey Savelyev <sergeysav.nn@gmail.com>
2026-01-02 12:29:35 -08:00
09eeceb212 Add Generic Command Registration (#11)
**Rationale:**

Realized that #10 did not include any way to register commands without going through the IntoCommandDefinition trait.

**Changes:**

- Adds a method to register a raw command handler

Reviewed-on: #11
Co-authored-by: Sergey Savelyev <sergeysav.nn@gmail.com>
Co-committed-by: Sergey Savelyev <sergeysav.nn@gmail.com>
2026-01-01 13:11:23 -08:00
788dd10a91 Replace gRPC Backend (#10)
**Rationale:**

Having two separate servers and communication methods resulted in additional maintenance & the need to convert often between backend & frontend data types.
By moving the backend communication off of gRPC and to just use websockets it both gives more control & allows for simplification of the implementation.

#8

**Changes:**

- Replaces gRPC backend.
  - New implementation automatically handles reconnect logic
- Implements an api layer
- Migrates examples to the api layer
- Implements a proc macro to make command handling easier
- Implements unit tests for the api layer (90+% coverage)
- Implements integration tests for the proc macro (90+% coverage)

Reviewed-on: #10
Co-authored-by: Sergey Savelyev <sergeysav.nn@gmail.com>
Co-committed-by: Sergey Savelyev <sergeysav.nn@gmail.com>
2026-01-01 10:11:53 -08:00
68 changed files with 3514 additions and 1299 deletions

913
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,40 @@
[workspace] [workspace]
members = ["api", "server", "examples/simple_producer", "examples/simple_command"] members = ["api", "api-core", "api-proc-macro", "server", "examples/simple_producer", "examples/simple_command"]
resolver = "2" 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"
fern = "0.7.1"
futures-util = "0.3.31"
log = "0.4.29"
num-traits = "0.2.19"
papaya = "0.2.3"
proc-macro-error = "1.0.4"
quote = "1.0.42"
serde = { version = "1.0.228" }
serde_json = "1.0.148"
sqlx = "0.8.6"
syn = "2.0.112"
thiserror = "2.0.17"
tokio = { version = "1.48.0" }
tokio-stream = "0.1.17"
tokio-test = "0.4.4"
tokio-tungstenite = { version = "0.28.0" }
tokio-util = "0.7.17"
trybuild = "1.0.114"
uuid = { version = "1.19.0", features = ["v4"] }
[profile.dev.package.sqlx-macros] [profile.dev.package.sqlx-macros]
opt-level = 3 opt-level = 3

12
api-core/Cargo.toml Normal file
View File

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

60
api-core/src/command.rs Normal file
View File

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

60
api-core/src/data_type.rs Normal file
View File

@@ -0,0 +1,60 @@
use crate::data_value::DataValue;
use derive_more::Display;
use serde::{Deserialize, Serialize};
#[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> {
const DATA_TYPE: DataType;
}
macro_rules! impl_to_data_type {
( $ty:ty, $value:expr ) => {
impl ToDataType for $ty {
const DATA_TYPE: DataType = $value;
}
};
}
impl_to_data_type!(f32, DataType::Float32);
impl_to_data_type!(f64, DataType::Float64);
impl_to_data_type!(bool, DataType::Boolean);
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);

View File

@@ -0,0 +1,36 @@
use crate::data_type::DataType;
use derive_more::{From, TryInto};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, From, TryInto)]
pub enum DataValue {
Float32(f32),
Float64(f64),
Boolean(bool),
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,
}
}
}

3
api-core/src/lib.rs Normal file
View File

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

19
api-proc-macro/Cargo.toml Normal file
View File

@@ -0,0 +1,19 @@
[package]
name = "api-proc-macro"
edition = "2021"
version.workspace = true
authors.workspace = true
[lib]
proc-macro = true
[dependencies]
api-core = { workspace = true }
proc-macro-error = { workspace = true }
quote = { workspace = true }
syn = { workspace = true }
[dev-dependencies]
api = { workspace = true }
trybuild = { workspace = true }

View File

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

11
api-proc-macro/src/lib.rs Normal file
View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -2,18 +2,23 @@
[package] [package]
name = "api" name = "api"
edition = "2021" edition = "2021"
version = "0.1.0" version.workspace = true
authors = ["Sergey <me@sergeysav.com>"] authors.workspace = true
[dependencies] [dependencies]
log = "0.4.29" api-core = { workspace = true }
thiserror = "2.0.17" api-proc-macro = { workspace = true }
serde = { version = "1.0.228", features = ["derive"] } chrono = { workspace = true, features = ["serde"] }
derive_more = { version = "2.1.0", features = ["from", "try_into"] } derive_more = { workspace = true, features = ["from", "try_into"] }
uuid = { version = "1.19.0", features = ["v4", "serde"] } futures-util = { workspace = true }
chrono = { version = "0.4.39", features = ["serde"] } log = { workspace = true }
tokio = { version = "1.43.0", features = ["rt", "macros"] } serde = { workspace = true, features = ["derive"] }
tokio-tungstenite = { version = "0.28.0", features = ["rustls-tls-native-roots"] } serde_json = { workspace = true }
tokio-util = "0.7.17" thiserror = { workspace = true }
futures-util = "0.3.31" tokio = { workspace = true, features = ["rt", "macros", "time"] }
serde_json = "1.0.145" tokio-tungstenite = { workspace = true, features = ["rustls-tls-native-roots"] }
tokio-util = { workspace = true }
uuid = { workspace = true, features = ["serde"] }
[dev-dependencies]
env_logger = { workspace = true }

462
api/src/client/command.rs Normal file
View File

@@ -0,0 +1,462 @@
use crate::client::Client;
use crate::messages::command::CommandResponse;
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 {
client: Arc<Client>,
}
impl CommandRegistry {
pub fn new(client: Arc<Client>) -> Self {
Self { client }
}
pub fn register_raw_handler<F, E: Display>(
&self,
command_definition: CommandDefinition,
mut callback: F,
) -> CommandHandle
where
F: FnMut(Command) -> Result<String, E> + Send + 'static,
{
let cancellation_token = CancellationToken::new();
let result = CommandHandle {
cancellation_token: cancellation_token.clone(),
};
let client = self.client.clone();
tokio::spawn(async move {
while !cancellation_token.is_cancelled() {
// This would only fail if the sender closed while trying to insert data
// It would wait until space is made
let Ok(mut rx) = client
.register_callback_channel(command_definition.clone())
.await
else {
continue;
};
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;
}
},
_ = 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 {
cancellation_token: CancellationToken,
}
impl Drop for CommandHandle {
fn drop(&mut self) {
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");
};
}
}

11
api/src/client/config.rs Normal file
View File

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

594
api/src/client/context.rs Normal file
View File

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

View File

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

View File

@@ -1,5 +1,10 @@
pub mod command;
mod config;
mod context;
pub mod error; pub mod error;
pub mod telemetry;
use crate::client::config::ClientConfiguration;
use crate::client::error::{MessageError, RequestError}; use crate::client::error::{MessageError, RequestError};
use crate::messages::callback::GenericCallbackError; use crate::messages::callback::GenericCallbackError;
use crate::messages::payload::RequestMessagePayload; use crate::messages::payload::RequestMessagePayload;
@@ -7,33 +12,26 @@ use crate::messages::payload::ResponseMessagePayload;
use crate::messages::{ use crate::messages::{
ClientMessage, RegisterCallback, RequestMessage, RequestResponse, ResponseMessage, ClientMessage, RegisterCallback, RequestMessage, RequestResponse, ResponseMessage,
}; };
use context::ClientContext;
use error::ConnectError; use error::ConnectError;
use futures_util::stream::StreamExt;
use futures_util::SinkExt;
use log::{debug, error, warn};
use std::collections::HashMap;
use std::sync::mpsc::sync_channel;
use std::sync::Arc; use std::sync::Arc;
use std::thread; use tokio::spawn;
use tokio::net::TcpStream; use tokio::sync::{mpsc, oneshot, watch, RwLock};
use tokio::sync::{mpsc, oneshot, RwLock, RwLockWriteGuard};
use tokio::{select, spawn};
use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::handshake::client::Request;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use uuid::Uuid; use uuid::Uuid;
type RegisteredCallback = mpsc::Sender<(ResponseMessage, oneshot::Sender<RequestMessagePayload>)>; type RegisteredCallback = mpsc::Sender<(ResponseMessage, oneshot::Sender<RequestMessagePayload>)>;
type ClientChannel = Arc<RwLock<mpsc::Sender<OutgoingMessage>>>; type ClientChannel = Arc<RwLock<mpsc::Sender<OutgoingMessage>>>;
#[derive(Debug)]
enum Callback { enum Callback {
None, None,
Once(oneshot::Sender<ResponseMessage>), Once(oneshot::Sender<ResponseMessage>),
Registered(RegisteredCallback), Registered(RegisteredCallback),
} }
#[derive(Debug)]
struct OutgoingMessage { struct OutgoingMessage {
msg: RequestMessage, msg: RequestMessage,
callback: Callback, callback: Callback,
@@ -42,29 +40,42 @@ struct OutgoingMessage {
pub struct Client { pub struct Client {
cancel: CancellationToken, cancel: CancellationToken,
channel: ClientChannel, channel: ClientChannel,
} connected_state_rx: watch::Receiver<bool>,
struct ClientContext {
cancel: CancellationToken,
request: Request,
} }
impl Client { impl Client {
pub fn connect<R>(request: R) -> Result<Self, ConnectError> pub fn connect<R>(request: R) -> Result<Self, ConnectError>
where
R: IntoClientRequest,
{
Self::connect_with_config(request, ClientConfiguration::default())
}
pub fn connect_with_config<R>(
request: R,
config: ClientConfiguration,
) -> Result<Self, ConnectError>
where where
R: IntoClientRequest, R: IntoClientRequest,
{ {
let (tx, _rx) = mpsc::channel(1); let (tx, _rx) = mpsc::channel(1);
let cancel = CancellationToken::new(); let cancel = CancellationToken::new();
let channel = Arc::new(RwLock::new(tx)); let channel = Arc::new(RwLock::new(tx));
let (connected_state_tx, connected_state_rx) = watch::channel(false);
let context = ClientContext { let context = ClientContext {
cancel: cancel.clone(), cancel: cancel.clone(),
request: request.into_client_request()?, request: request.into_client_request()?,
connected_state_tx,
client_configuration: config,
}; };
context.start(channel.clone())?; context.start(channel.clone())?;
Ok(Self { cancel, channel }) Ok(Self {
cancel,
channel,
connected_state_rx,
})
} }
pub async fn send_message<M: ClientMessage>(&self, msg: M) -> Result<(), MessageError> { pub async fn send_message<M: ClientMessage>(&self, msg: M) -> Result<(), MessageError> {
@@ -226,211 +237,28 @@ impl Client {
Ok(()) Ok(())
} }
}
impl ClientContext { pub async fn wait_connected(&self) {
fn start(mut self, channel: ClientChannel) -> Result<(), ConnectError> { let mut connected_rx = self.connected_state_rx.clone();
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let (tx, rx) = sync_channel::<()>(1); // If we aren't currently connected
if !*connected_rx.borrow_and_update() {
let _detached = thread::Builder::new() // Wait for a change notification
.name("tlm-client".to_string()) // If the channel is closed there is nothing we can do
.spawn(move || { let _ = connected_rx.changed().await;
runtime.block_on(async {
let mut write_lock = channel.write().await;
// This cannot fail
let _ = tx.send(());
while !self.cancel.is_cancelled() {
write_lock = self.run_connection(write_lock, &channel).await;
}
drop(write_lock);
});
})?;
// This cannot fail
let _ = rx.recv();
Ok(())
}
async fn run_connection<'a>(
&mut self,
mut write_lock: RwLockWriteGuard<'a, mpsc::Sender<OutgoingMessage>>,
channel: &'a ClientChannel,
) -> RwLockWriteGuard<'a, mpsc::Sender<OutgoingMessage>> {
let mut ws = match connect_async(self.request.clone()).await {
Ok((ws, _)) => ws,
Err(e) => {
error!("Connect Error: {e}");
return write_lock;
}
};
let (tx, rx) = mpsc::channel(128);
*write_lock = tx;
drop(write_lock);
let close_connection = self.handle_connection(&mut ws, rx, channel).await;
let write_lock = channel.write().await;
if close_connection {
if let Err(e) = ws.close(None).await {
println!("Close Error {e}");
}
}
write_lock
}
async fn handle_connection(
&mut self,
ws: &mut WebSocketStream<MaybeTlsStream<TcpStream>>,
mut rx: mpsc::Receiver<OutgoingMessage>,
channel: &ClientChannel,
) -> bool {
let mut callbacks = HashMap::<Uuid, Callback>::new();
loop {
select! {
_ = self.cancel.cancelled() => { break; },
Some(msg) = ws.next() => {
match msg {
Ok(msg) => {
match msg {
Message::Text(msg) => {
let msg: ResponseMessage = match serde_json::from_str(&msg) {
Ok(m) => m,
Err(e) => {
error!("Failed to deserialize {e}");
break;
}
};
self.handle_incoming(msg, &mut callbacks, channel).await;
}
Message::Binary(_) => unimplemented!("Binary Data Not Implemented"),
Message::Ping(data) => {
if let Err(e) = ws.send(Message::Pong(data)).await {
error!("Failed to send Pong {e}");
break;
}
}
Message::Pong(_) => {
// Intentionally Left Empty
}
Message::Close(_) => {
debug!("Websocket Closed");
return false;
}
Message::Frame(_) => unreachable!("Not Possible"),
}
}
Err(e) => {
error!("Receive Error {e}");
break;
}
}
}
Some(msg) = rx.recv() => {
// Insert a callback if it isn't a None callback
if !matches!(msg.callback, Callback::None) {
callbacks.insert(msg.msg.uuid, msg.callback);
}
let msg = match serde_json::to_string(&msg.msg) {
Ok(m) => m,
Err(e) => {
error!("Encode Error {e}");
break;
}
};
if let Err(e) = ws.send(Message::Text(msg.into())).await {
error!("Send Error {e}");
break;
}
}
else => { break; },
}
}
true
}
async fn handle_incoming(
&mut self,
msg: ResponseMessage,
callbacks: &mut HashMap<Uuid, Callback>,
channel: &ClientChannel,
) {
if let Some(response_uuid) = msg.response {
match callbacks.get(&response_uuid) {
Some(Callback::None) => {
callbacks.remove(&response_uuid);
unreachable!("We skip registering callbacks of None type");
}
Some(Callback::Once(_)) => {
let Some(Callback::Once(callback)) = callbacks.remove(&response_uuid) else {
return;
};
let _ = callback.send(msg);
}
Some(Callback::Registered(callback)) => {
let callback = callback.clone();
spawn(Self::handle_registered_callback(
callback,
msg,
channel.clone(),
));
}
None => {
warn!("No Callback Registered for {response_uuid}");
}
}
} }
} }
async fn handle_registered_callback( pub async fn wait_disconnected(&self) {
callback: RegisteredCallback, let mut connected_rx = self.connected_state_rx.clone();
msg: ResponseMessage,
channel: ClientChannel,
) {
let (tx, rx) = oneshot::channel();
let uuid = msg.uuid; // If we are currently connected
if *connected_rx.borrow_and_update() {
let response = match callback.send((msg, tx)).await { // Wait for a change notification
Err(_) => GenericCallbackError::CallbackClosed.into(), // If the channel is closed there is nothing we can do
Ok(()) => rx let _ = connected_rx.changed().await;
.await
.unwrap_or_else(|_| GenericCallbackError::CallbackClosed.into()),
};
if let Err(e) = Self::send_response(channel, response, uuid).await {
error!("Failed to send response {e}");
} }
} }
async fn send_response(
channel: ClientChannel,
payload: RequestMessagePayload,
response_uuid: Uuid,
) -> Result<(), MessageError> {
// If this failed that means we're in the middle of reconnecting, so our callbacks
// are all being cleaned up as-is. No response needed.
let sender = channel.try_read()?;
let data = sender.reserve().await?;
data.send(OutgoingMessage {
msg: RequestMessage {
uuid: Uuid::new_v4(),
response: Some(response_uuid),
payload,
},
callback: Callback::None,
});
Ok(())
}
} }
impl Drop for Client { impl Drop for Client {
@@ -438,3 +266,332 @@ impl Drop for Client {
self.cancel.cancel(); 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(),
}
}
}

461
api/src/client/telemetry.rs Normal file
View File

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

View File

@@ -1,8 +0,0 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum DataType {
Float32,
Float64,
Boolean,
}

View File

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

View File

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

View File

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

View File

@@ -1,34 +1,14 @@
use crate::data_type::DataType;
use crate::data_value::DataValue;
use crate::messages::RegisterCallback; use crate::messages::RegisterCallback;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::HashMap;
#[derive(Debug, Clone, Serialize, Deserialize)] pub use api_core::command::*;
pub struct CommandParameterDefinition {
pub name: String,
pub data_type: DataType,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommandDefinition {
pub name: String,
pub parameters: Vec<CommandParameterDefinition>,
}
impl RegisterCallback for CommandDefinition { impl RegisterCallback for CommandDefinition {
type Callback = Command; type Callback = Command;
type Response = CommandResponse; type Response = CommandResponse;
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Command {
pub timestamp: DateTime<Utc>,
pub parameters: HashMap<String, DataValue>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommandResponse { pub struct CommandResponse {
pub success: bool, pub success: bool,
pub response: String, pub response: String,

View File

@@ -11,14 +11,18 @@ use uuid::Uuid;
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RequestMessage { pub struct RequestMessage {
pub uuid: Uuid, pub uuid: Uuid,
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
pub response: Option<Uuid>, pub response: Option<Uuid>,
#[serde(flatten)] #[serde(flatten)]
pub payload: RequestMessagePayload, pub payload: RequestMessagePayload,
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ResponseMessage { pub struct ResponseMessage {
pub uuid: Uuid, pub uuid: Uuid,
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
pub response: Option<Uuid>, pub response: Option<Uuid>,
#[serde(flatten)] #[serde(flatten)]
pub payload: ResponseMessagePayload, pub payload: ResponseMessagePayload,

View File

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

View File

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

View File

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

View File

@@ -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>>
}
})),
},
)
}

1
api/src/test/mod.rs Normal file
View File

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

View File

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

View File

@@ -1,40 +1,16 @@
use anyhow::anyhow; use api::client::command::CommandRegistry;
use api::client::Client; use api::client::Client;
use api::data_type::DataType; use api::macros::IntoCommandDefinition;
use api::messages::command::{ use api::messages::command::CommandHeader;
Command, CommandDefinition, CommandParameterDefinition, CommandResponse, use log::info;
};
use std::error::Error; use std::error::Error;
use std::sync::Arc; use std::sync::Arc;
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
fn handle_command(command: Command) -> anyhow::Result<String> {
let timestamp = command.timestamp;
let a: f32 = (*command
.parameters
.get("a")
.ok_or(anyhow!("Parameter 'a' Missing"))?)
.try_into()?;
let b: f64 = (*command
.parameters
.get("b")
.ok_or(anyhow!("Parameter 'b' Missing"))?)
.try_into()?;
let c: bool = (*command
.parameters
.get("c")
.ok_or(anyhow!("Parameter 'c' Missing"))?)
.try_into()?;
println!("Command Received:\n timestamp: {timestamp}\n a: {a}\n b: {b}\n c: {c}");
Ok(format!(
"Successfully Received Command! timestamp: {timestamp} a: {a} b: {b} c: {c}"
))
}
#[tokio::main] #[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> { async fn main() -> Result<(), Box<dyn Error>> {
env_logger::init();
let cancellation_token = CancellationToken::new(); let cancellation_token = CancellationToken::new();
{ {
let cancellation_token = cancellation_token.clone(); let cancellation_token = cancellation_token.clone();
@@ -44,41 +20,34 @@ async fn main() -> Result<(), Box<dyn Error>> {
}); });
} }
let client = Arc::new(Client::connect("ws://[::1]:8080/backend")?); let client = Arc::new(Client::connect("ws://localhost:8080/backend")?);
let cmd = CommandRegistry::new(client);
client let handle = cmd.register_handler("simple_command/a", handle_command);
.register_callback_fn(
CommandDefinition {
name: "simple_command/a".to_string(),
parameters: vec![
CommandParameterDefinition {
name: "a".to_string(),
data_type: DataType::Float32,
},
CommandParameterDefinition {
name: "b".to_string(),
data_type: DataType::Float64,
},
CommandParameterDefinition {
name: "c".to_string(),
data_type: DataType::Boolean,
},
],
},
|command| match handle_command(command) {
Ok(response) => CommandResponse {
success: true,
response,
},
Err(error) => CommandResponse {
success: false,
response: error.to_string(),
},
},
)
.await?;
cancellation_token.cancelled().await; cancellation_token.cancelled().await;
// This will automatically drop when we return
drop(handle);
Ok(()) Ok(())
} }
#[derive(IntoCommandDefinition)]
struct SimpleCommandA {
a: f32,
b: f64,
c: bool,
}
fn handle_command(header: CommandHeader, command: SimpleCommandA) -> anyhow::Result<String> {
let timestamp = header.timestamp;
let SimpleCommandA { a, b, c } = command;
info!("Command Received:\n timestamp: {timestamp}\n a: {a}\n b: {b}\n c: {c}");
// This gets sent back to the source of the command
Ok(format!(
"Successfully Received Command! timestamp: {timestamp} a: {a} b: {b} c: {c}"
))
}

View File

@@ -4,14 +4,11 @@ name = "simple_producer"
edition = "2021" edition = "2021"
[dependencies] [dependencies]
api = { path = "../../api" } anyhow = { workspace = true }
tokio = { version = "1.43.0", features = ["rt-multi-thread", "signal", "time", "macros"] } api = { workspace = true }
chrono = "0.4.39" chrono = { workspace = true }
num-traits = "0.2.19" env_logger = { workspace = true }
tokio-tungstenite = { version = "0.28.0", features = ["rustls-tls-native-roots"] } futures-util = { workspace = true }
anyhow = "1.0.100" num-traits = { workspace = true }
futures-util = "0.3.31" tokio = { workspace = true, features = ["rt-multi-thread", "signal", "time", "macros"] }
serde_json = "1.0.145" tokio-util = { workspace = true }
uuid = { version = "1.19.0", features = ["v4"] }
tokio-util = "0.7.17"
tokio-stream = "0.1.17"

View File

@@ -1,129 +1,44 @@
use api::client::error::MessageError; use api::client::telemetry::TelemetryRegistry;
use api::client::Client; use api::client::Client;
use api::data_type::DataType; use chrono::{TimeDelta, Utc};
use api::data_value::DataValue;
use api::messages::telemetry_definition::TelemetryDefinitionRequest;
use api::messages::telemetry_entry::TelemetryEntry;
use chrono::{DateTime, TimeDelta, Utc};
use futures_util::future::join_all; use futures_util::future::join_all;
use num_traits::FloatConst; use num_traits::FloatConst;
use std::f64;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::time::{sleep_until, Instant}; use tokio::time::{sleep_until, Instant};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use uuid::Uuid;
struct Telemetry {
client: Arc<Client>,
}
struct TelemetryItemHandle {
uuid: Uuid,
client: Arc<Client>,
}
impl Telemetry {
pub fn new(client: Arc<Client>) -> Self {
Self { client }
}
pub async fn register(
&self,
name: String,
data_type: DataType,
) -> anyhow::Result<TelemetryItemHandle> {
let response = self
.client
.send_request(TelemetryDefinitionRequest { name, data_type })
.await?;
Ok(TelemetryItemHandle {
uuid: response.uuid,
client: self.client.clone(),
})
}
}
impl TelemetryItemHandle {
pub async fn publish(&self, value: DataValue, timestamp: DateTime<Utc>) -> anyhow::Result<()> {
self.client
.send_message_if_connected(TelemetryEntry {
uuid: self.uuid,
value,
timestamp,
})
.await
.or_else(|e| match e {
MessageError::TokioLockError(_) => Ok(()),
e => Err(e),
})?;
Ok(())
}
}
#[tokio::main] #[tokio::main]
async fn main() -> anyhow::Result<()> { async fn main() -> anyhow::Result<()> {
let client = Arc::new(Client::connect("ws://[::1]:8080/backend")?); env_logger::init();
let tlm = Telemetry::new(client);
{ let client = Arc::new(Client::connect("ws://localhost:8080/backend")?);
let time_offset = tlm let tlm = TelemetryRegistry::new(client);
.register("simple_producer/time_offset".into(), DataType::Float64)
.await?;
let publish_offset = tlm let time_offset = tlm.register::<f64>("simple_producer/time_offset").await;
.register("simple_producer/publish_offset".into(), DataType::Float64)
.await?;
let await_offset = tlm let publish_offset = tlm.register::<f64>("simple_producer/publish_offset").await;
.register("simple_producer/await_offset".into(), DataType::Float64)
.await?;
let sin_tlm_handle = tlm let await_offset = tlm.register::<f64>("simple_producer/await_offset").await;
.register("simple_producer/sin".into(), DataType::Float32)
.await?;
let cos_tlm_handle = tlm
.register("simple_producer/cos".into(), DataType::Float64)
.await?;
let bool_tlm_handle = tlm
.register("simple_producer/bool".into(), DataType::Boolean)
.await?;
let sin2_tlm_handle = tlm let bool_tlm_handle = tlm.register::<bool>("simple_producer/bool").await;
.register("simple_producer/sin2".into(), DataType::Float32)
.await?;
let cos2_tlm_handle = tlm
.register("simple_producer/cos2".into(), DataType::Float64)
.await?;
let sin3_tlm_handle = tlm let sin_handles = join_all((1..=6).map(|i| {
.register("simple_producer/sin3".into(), DataType::Float32) tlm.register::<f32>(format!(
.await?; "simple_producer/sin{}",
let cos3_tlm_handle = tlm if i == 1 { "".into() } else { i.to_string() }
.register("simple_producer/cos3".into(), DataType::Float64) ))
.await?; }))
.await;
let sin4_tlm_handle = tlm let cos_handles = join_all((1..=6).map(|i| {
.register("simple_producer/sin4".into(), DataType::Float32) tlm.register::<f64>(format!(
.await?; "simple_producer/cos{}",
let cos4_tlm_handle = tlm if i == 1 { "".into() } else { i.to_string() }
.register("simple_producer/cos4".into(), DataType::Float64) ))
.await?; }))
.await;
let sin5_tlm_handle = tlm
.register("simple_producer/sin5".into(), DataType::Float32)
.await?;
let cos5_tlm_handle = tlm
.register("simple_producer/cos5".into(), DataType::Float64)
.await?;
let sin6_tlm_handle = tlm
.register("simple_producer/sin6".into(), DataType::Float32)
.await?;
let cos6_tlm_handle = tlm
.register("simple_producer/cos6".into(), DataType::Float64)
.await?;
let cancellation_token = CancellationToken::new(); let cancellation_token = CancellationToken::new();
{ {
@@ -131,7 +46,6 @@ async fn main() -> anyhow::Result<()> {
tokio::spawn(async move { tokio::spawn(async move {
let _ = tokio::signal::ctrl_c().await; let _ = tokio::signal::ctrl_c().await;
cancellation_token.cancel(); cancellation_token.cancel();
println!("Cancellation Token Cancelled");
}); });
} }
@@ -139,87 +53,45 @@ async fn main() -> anyhow::Result<()> {
let start_instant = Instant::now(); let start_instant = Instant::now();
let mut next_time = start_instant; let mut next_time = start_instant;
let mut index = 0; let mut index = 0;
let mut tasks = vec![];
while !cancellation_token.is_cancelled() { while !cancellation_token.is_cancelled() {
next_time += Duration::from_millis(10); next_time += Duration::from_millis(10);
index += 1; index += 1;
sleep_until(next_time).await; sleep_until(next_time).await;
let publish_time = start_time + TimeDelta::from_std(next_time - start_instant).unwrap(); let publish_time = start_time + TimeDelta::from_std(next_time - start_instant).unwrap();
let actual_time = Instant::now(); let actual_time = Instant::now();
tasks.push(time_offset.publish(
DataValue::Float64((actual_time - next_time).as_secs_f64()),
Utc::now(),
));
tasks.push(sin_tlm_handle.publish(
DataValue::Float32((f32::TAU() * (index as f32) / (1000.0_f32)).sin()),
publish_time,
));
tasks.push(cos_tlm_handle.publish(
DataValue::Float64((f64::TAU() * (index as f64) / (1000.0_f64)).cos()),
publish_time,
));
tasks.push(
bool_tlm_handle.publish(DataValue::Boolean(index % 1000 > 500), publish_time),
);
tasks.push(sin2_tlm_handle.publish(
DataValue::Float32((f32::TAU() * (index as f32) / (500.0_f32)).sin()),
publish_time,
));
tasks.push(cos2_tlm_handle.publish(
DataValue::Float64((f64::TAU() * (index as f64) / (500.0_f64)).cos()),
publish_time,
));
tasks.push(sin3_tlm_handle.publish(
DataValue::Float32((f32::TAU() * (index as f32) / (333.0_f32)).sin()),
publish_time,
));
tasks.push(cos3_tlm_handle.publish(
DataValue::Float64((f64::TAU() * (index as f64) / (333.0_f64)).cos()),
publish_time,
));
tasks.push(sin4_tlm_handle.publish(
DataValue::Float32((f32::TAU() * (index as f32) / (250.0_f32)).sin()),
publish_time,
));
tasks.push(cos4_tlm_handle.publish(
DataValue::Float64((f64::TAU() * (index as f64) / (250.0_f64)).cos()),
publish_time,
));
tasks.push(sin5_tlm_handle.publish(
DataValue::Float32((f32::TAU() * (index as f32) / (200.0_f32)).sin()),
publish_time,
));
tasks.push(cos5_tlm_handle.publish(
DataValue::Float64((f64::TAU() * (index as f64) / (200.0_f64)).cos()),
publish_time,
));
tasks.push(sin6_tlm_handle.publish(
DataValue::Float32((f32::TAU() * (index as f32) / (166.0_f32)).sin()),
publish_time,
));
tasks.push(cos6_tlm_handle.publish(
DataValue::Float64((f64::TAU() * (index as f64) / (166.0_f64)).cos()),
publish_time,
));
tasks.push(publish_offset.publish( // Due to how telemetry handles are implemented, unless the send buffer is full awaiting
DataValue::Float64((Instant::now() - actual_time).as_secs_f64()), // these will return immediately
Utc::now(), time_offset
)); .publish_now((actual_time - next_time).as_secs_f64())
.await?;
bool_tlm_handle
.publish(index % 1000 > 500, publish_time)
.await?;
// Join the tasks so they all run in parallel for (i, sin) in sin_handles.iter().enumerate() {
for task in join_all(tasks.drain(..)).await { sin.publish(
task?; (f32::TAU() * (index as f32) / (1000.0_f32 / (i + 1) as f32)).sin(),
publish_time,
)
.await?;
}
for (i, cos) in cos_handles.iter().enumerate() {
cos.publish(
(f64::TAU() * (index as f64) / (1000.0_f64 / (i + 1) as f64)).cos(),
publish_time,
)
.await?;
} }
tasks.push(await_offset.publish( publish_offset
DataValue::Float64((Instant::now() - actual_time).as_secs_f64()), .publish((Instant::now() - actual_time).as_secs_f64(), Utc::now())
Utc::now(), .await?;
));
await_offset
.publish((Instant::now() - actual_time).as_secs_f64(), Utc::now())
.await?;
} }
println!("Exiting Loop");
}
drop(tlm);
Ok(()) Ok(())
} }

View File

@@ -3,8 +3,10 @@ import { computed, onMounted } from 'vue';
import { import {
type AnyTypeId, type AnyTypeId,
type DynamicDataType, type DynamicDataType,
getLimits,
isBooleanType, isBooleanType,
isNumericType, isNumericType,
type NumericLimits,
} from '@/composables/dynamic.ts'; } from '@/composables/dynamic.ts';
const props = defineProps<{ const props = defineProps<{
@@ -17,23 +19,100 @@ const is_numeric = computed(() => {
return isNumericType(props.type); 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(() => { const is_boolean = computed(() => {
return isBooleanType(props.type); return isBooleanType(props.type);
}); });
// Initialize the parameter to some value: // Initialize the parameter to some value:
onMounted(() => { onMounted(() => {
if (model.value === undefined) {
if (is_numeric.value) { 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; model.value = 0.0;
}
} else if (is_boolean.value) { } else if (is_boolean.value) {
model.value = false; 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> </script>
<template> <template>
<input v-if="is_numeric" type="number" v-model="model" /> <input
<input v-else-if="is_boolean" type="checkbox" v-model="model" /> 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> <span v-else>UNKNOWN INPUT</span>
</template> </template>

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -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 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 const BooleanTypes = ['Boolean'] as const;
export type BooleanTypeId = (typeof BooleanTypes)[number]; export type BooleanTypeId = (typeof BooleanTypes)[number];
export const AnyTypes = [...NumericTypes, ...BooleanTypes] as const; 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); return BooleanTypes.some((it) => it == type);
} }
export type DynamicDataType = number | boolean; export type DynamicDataType = bigint | number | boolean;
export type CommandParameterData = export type CommandParameterData =
| { | {

View File

@@ -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;
});
}

View File

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

View File

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

View File

@@ -1,5 +1,6 @@
import type { RouteLocationRaw } from 'vue-router'; import type { RouteLocationRaw } from 'vue-router';
import { ref, type Ref, watchEffect } from 'vue'; import { ref, type Ref, watchEffect } from 'vue';
import { parseJsonString } from '@/composables/json.ts';
export enum PanelHeirarchyType { export enum PanelHeirarchyType {
LEAF = 'leaf', LEAF = 'leaf',
@@ -61,7 +62,11 @@ export function usePanelHeirarchy(): Ref<PanelHeirarchyChildren> {
watchEffect(async () => { watchEffect(async () => {
try { try {
const res = await fetch(`/api/panel`); 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 = { const server_panels: PanelHeirarchyFolder = {
name: 'Server Panels', name: 'Server Panels',

View File

@@ -1,7 +1,8 @@
<script setup lang="ts"> <script setup lang="ts">
import DynamicComponent from '@/components/DynamicComponent.vue'; import DynamicComponent from '@/components/DynamicComponent.vue';
import type { OptionalDynamicComponentData } from '@/composables/dynamic.ts'; 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' }); const data = ref<OptionalDynamicComponentData>({ type: 'none' });
@@ -19,7 +20,11 @@ const unselect = () => {
async function reload_panel_list() { async function reload_panel_list() {
const res = await fetch(`/api/panel`); 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); watchEffect(reload_panel_list);
@@ -28,10 +33,9 @@ async function load(id: string) {
load_screen.value = false; load_screen.value = false;
loading.value = true; loading.value = true;
const panel_data = await fetch(`/api/panel/${id}`); const panel_data = await fetch(`/api/panel/${id}`);
const panel_json_value = await panel_data.json(); const panel_text_value = await panel_data.text();
data.value = JSON.parse( const panel_json_value = parseJsonString(panel_text_value);
panel_json_value['data'], data.value = panel_json_value['data'] as OptionalDynamicComponentData;
) as OptionalDynamicComponentData;
panel_name.value = panel_json_value['name']; panel_name.value = panel_json_value['name'];
panel_id.value = id; panel_id.value = id;
loading.value = false; loading.value = false;
@@ -50,29 +54,34 @@ async function save() {
if (panel_id_value) { if (panel_id_value) {
const res = await fetch(`/api/panel/${panel_id_value}`, { const res = await fetch(`/api/panel/${panel_id_value}`, {
method: 'PUT', method: 'PUT',
body: JSON.stringify({ body: toJsonString({
name: panel_name.value, name: panel_name.value,
data: JSON.stringify(data.value), data: data.value,
}), }),
headers: { headers: {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
}, },
}); });
await res.json(); await res.text();
// TODO: Handle failures
} else { } else {
const res = await fetch('/api/panel', { const res = await fetch('/api/panel', {
method: 'POST', method: 'POST',
body: JSON.stringify({ body: toJsonString({
name: panel_name.value, name: panel_name.value,
data: JSON.stringify(data.value), data: data.value,
}), }),
headers: { headers: {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
}, },
}); });
const uuid = await res.json(); const text_response = await res.text();
if (res.ok) {
const uuid = parseJsonString(text_response);
panel_id.value = uuid as string; panel_id.value = uuid as string;
} }
// TODO: Handle failures
}
loading.value = false; loading.value = false;
} }
@@ -80,6 +89,35 @@ async function showLoadScreen() {
load_screen.value = true; load_screen.value = true;
await reload_panel_list(); 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> </script>
<template> <template>
@@ -112,6 +150,15 @@ async function showLoadScreen() {
{{ panel_id ? 'Save' : 'New' }} {{ panel_id ? 'Save' : 'New' }}
</button> </button>
<button @click.prevent.stop="showLoadScreen">Load</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>
<div id="inspector" class="column"></div> <div id="inspector" class="column"></div>
</div> </div>
@@ -126,4 +173,8 @@ async function showLoadScreen() {
height: 100vh; height: 100vh;
background: variables.$dark-background-color; background: variables.$dark-background-color;
} }
input.file {
display: none;
}
</style> </style>

View File

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

View File

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

View File

@@ -7,10 +7,11 @@ use crate::command::error::Error::{
use anyhow::bail; use anyhow::bail;
use api::data_type::DataType; use api::data_type::DataType;
use api::data_value::DataValue; use api::data_value::DataValue;
use api::messages::command::{Command, CommandDefinition, CommandResponse}; use api::messages::command::{Command, CommandDefinition, CommandHeader, CommandResponse};
use api::messages::ResponseMessage; use api::messages::ResponseMessage;
use chrono::Utc; use chrono::Utc;
use log::error; use log::error;
use num_traits::FromPrimitive;
use papaya::HashMap; use papaya::HashMap;
use std::collections::HashMap as StdHashMap; use std::collections::HashMap as StdHashMap;
use tokio::sync::oneshot; use tokio::sync::oneshot;
@@ -102,10 +103,34 @@ impl CommandManagementService {
let Some(param_value) = parameters.get(&parameter.name) else { let Some(param_value) = parameters.get(&parameter.name) else {
return Err(MisingParameter(parameter.name.clone())); return Err(MisingParameter(parameter.name.clone()));
}; };
let Some(param_value) = (match parameter.data_type { let Some(Some(param_value)) = (match parameter.data_type {
DataType::Float32 => param_value.as_f64().map(|v| DataValue::Float32(v as f32)), DataType::Float32 => param_value
DataType::Float64 => param_value.as_f64().map(DataValue::Float64), .as_f64()
DataType::Boolean => param_value.as_bool().map(DataValue::Boolean), .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 { }) else {
return Err(WrongParameterType { return Err(WrongParameterType {
name: parameter.name.clone(), name: parameter.name.clone(),
@@ -137,7 +162,7 @@ impl CommandManagementService {
uuid, uuid,
response: Some(response_uuid), response: Some(response_uuid),
payload: Command { payload: Command {
timestamp, header: CommandHeader { timestamp },
parameters: result_parameters, parameters: result_parameters,
} }
.into(), .into(),

View File

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

View File

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

View File

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

View File

@@ -1,3 +1,4 @@
use futures_util::stream::StreamExt;
mod connection; mod connection;
use crate::command::service::CommandManagementService; use crate::command::service::CommandManagementService;
@@ -8,7 +9,6 @@ use log::{error, trace};
use std::sync::Arc; use std::sync::Arc;
use tokio::select; use tokio::select;
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use tonic::codegen::tokio_stream::StreamExt;
async fn backend_connect( async fn backend_connect(
req: HttpRequest, req: HttpRequest,

View File

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

View File

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

View File

@@ -6,6 +6,7 @@ use crate::telemetry::management_service::TelemetryManagementService;
use actix_web::{rt, web, HttpRequest, HttpResponse}; use actix_web::{rt, web, HttpRequest, HttpResponse};
use actix_ws::{AggregatedMessage, ProtocolError, Session}; use actix_ws::{AggregatedMessage, ProtocolError, Session};
use anyhow::anyhow; use anyhow::anyhow;
use futures_util::StreamExt;
use log::{error, trace}; use log::{error, trace};
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
@@ -14,7 +15,6 @@ use tokio::select;
use tokio::sync::mpsc::Sender; use tokio::sync::mpsc::Sender;
use tokio::time::{sleep_until, Instant}; use tokio::time::{sleep_until, Instant};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use tonic::codegen::tokio_stream::StreamExt;
use uuid::Uuid; use uuid::Uuid;
pub mod request; pub mod request;

View File

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

View File

@@ -14,10 +14,12 @@ impl PanelService {
Self { pool } 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 = Uuid::new_v4();
let id_string = id.to_string(); let id_string = id.to_string();
let data = serde_json::to_string(data)?;
let mut transaction = self.pool.begin().await?; let mut transaction = self.pool.begin().await?;
let _ = sqlx::query!( let _ = sqlx::query!(

View File

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

View File

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

View File

@@ -3,8 +3,6 @@ use crate::telemetry::data_item::TelemetryDataItem;
use crate::telemetry::definition::TelemetryDefinition; use crate::telemetry::definition::TelemetryDefinition;
use crate::telemetry::history::{TelemetryHistory, TelemetryHistoryService}; use crate::telemetry::history::{TelemetryHistory, TelemetryHistoryService};
use anyhow::bail; use anyhow::bail;
use api::data_type::DataType;
use api::data_value::DataValue;
use api::messages::telemetry_definition::{ use api::messages::telemetry_definition::{
TelemetryDefinitionRequest, TelemetryDefinitionResponse, TelemetryDefinitionRequest, TelemetryDefinitionResponse,
}; };
@@ -144,11 +142,7 @@ impl TelemetryManagementService {
bail!("Telemetry Item Not Found"); bail!("Telemetry Item Not Found");
}; };
let expected_type = match &tlm_item.value { let expected_type = tlm_item.value.to_data_type();
DataValue::Float32(_) => DataType::Float32,
DataValue::Float64(_) => DataType::Float64,
DataValue::Boolean(_) => DataType::Boolean,
};
if expected_type != tlm_data.data.definition.data_type { if expected_type != tlm_data.data.definition.data_type {
bail!("Data Type Mismatch"); bail!("Data Type Mismatch");
}; };