From 8d737e8f3310aed8e4e137a403fd57298ccd1e49 Mon Sep 17 00:00:00 2001 From: Sergey Savelyev Date: Mon, 29 Dec 2025 19:08:13 -0500 Subject: [PATCH] updates telemetry to no longer use grpc --- Cargo.lock | 203 +++++- Cargo.toml | 2 +- api/Cargo.toml | 14 + api/src/data_type.rs | 8 + .../src/telemetry => api/src}/data_value.rs | 4 +- api/src/lib.rs | 4 + api/src/request.rs | 32 + api/src/response.rs | 20 + examples/simple_producer/Cargo.toml | 13 +- examples/simple_producer/src/main.rs | 606 +++++++++++------- server/Cargo.toml | 1 + server/proto/core.proto | 23 - server/src/grpc/cmd.rs | 13 +- server/src/grpc/mod.rs | 11 - server/src/grpc/tlm.rs | 141 ---- server/src/http/api/tlm.rs | 9 +- server/src/http/backend/connection.rs | 81 +++ server/src/http/backend/mod.rs | 57 ++ server/src/http/error.rs | 6 +- server/src/http/mod.rs | 3 + server/src/http/websocket/mod.rs | 15 +- server/src/http/websocket/request.rs | 5 +- server/src/http/websocket/response.rs | 3 +- server/src/lib.rs | 2 +- server/src/telemetry/data_item.rs | 4 +- server/src/telemetry/definition.rs | 11 +- server/src/telemetry/history.rs | 62 +- server/src/telemetry/management_service.rs | 71 +- server/src/telemetry/mod.rs | 1 - 29 files changed, 911 insertions(+), 514 deletions(-) create mode 100644 api/Cargo.toml create mode 100644 api/src/data_type.rs rename {server/src/telemetry => api/src}/data_value.rs (54%) create mode 100644 api/src/lib.rs create mode 100644 api/src/request.rs create mode 100644 api/src/response.rs delete mode 100644 server/src/grpc/tlm.rs create mode 100644 server/src/http/backend/connection.rs create mode 100644 server/src/http/backend/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 8f99aab..986edb0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -251,6 +251,18 @@ version = "1.0.100" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" +[[package]] +name = "api" +version = "0.1.0" +dependencies = [ + "chrono", + "derive_more", + "log", + "serde", + "thiserror", + "uuid", +] + [[package]] name = "async-stream" version = "0.3.6" @@ -417,9 +429,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" [[package]] name = "bytes" -version = "1.7.2" +version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "428d9aa8fbc0670b7b8d6030a7fadd0f86151cae55e4dbbece15f3780a3dfaf3" +checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3" [[package]] name = "bytestring" @@ -456,6 +468,7 @@ dependencies = [ "iana-time-zone", "js-sys", "num-traits", + "serde", "wasm-bindgen", "windows-link", ] @@ -505,6 +518,16 @@ dependencies = [ "libc", ] +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -569,6 +592,12 @@ dependencies = [ "typenum", ] +[[package]] +name = "data-encoding" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476" + [[package]] name = "der" version = "0.7.10" @@ -819,6 +848,17 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" +[[package]] +name = "futures-macro" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "futures-sink" version = "0.3.31" @@ -839,6 +879,7 @@ checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" dependencies = [ "futures-core", "futures-io", + "futures-macro", "futures-sink", "futures-task", "memchr", @@ -1443,10 +1484,10 @@ dependencies = [ "libc", "log", "openssl", - "openssl-probe", + "openssl-probe 0.1.6", "openssl-sys", "schannel", - "security-framework", + "security-framework 2.11.1", "security-framework-sys", "tempfile", ] @@ -1541,6 +1582,12 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" +[[package]] +name = "openssl-probe" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f50d9b3dabb09ecd771ad0aa242ca6894994c130308ca3d7684634df8037391" + [[package]] name = "openssl-sys" version = "0.9.111" @@ -1892,6 +1939,20 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" +[[package]] +name = "ring" +version = "0.17.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e75ec5e92c4d8aede845126adc388046234541629e76029599ed35a003c7ed24" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.15", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rsa" version = "0.9.9" @@ -1934,6 +1995,51 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rustls" +version = "0.23.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "533f54bc6a7d4f647e46ad909549eda97bf5afc1585190ef692b4286b198bd8f" +dependencies = [ + "once_cell", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-native-certs" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63" +dependencies = [ + "openssl-probe 0.2.0", + "rustls-pki-types", + "schannel", + "security-framework 3.5.1", +] + +[[package]] +name = "rustls-pki-types" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21e6f2ab2928ca4291b86736a8bd920a277a399bba1589409d72154ff87c1282" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ffdfa2f5286e2247234e03f680868ac2815974dc39e00ea15adc445d0aafe52" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.18" @@ -1968,7 +2074,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" dependencies = [ "bitflags", - "core-foundation", + "core-foundation 0.9.4", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework" +version = "3.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3297343eaf830f66ede390ea39da1d462b6b0c1b000f420d0a83f898bbbe6ef" +dependencies = [ + "bitflags", + "core-foundation 0.10.1", "core-foundation-sys", "libc", "security-framework-sys", @@ -2062,6 +2181,7 @@ dependencies = [ "actix-web", "actix-ws", "anyhow", + "api", "chrono", "derive_more", "fern", @@ -2146,12 +2266,17 @@ dependencies = [ name = "simple_producer" version = "0.0.0" dependencies = [ + "anyhow", + "api", "chrono", + "futures-util", "num-traits", - "server", + "serde_json", "tokio", + "tokio-stream", + "tokio-tungstenite", "tokio-util", - "tonic", + "uuid", ] [[package]] @@ -2575,16 +2700,42 @@ dependencies = [ ] [[package]] -name = "tokio-stream" -version = "0.1.16" +name = "tokio-rustls" +version = "0.26.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f4e6ce100d0eb49a2734f8c0812bcd324cf357d21810932c5df6b96ef2b86f1" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tokio-stream" +version = "0.1.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047" dependencies = [ "futures-core", "pin-project-lite", "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.17" @@ -2726,6 +2877,25 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +dependencies = [ + "bytes", + "data-encoding", + "http 1.1.0", + "httparse", + "log", + "rand 0.9.0", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror", + "utf-8", +] + [[package]] name = "typenum" version = "1.17.0" @@ -2771,6 +2941,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + [[package]] name = "url" version = "2.5.7" @@ -2783,6 +2959,12 @@ dependencies = [ "serde", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -2797,6 +2979,7 @@ checksum = "e2e054861b4bd027cd373e18e8d8d8e6548085000e41290d95ce0c373a654b4a" dependencies = [ "getrandom 0.3.1", "js-sys", + "serde_core", "wasm-bindgen", ] diff --git a/Cargo.toml b/Cargo.toml index 5567a9c..243b5fb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["server", "examples/simple_producer", "examples/simple_command"] +members = ["api", "server", "examples/simple_producer", "examples/simple_command"] resolver = "2" [profile.dev.package.sqlx-macros] diff --git a/api/Cargo.toml b/api/Cargo.toml new file mode 100644 index 0000000..000a12d --- /dev/null +++ b/api/Cargo.toml @@ -0,0 +1,14 @@ + +[package] +name = "api" +edition = "2021" +version = "0.1.0" +authors = ["Sergey "] + +[dependencies] +log = "0.4.29" +thiserror = "2.0.17" +serde = { version = "1.0.228", features = ["derive"] } +derive_more = { version = "2.1.0", features = ["from", "try_into"] } +uuid = { version = "1.19.0", features = ["v4", "serde"] } +chrono = { version = "0.4.39", features = ["serde"] } diff --git a/api/src/data_type.rs b/api/src/data_type.rs new file mode 100644 index 0000000..a9292f3 --- /dev/null +++ b/api/src/data_type.rs @@ -0,0 +1,8 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum DataType { + Float32, + Float64, + Boolean, +} diff --git a/server/src/telemetry/data_value.rs b/api/src/data_value.rs similarity index 54% rename from server/src/telemetry/data_value.rs rename to api/src/data_value.rs index 328f2f8..43e8d9e 100644 --- a/server/src/telemetry/data_value.rs +++ b/api/src/data_value.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; -#[derive(Debug, Clone, Serialize, Deserialize)] -pub enum TelemetryDataValue { +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +pub enum DataValue { Float32(f32), Float64(f64), Boolean(bool), diff --git a/api/src/lib.rs b/api/src/lib.rs new file mode 100644 index 0000000..d45cf7a --- /dev/null +++ b/api/src/lib.rs @@ -0,0 +1,4 @@ +pub mod data_type; +pub mod data_value; +pub mod request; +pub mod response; diff --git a/api/src/request.rs b/api/src/request.rs new file mode 100644 index 0000000..4653a09 --- /dev/null +++ b/api/src/request.rs @@ -0,0 +1,32 @@ +use crate::data_type::DataType; +use crate::data_value::DataValue; +use chrono::{DateTime, Utc}; +use derive_more::From; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelemetryDefinitionRequest { + pub name: String, + pub data_type: DataType, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelemetryEntry { + pub uuid: Uuid, + pub value: DataValue, + pub timestamp: DateTime, +} + +#[derive(Debug, Clone, Serialize, Deserialize, From)] +pub enum RequestMessagePayload { + TelemetryDefinitionRequest(TelemetryDefinitionRequest), + TelemetryEntry(TelemetryEntry), +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RequestMessage { + pub uuid: Uuid, + #[serde(flatten)] + pub payload: RequestMessagePayload, +} diff --git a/api/src/response.rs b/api/src/response.rs new file mode 100644 index 0000000..5c7a41f --- /dev/null +++ b/api/src/response.rs @@ -0,0 +1,20 @@ +use derive_more::{From, TryInto}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelemetryDefinitionResponse { + pub uuid: Uuid, +} + +#[derive(Debug, Clone, Serialize, Deserialize, From, TryInto)] +pub enum ResponseMessagePayload { + TelemetryDefinitionResponse(TelemetryDefinitionResponse), +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResponseMessage { + pub uuid: Uuid, + #[serde(flatten)] + pub payload: ResponseMessagePayload, +} diff --git a/examples/simple_producer/Cargo.toml b/examples/simple_producer/Cargo.toml index e22533c..3a9a9bd 100644 --- a/examples/simple_producer/Cargo.toml +++ b/examples/simple_producer/Cargo.toml @@ -4,9 +4,14 @@ name = "simple_producer" edition = "2021" [dependencies] -server = { path = "../../server" } -tonic = "0.12.3" -tokio = { version = "1.43.0", features = ["rt-multi-thread", "signal"] } +api = { path = "../../api" } +tokio = { version = "1.43.0", features = ["rt-multi-thread", "signal", "time", "macros"] } chrono = "0.4.39" -tokio-util = "0.7.13" num-traits = "0.2.19" +tokio-tungstenite = { version = "0.28.0", features = ["rustls-tls-native-roots"] } +anyhow = "1.0.100" +futures-util = "0.3.31" +serde_json = "1.0.145" +uuid = { version = "1.19.0", features = ["v4"] } +tokio-util = "0.7.17" +tokio-stream = "0.1.17" diff --git a/examples/simple_producer/src/main.rs b/examples/simple_producer/src/main.rs index 8aaecb7..bec85d6 100644 --- a/examples/simple_producer/src/main.rs +++ b/examples/simple_producer/src/main.rs @@ -1,110 +1,247 @@ -use chrono::DateTime; -use num_traits::float::FloatConst; -use server::core::telemetry_service_client::TelemetryServiceClient; -use server::core::telemetry_value::Value; -use server::core::{ - TelemetryDataType, TelemetryDefinitionRequest, TelemetryItem, TelemetryValue, Timestamp, Uuid, +use anyhow::anyhow; +use api::data_type::DataType; +use api::data_value::DataValue; +use api::request::{ + RequestMessage, RequestMessagePayload, TelemetryDefinitionRequest, TelemetryEntry, }; -use std::error::Error; +use api::response::{ResponseMessage, ResponseMessagePayload, TelemetryDefinitionResponse}; +use chrono::{DateTime, TimeDelta, Utc}; +use futures_util::future::join_all; +use futures_util::{SinkExt, StreamExt}; +use num_traits::FloatConst; +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use std::time::Duration; -use tokio::select; +use tokio::net::TcpStream; use tokio::sync::mpsc; -use tokio::sync::mpsc::Sender; -use tokio::time::Instant; +use tokio::sync::mpsc::channel; +use tokio::time::{sleep, sleep_until, Instant}; +use tokio::{pin, select}; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; use tokio_util::sync::CancellationToken; -use tonic::codegen::tokio_stream::wrappers::ReceiverStream; -use tonic::codegen::tokio_stream::StreamExt; -use tonic::codegen::StdError; -use tonic::transport::Channel; -use tonic::Request; +use uuid::Uuid; struct Telemetry { - client: TelemetryServiceClient, - tx: Sender, + is_connected: Arc, + tx: mpsc::Sender, + callback_new: mpsc::Sender<(Uuid, mpsc::Sender)>, + callback_delete: mpsc::Sender, cancel: CancellationToken, } -struct TelemetryItemHandle { - uuid: String, - tx: Sender, +struct TelemetryItemHandle<'a> { + uuid: Uuid, + tlm: &'a Telemetry, } impl Telemetry { - pub async fn new(dst: D) -> Result> + async fn connect(request: R) -> anyhow::Result>> where - D: TryInto, - D::Error: Into, + R: IntoClientRequest + Unpin, + { + let (client, _) = match connect_async(request).await { + Ok(o) => o, + Err(e) => { + sleep(Duration::from_secs(1)).await; + return Err(e.into()); + } + }; + Ok(client) + } + + pub async fn new(request: R) -> anyhow::Result + where + R: IntoClientRequest + Unpin + Clone + Send + 'static, { - let mut client = TelemetryServiceClient::connect(dst).await?; - let client_stored = client.clone(); let cancel = CancellationToken::new(); let cancel_stored = cancel.clone(); - let (local_tx, mut local_rx) = mpsc::channel(16); + let (tx_tx, mut tx_rx) = channel(128); + let (callback_new_tx, mut callback_new_rx) = channel(16); + let (callback_delete_tx, mut callback_delete_rx) = channel(16); + + let is_connected = Arc::new(AtomicBool::new(false)); + let connected = is_connected.clone(); tokio::spawn(async move { while !cancel.is_cancelled() { - let (server_tx, server_rx) = mpsc::channel(16); - let response_stream = client - .insert_telemetry(ReceiverStream::new(server_rx)) - .await; - if let Ok(response_stream) = response_stream { - let mut response_stream = response_stream.into_inner(); - loop { - select! { - _ = cancel.cancelled() => { - break; - }, - Some(item) = local_rx.recv() => { - match server_tx.send(item).await { - Ok(_) => {} - Err(_) => break, - } - }, - Some(response) = response_stream.next() => { - match response { - Ok(_) => {} - Err(_) => { - break; + let client = Self::connect(request.clone()); + pin!(client); + let mut client = match loop { + break select! { + c = &mut client => { + c + }, + Some(_) = tx_rx.recv() => {continue;}, + Some(_) = callback_new_rx.recv() => {continue;}, + Some(_) = callback_delete_rx.recv() => {continue;}, + }; + } { + Ok(c) => c, + Err(e) => { + println!("Connect Error: {e}"); + continue; + } + }; + + let mut close_connection = true; + let mut callbacks = HashMap::>::new(); + connected.store(true, Ordering::SeqCst); + + loop { + select! { + biased; + _ = cancel.cancelled() => { break; }, + Some(msg) = callback_new_rx.recv() => { + let (uuid, callback) = msg; + callbacks.insert(uuid, callback); + }, + Some(msg) = callback_delete_rx.recv() => { + callbacks.remove(&msg); + }, + Some(msg) = client.next() => { + match msg { + Ok(msg) => { + match msg { + Message::Text(msg) => { + let msg: ResponseMessage = match serde_json::from_str(&msg) { + Ok(m) => m, + Err(e) => { + println!("Failed to deserialize {e}"); + break; + } + }; + if let Some(cb) = callbacks.get_mut(&msg.uuid) { + if let Err(e) = cb.send(msg.payload).await { + println!("Failed to call callback {e}"); + callbacks.remove(&msg.uuid); + } + } else { + unimplemented!("Unexpected Message: {msg:?}"); + } + } + Message::Binary(_) => unimplemented!("Binary Unsupported"), + Message::Ping(data) => { + if let Err(e) = client.send(Message::Pong(data)).await { + println!("Failed to send Pong {e}"); + break; + } + } + Message::Pong(_) => { + // Intentionally Left Empty + } + Message::Close(_) => { + println!("Websocket Closed"); + close_connection = false; + break; + } + Message::Frame(_) => unreachable!("Not Possible"), } } - }, - else => break, - } + Err(e) => { + println!("Receive Error {e}"); + break; + } + } + }, + Some(msg) = tx_rx.recv() => { + let msg = match serde_json::to_string(&msg) { + Ok(m) => m, + Err(e) => { + println!("Encode Error {e}"); + break; + } + }; + if let Err(e) = client.send(Message::Text(msg.into())).await { + println!("Send Error {e}"); + break; + } + }, + else => { break; }, } - } else { - tokio::time::sleep(Duration::from_secs(1)).await; } + connected.store(false, Ordering::SeqCst); + if close_connection { + if let Err(e) = client.close(None).await { + println!("Close Error {e}"); + } + } + + // Callbacks gets dropped here - closing all listeners automatically } }); Ok(Self { - client: client_stored, - tx: local_tx, + tx: tx_tx, + callback_new: callback_new_tx, + callback_delete: callback_delete_tx, cancel: cancel_stored, + is_connected, }) } - pub async fn register( - &mut self, - name: String, - data_type: TelemetryDataType, - ) -> Result> { - let response = self - .client - .new_telemetry(Request::new(TelemetryDefinitionRequest { - name, - data_type: data_type.into(), - })) - .await? - .into_inner(); + pub async fn wait_connected(&self) { + while !self.is_connected.load(Ordering::Relaxed) { + sleep(Duration::from_millis(10)).await; + } + } - let Some(uuid) = response.uuid else { - return Err("UUID Missing".into()); - }; + /// Send a message and don't expect a response + pub async fn send_message>( + &self, + payload: P, + ) -> anyhow::Result<()> { + self.tx + .send(RequestMessage { + uuid: Uuid::new_v4(), + payload: payload.into(), + }) + .await?; + Ok(()) + } + + async fn send_request< + P: Into, + R: TryFrom, + >( + &self, + payload: P, + ) -> anyhow::Result { + let uuid = Uuid::new_v4(); + let (tx, mut rx) = channel(1); + + self.wait_connected().await; + self.callback_new.send((uuid, tx)).await?; + self.tx + .send(RequestMessage { + uuid, + payload: payload.into(), + }) + .await?; + + let response = rx.recv().await.ok_or(anyhow!("No Response Received"))?; + + let response = R::try_from(response)?; + + self.callback_delete.send(uuid).await?; + + Ok(response) + } + + pub async fn register( + &self, + name: String, + data_type: DataType, + ) -> anyhow::Result> { + let response: TelemetryDefinitionResponse = self + .send_request(TelemetryDefinitionRequest { name, data_type }) + .await?; Ok(TelemetryItemHandle { - uuid: uuid.value, - tx: self.tx.clone(), + uuid: response.uuid, + tlm: self, }) } } @@ -115,24 +252,13 @@ impl Drop for Telemetry { } } -impl TelemetryItemHandle { - pub async fn publish( - &self, - value: Value, - timestamp: DateTime, - ) -> Result<(), Box> { - let offset_from_unix_epoch = - timestamp - DateTime::from_timestamp(0, 0).expect("Could not get Unix epoch"); - self.tx - .send(TelemetryItem { - uuid: Some(Uuid { - value: self.uuid.clone(), - }), - value: Some(TelemetryValue { value: Some(value) }), - timestamp: Some(Timestamp { - secs: offset_from_unix_epoch.num_seconds(), - nanos: offset_from_unix_epoch.subsec_nanos(), - }), +impl<'a> TelemetryItemHandle<'a> { + pub async fn publish(&self, value: DataValue, timestamp: DateTime) -> anyhow::Result<()> { + self.tlm + .send_message(TelemetryEntry { + uuid: self.uuid, + value, + timestamp, }) .await?; Ok(()) @@ -140,164 +266,162 @@ impl TelemetryItemHandle { } #[tokio::main] -async fn main() -> Result<(), Box> { - let mut tlm = Telemetry::new("http://[::1]:50051").await?; +async fn main() -> anyhow::Result<()> { + let tlm = Telemetry::new("ws://[::1]:8080/backend").await?; - let index_handle = tlm - .register( - "simple_producer/time_offset".into(), - TelemetryDataType::Float64, - ) - .await?; - - let publish_offset = tlm - .register( - "simple_producer/publish_offset".into(), - TelemetryDataType::Float64, - ) - .await?; - - let await_offset = tlm - .register( - "simple_producer/await_offset".into(), - TelemetryDataType::Float64, - ) - .await?; - - let sin_tlm_handle = tlm - .register("simple_producer/sin".into(), TelemetryDataType::Float32) - .await?; - let cos_tlm_handle = tlm - .register("simple_producer/cos".into(), TelemetryDataType::Float64) - .await?; - let bool_tlm_handle = tlm - .register("simple_producer/bool".into(), TelemetryDataType::Boolean) - .await?; - - let sin2_tlm_handle = tlm - .register("simple_producer/sin2".into(), TelemetryDataType::Float32) - .await?; - let cos2_tlm_handle = tlm - .register("simple_producer/cos2".into(), TelemetryDataType::Float64) - .await?; - - let sin3_tlm_handle = tlm - .register("simple_producer/sin3".into(), TelemetryDataType::Float32) - .await?; - let cos3_tlm_handle = tlm - .register("simple_producer/cos3".into(), TelemetryDataType::Float64) - .await?; - - let sin4_tlm_handle = tlm - .register("simple_producer/sin4".into(), TelemetryDataType::Float32) - .await?; - let cos4_tlm_handle = tlm - .register("simple_producer/cos4".into(), TelemetryDataType::Float64) - .await?; - - let sin5_tlm_handle = tlm - .register("simple_producer/sin5".into(), TelemetryDataType::Float32) - .await?; - let cos5_tlm_handle = tlm - .register("simple_producer/cos5".into(), TelemetryDataType::Float64) - .await?; - - let sin6_tlm_handle = tlm - .register("simple_producer/sin6".into(), TelemetryDataType::Float32) - .await?; - let cos6_tlm_handle = tlm - .register("simple_producer/cos6".into(), TelemetryDataType::Float64) - .await?; - - let cancellation_token = CancellationToken::new(); { - let cancellation_token = cancellation_token.clone(); - tokio::spawn(async move { - let _ = tokio::signal::ctrl_c().await; - cancellation_token.cancel(); - }); - } + let time_offset = tlm + .register("simple_producer/time_offset".into(), DataType::Float64) + .await?; - let start_time = chrono::Utc::now(); - let start_instant = Instant::now(); - let mut next_time = start_instant; - let mut index = 0; - let mut tasks = vec![]; - while !cancellation_token.is_cancelled() { - next_time += Duration::from_millis(10); - index += 1; - tokio::time::sleep_until(next_time).await; - let publish_time = - start_time + chrono::TimeDelta::from_std(next_time - start_instant).unwrap(); - let actual_time = Instant::now(); - tasks.push(index_handle.publish( - Value::Float64((actual_time - next_time).as_secs_f64()), - chrono::Utc::now(), - )); - tasks.push(sin_tlm_handle.publish( - Value::Float32((f32::TAU() * (index as f32) / (1000.0_f32)).sin()), - publish_time, - )); - tasks.push(cos_tlm_handle.publish( - Value::Float64((f64::TAU() * (index as f64) / (1000.0_f64)).cos()), - publish_time, - )); - tasks.push(bool_tlm_handle.publish(Value::Boolean(index % 1000 > 500), publish_time)); - tasks.push(sin2_tlm_handle.publish( - Value::Float32((f32::TAU() * (index as f32) / (500.0_f32)).sin()), - publish_time, - )); - tasks.push(cos2_tlm_handle.publish( - Value::Float64((f64::TAU() * (index as f64) / (500.0_f64)).cos()), - publish_time, - )); - tasks.push(sin3_tlm_handle.publish( - Value::Float32((f32::TAU() * (index as f32) / (333.0_f32)).sin()), - publish_time, - )); - tasks.push(cos3_tlm_handle.publish( - Value::Float64((f64::TAU() * (index as f64) / (333.0_f64)).cos()), - publish_time, - )); - tasks.push(sin4_tlm_handle.publish( - Value::Float32((f32::TAU() * (index as f32) / (250.0_f32)).sin()), - publish_time, - )); - tasks.push(cos4_tlm_handle.publish( - Value::Float64((f64::TAU() * (index as f64) / (250.0_f64)).cos()), - publish_time, - )); - tasks.push(sin5_tlm_handle.publish( - Value::Float32((f32::TAU() * (index as f32) / (200.0_f32)).sin()), - publish_time, - )); - tasks.push(cos5_tlm_handle.publish( - Value::Float64((f64::TAU() * (index as f64) / (200.0_f64)).cos()), - publish_time, - )); - tasks.push(sin6_tlm_handle.publish( - Value::Float32((f32::TAU() * (index as f32) / (166.0_f32)).sin()), - publish_time, - )); - tasks.push(cos6_tlm_handle.publish( - Value::Float64((f64::TAU() * (index as f64) / (166.0_f64)).cos()), - publish_time, - )); + let publish_offset = tlm + .register("simple_producer/publish_offset".into(), DataType::Float64) + .await?; - tasks.push(publish_offset.publish( - Value::Float64((Instant::now() - actual_time).as_secs_f64()), - chrono::Utc::now(), - )); + let await_offset = tlm + .register("simple_producer/await_offset".into(), DataType::Float64) + .await?; - for task in tasks.drain(..) { - task.await?; + let sin_tlm_handle = tlm + .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 + .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 + .register("simple_producer/sin3".into(), DataType::Float32) + .await?; + let cos3_tlm_handle = tlm + .register("simple_producer/cos3".into(), DataType::Float64) + .await?; + + let sin4_tlm_handle = tlm + .register("simple_producer/sin4".into(), DataType::Float32) + .await?; + let cos4_tlm_handle = tlm + .register("simple_producer/cos4".into(), DataType::Float64) + .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 = cancellation_token.clone(); + tokio::spawn(async move { + let _ = tokio::signal::ctrl_c().await; + cancellation_token.cancel(); + println!("Cancellation Token Cancelled"); + }); } - tasks.push(await_offset.publish( - Value::Float64((Instant::now() - actual_time).as_secs_f64()), - chrono::Utc::now(), - )); + let start_time = Utc::now(); + let start_instant = Instant::now(); + let mut next_time = start_instant; + let mut index = 0; + let mut tasks = vec![]; + while !cancellation_token.is_cancelled() { + next_time += Duration::from_millis(10); + index += 1; + sleep_until(next_time).await; + let publish_time = start_time + TimeDelta::from_std(next_time - start_instant).unwrap(); + let actual_time = Instant::now(); + tasks.push(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( + DataValue::Float64((Instant::now() - actual_time).as_secs_f64()), + Utc::now(), + )); + + // Join the tasks so they all run in parallel + for task in join_all(tasks.drain(..)).await { + task?; + } + + tasks.push(await_offset.publish( + DataValue::Float64((Instant::now() - actual_time).as_secs_f64()), + Utc::now(), + )); + } + println!("Exiting Loop"); } + drop(tlm); Ok(()) } diff --git a/server/Cargo.toml b/server/Cargo.toml index 3151585..2e91c62 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -25,6 +25,7 @@ derive_more = { version = "2.1.0", features = ["from"] } anyhow = "1.0.100" sqlx = { version = "0.8.6", features = [ "runtime-tokio", "tls-native-tls", "sqlite" ] } uuid = { version = "1.19.0", features = ["v4"] } +api = { path = "../api" } [build-dependencies] tonic-build = "0.12.3" diff --git a/server/proto/core.proto b/server/proto/core.proto index 18d0ea9..961ae34 100644 --- a/server/proto/core.proto +++ b/server/proto/core.proto @@ -26,29 +26,6 @@ message Timestamp { sfixed32 nanos = 2; } -message TelemetryDefinitionRequest { - string name = 1; - TelemetryDataType data_type = 2; -} - -message TelemetryDefinitionResponse { - UUID uuid = 1; -} - -message TelemetryItem { - UUID uuid = 1; - TelemetryValue value = 2; - Timestamp timestamp = 3; -} - -message TelemetryInsertResponse { -} - -service TelemetryService { - rpc NewTelemetry (TelemetryDefinitionRequest) returns (TelemetryDefinitionResponse); - rpc InsertTelemetry (stream TelemetryItem) returns (stream TelemetryInsertResponse); -} - message CommandParameterDefinition { string name = 1; TelemetryDataType data_type = 2; diff --git a/server/src/grpc/cmd.rs b/server/src/grpc/cmd.rs index 900201e..711b3ec 100644 --- a/server/src/grpc/cmd.rs +++ b/server/src/grpc/cmd.rs @@ -135,11 +135,14 @@ impl CommandService for CoreCommandService { } } for (key, sender) in in_progress.drain() { - if sender.send(CommandResponse { - uuid: Some(Uuid::from(key)), - success: false, - response: "Command Handler Shut Down".to_string(), - }).is_err() { + if sender + .send(CommandResponse { + uuid: Some(Uuid::from(key)), + success: false, + response: "Command Handler Shut Down".to_string(), + }) + .is_err() + { error!("Failed to send command response on shutdown"); } } diff --git a/server/src/grpc/mod.rs b/server/src/grpc/mod.rs index 1e7b01b..1e5eaf5 100644 --- a/server/src/grpc/mod.rs +++ b/server/src/grpc/mod.rs @@ -1,12 +1,8 @@ mod cmd; -mod tlm; use crate::command::service::CommandManagementService; use crate::core::command_service_server::CommandServiceServer; -use crate::core::telemetry_service_server::TelemetryServiceServer; use crate::grpc::cmd::CoreCommandService; -use crate::grpc::tlm::CoreTelemetryService; -use crate::telemetry::management_service::TelemetryManagementService; use log::{error, info}; use std::sync::Arc; use tokio::task::JoinHandle; @@ -15,16 +11,10 @@ use tonic::transport::Server; pub fn setup( token: CancellationToken, - telemetry_management_service: Arc, command_service: Arc, ) -> anyhow::Result> { let addr = "[::1]:50051".parse()?; Ok(tokio::spawn(async move { - let tlm_service = CoreTelemetryService { - tlm_management: telemetry_management_service, - cancellation_token: token.clone(), - }; - let cmd_service = CoreCommandService { command_service, cancellation_token: token.clone(), @@ -32,7 +22,6 @@ pub fn setup( info!("Starting gRPC Server"); let result = Server::builder() - .add_service(TelemetryServiceServer::new(tlm_service)) .add_service(CommandServiceServer::new(cmd_service)) .serve_with_shutdown(addr, token.cancelled_owned()) .await; diff --git a/server/src/grpc/tlm.rs b/server/src/grpc/tlm.rs deleted file mode 100644 index fff260d..0000000 --- a/server/src/grpc/tlm.rs +++ /dev/null @@ -1,141 +0,0 @@ -use crate::core::telemetry_service_server::TelemetryService; -use crate::core::telemetry_value::Value; -use crate::core::{ - TelemetryDataType, TelemetryDefinitionRequest, TelemetryDefinitionResponse, - TelemetryInsertResponse, TelemetryItem, TelemetryValue, Uuid, -}; -use crate::telemetry::data_item::TelemetryDataItem; -use crate::telemetry::data_value::TelemetryDataValue; -use crate::telemetry::history::TelemetryHistory; -use crate::telemetry::management_service::TelemetryManagementService; -use chrono::{DateTime, SecondsFormat}; -use log::trace; -use std::pin::Pin; -use std::sync::Arc; -use tokio::select; -use tokio::sync::mpsc; -use tokio_util::sync::CancellationToken; -use tonic::codegen::tokio_stream::wrappers::ReceiverStream; -use tonic::codegen::tokio_stream::{Stream, StreamExt}; -use tonic::{Request, Response, Status, Streaming}; - -pub struct CoreTelemetryService { - pub tlm_management: Arc, - pub cancellation_token: CancellationToken, -} - -#[tonic::async_trait] -impl TelemetryService for CoreTelemetryService { - async fn new_telemetry( - &self, - request: Request, - ) -> Result, Status> { - trace!("CoreTelemetryService::new_telemetry"); - self.tlm_management - .register(request.into_inner()) - .map(|uuid| { - Response::new(TelemetryDefinitionResponse { - uuid: Some(Uuid { value: uuid }), - }) - }) - .map_err(|err| Status::already_exists(err.to_string())) - } - - type InsertTelemetryStream = - Pin> + Send>>; - - async fn insert_telemetry( - &self, - request: Request>, - ) -> Result, Status> { - trace!("CoreTelemetryService::insert_telemetry"); - - let cancel_token = self.cancellation_token.clone(); - let tlm_management = self.tlm_management.clone(); - let mut in_stream = request.into_inner(); - let (tx, rx) = mpsc::channel(128); - - tokio::spawn(async move { - loop { - select! { - _ = cancel_token.cancelled() => { - break; - }, - Some(message) = in_stream.next() => { - match message { - Ok(tlm_item) => { - tx - .send(Self::handle_new_tlm_item(&tlm_management, &tlm_item)) - .await - .expect("working rx"); - } - Err(err) => { - let _ = tx.send(Err(err)).await; - } - } - }, - else => break, - } - } - }); - - Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) - } -} - -impl CoreTelemetryService { - #[allow(clippy::result_large_err)] - fn handle_new_tlm_item( - tlm_management: &Arc, - tlm_item: &TelemetryItem, - ) -> Result { - trace!("CoreTelemetryService::handle_new_tlm_item {:?}", tlm_item); - let Some(ref uuid) = tlm_item.uuid else { - return Err(Status::failed_precondition("UUID Missing")); - }; - let tlm_management_pin = tlm_management.pin(); - let Some(tlm_data) = tlm_management_pin.get_by_uuid(&uuid.value) else { - return Err(Status::not_found("Telemetry Item Not Found")); - }; - - let Some(TelemetryValue { value: Some(value) }) = tlm_item.value else { - return Err(Status::failed_precondition("Value Missing")); - }; - - let Some(timestamp) = tlm_item.timestamp else { - return Err(Status::failed_precondition("Timestamp Missing")); - }; - - let expected_type = match value { - Value::Float32(_) => TelemetryDataType::Float32, - Value::Float64(_) => TelemetryDataType::Float64, - Value::Boolean(_) => TelemetryDataType::Boolean, - }; - if expected_type != tlm_data.data.definition.data_type { - return Err(Status::failed_precondition("Data Type Mismatch")); - }; - - let Some(timestamp) = DateTime::from_timestamp(timestamp.secs, timestamp.nanos as u32) - else { - return Err(Status::invalid_argument("Failed to construct UTC DateTime")); - }; - - let value = match value { - Value::Float32(x) => TelemetryDataValue::Float32(x), - Value::Float64(x) => TelemetryDataValue::Float64(x), - Value::Boolean(x) => TelemetryDataValue::Boolean(x), - }; - let _ = tlm_data.data.data.send_replace(Some(TelemetryDataItem { - value: value.clone(), - timestamp: timestamp.to_rfc3339_opts(SecondsFormat::Millis, true), - })); - TelemetryHistory::insert_sync( - tlm_data.clone(), - tlm_management.history_service(), - value, - timestamp, - ); - - Ok(TelemetryInsertResponse {}) - } -} diff --git a/server/src/http/api/tlm.rs b/server/src/http/api/tlm.rs index f400db8..38d7156 100644 --- a/server/src/http/api/tlm.rs +++ b/server/src/http/api/tlm.rs @@ -7,6 +7,7 @@ use serde::Deserialize; use std::sync::Arc; use std::time::Duration; use tokio::time::timeout; +use uuid::Uuid; #[get("/tlm/info/{name:[\\w\\d/_-]+}")] pub(super) async fn get_tlm_definition( @@ -36,13 +37,17 @@ struct HistoryQuery { resolution: i64, } -#[get("/tlm/history/{uuid:[0-9a-f]+}")] +#[get("/tlm/history/{uuid:[0-9a-f-]+}")] pub(super) async fn get_tlm_history( data_arc: web::Data>, uuid: web::Path, info: web::Query, ) -> Result { - let uuid = uuid.to_string(); + let Ok(uuid) = Uuid::parse_str(&uuid) else { + return Err(HttpServerResultError::InvalidUuid { + uuid: uuid.to_string(), + }); + }; trace!( "get_tlm_history {} from {} to {} resolution {}", uuid, diff --git a/server/src/http/backend/connection.rs b/server/src/http/backend/connection.rs new file mode 100644 index 0000000..895bdaf --- /dev/null +++ b/server/src/http/backend/connection.rs @@ -0,0 +1,81 @@ +use crate::telemetry::management_service::TelemetryManagementService; +use actix_ws::{AggregatedMessage, ProtocolError, Session}; +use anyhow::bail; +use api::request::{RequestMessage, RequestMessagePayload}; +use api::response::ResponseMessage; +use std::sync::Arc; +use tokio::sync::mpsc::{Receiver, Sender}; + +pub(super) struct BackendConnection { + session: Session, + tlm_management: Arc, + tx: Sender, + pub rx: Receiver, + pub should_close: bool, +} + +impl BackendConnection { + pub fn new(session: Session, tlm_management: Arc) -> Self { + let (tx, rx) = tokio::sync::mpsc::channel::(128); + Self { + session, + tlm_management, + tx, + rx, + should_close: false, + } + } + + async fn handle_request(&mut self, msg: RequestMessage) -> anyhow::Result<()> { + match msg.payload { + RequestMessagePayload::TelemetryDefinitionRequest(tlm_def) => { + self.tx + .send(ResponseMessage { + uuid: msg.uuid, + payload: self.tlm_management.register(tlm_def)?.into(), + }) + .await?; + } + RequestMessagePayload::TelemetryEntry(tlm_entry) => { + self.tlm_management.add_tlm_item(tlm_entry)?; + } + } + Ok(()) + } + + pub async fn handle_request_message( + &mut self, + msg: Result, + ) -> anyhow::Result<()> { + let msg = msg?; + match msg { + AggregatedMessage::Text(data) => { + self.handle_request(serde_json::from_str(&data)?).await?; + } + AggregatedMessage::Binary(_) => { + bail!("Binary Messages Unsupported"); + } + AggregatedMessage::Ping(bytes) => { + self.session.pong(&bytes).await?; + } + AggregatedMessage::Pong(_) => { + // Intentionally Ignore + } + AggregatedMessage::Close(_) => { + self.should_close = true; + } + } + Ok(()) + } + + pub async fn handle_response(&mut self, msg: ResponseMessage) -> anyhow::Result<()> { + let msg_json = serde_json::to_string(&msg)?; + self.session.text(msg_json).await?; + Ok(()) + } + + pub async fn cleanup(mut self) { + self.rx.close(); + let _ = self.session.close(None).await; + } +} diff --git a/server/src/http/backend/mod.rs b/server/src/http/backend/mod.rs new file mode 100644 index 0000000..3285285 --- /dev/null +++ b/server/src/http/backend/mod.rs @@ -0,0 +1,57 @@ +mod connection; + +use crate::http::backend::connection::BackendConnection; +use crate::telemetry::management_service::TelemetryManagementService; +use actix_web::{rt, web, HttpRequest, HttpResponse}; +use log::{error, trace}; +use std::sync::Arc; +use tokio::select; +use tokio_util::sync::CancellationToken; +use tonic::codegen::tokio_stream::StreamExt; + +async fn backend_connect( + req: HttpRequest, + stream: web::Payload, + cancel_token: web::Data, + telemetry_management_service: web::Data>, +) -> Result { + trace!("backend_connect"); + let (res, session, stream) = actix_ws::handle(&req, stream)?; + + let mut stream = stream + .aggregate_continuations() + // up to 1 MiB + .max_continuation_size(2_usize.pow(20)); + + let cancel_token = cancel_token.get_ref().clone(); + let tlm_management = telemetry_management_service.get_ref().clone(); + + rt::spawn(async move { + let mut connection = BackendConnection::new(session, tlm_management); + while !connection.should_close { + let result = select! { + _ = cancel_token.cancelled() => { + connection.should_close = true; + Ok(()) + }, + Some(msg) = connection.rx.recv() => connection.handle_response(msg).await, + Some(msg) = stream.next() => connection.handle_request_message(msg).await, + else => { + connection.should_close = true; + Ok(()) + }, + }; + if let Err(e) = result { + error!("backend socket error: {e}"); + connection.should_close = true; + } + } + connection.cleanup().await; + }); + + Ok(res) +} + +pub fn setup_backend(cfg: &mut web::ServiceConfig) { + cfg.route("", web::get().to(backend_connect)); +} diff --git a/server/src/http/error.rs b/server/src/http/error.rs index 36ae486..f2d8113 100644 --- a/server/src/http/error.rs +++ b/server/src/http/error.rs @@ -3,13 +3,16 @@ use actix_web::http::header::ContentType; use actix_web::http::StatusCode; use actix_web::HttpResponse; use thiserror::Error; +use uuid::Uuid; #[derive(Error, Debug)] pub enum HttpServerResultError { #[error("Telemetry Name Not Found: {tlm}")] TlmNameNotFound { tlm: String }, + #[error("Invalid Uuid: {uuid}")] + InvalidUuid { uuid: String }, #[error("Telemetry Uuid Not Found: {uuid}")] - TlmUuidNotFound { uuid: String }, + TlmUuidNotFound { uuid: Uuid }, #[error("DateTime Parsing Error: {date_time}")] InvalidDateTime { date_time: String }, #[error("Timed out")] @@ -26,6 +29,7 @@ impl ResponseError for HttpServerResultError { fn status_code(&self) -> StatusCode { match self { HttpServerResultError::TlmNameNotFound { .. } => StatusCode::NOT_FOUND, + HttpServerResultError::InvalidUuid { .. } => StatusCode::BAD_REQUEST, HttpServerResultError::TlmUuidNotFound { .. } => StatusCode::NOT_FOUND, HttpServerResultError::InvalidDateTime { .. } => StatusCode::BAD_REQUEST, HttpServerResultError::Timeout => StatusCode::GATEWAY_TIMEOUT, diff --git a/server/src/http/mod.rs b/server/src/http/mod.rs index 250eafb..9b82a48 100644 --- a/server/src/http/mod.rs +++ b/server/src/http/mod.rs @@ -1,9 +1,11 @@ mod api; +mod backend; mod error; mod websocket; use crate::command::service::CommandManagementService; use crate::http::api::setup_api; +use crate::http::backend::setup_backend; use crate::http::websocket::setup_websocket; use crate::panels::PanelService; use crate::telemetry::management_service::TelemetryManagementService; @@ -31,6 +33,7 @@ pub async fn setup( .app_data(cancel_token.clone()) .app_data(panel_service.clone()) .app_data(command_service.clone()) + .service(web::scope("/backend").configure(setup_backend)) .service(web::scope("/ws").configure(setup_websocket)) .service(web::scope("/api").configure(setup_api)) .wrap(Logger::default()) diff --git a/server/src/http/websocket/mod.rs b/server/src/http/websocket/mod.rs index 6e2911c..8b10ef1 100644 --- a/server/src/http/websocket/mod.rs +++ b/server/src/http/websocket/mod.rs @@ -15,6 +15,7 @@ use tokio::sync::mpsc::Sender; use tokio::time::{sleep_until, Instant}; use tokio_util::sync::CancellationToken; use tonic::codegen::tokio_stream::StreamExt; +use uuid::Uuid; pub mod request; pub mod response; @@ -23,11 +24,11 @@ fn handle_register_tlm_listener( data: &Arc, request: RegisterTlmListenerRequest, tx: &Sender, - tlm_listeners: &mut HashMap, + tlm_listeners: &mut HashMap, ) { if let Some(tlm_data) = data.get_by_uuid(&request.uuid) { let token = CancellationToken::new(); - if let Some(token) = tlm_listeners.insert(tlm_data.definition.uuid.clone(), token.clone()) { + if let Some(token) = tlm_listeners.insert(tlm_data.definition.uuid, token.clone()) { token.cancel(); } let minimum_separation = Duration::from_millis(request.minimum_separation_ms as u64); @@ -46,7 +47,7 @@ fn handle_register_tlm_listener( ref_val.clone() }; let _ = tx.send(TlmValueResponse { - uuid: request.uuid.clone(), + uuid: request.uuid, value, }.into()).await; now @@ -65,7 +66,7 @@ fn handle_register_tlm_listener( fn handle_unregister_tlm_listener( request: UnregisterTlmListenerRequest, - tlm_listeners: &mut HashMap, + tlm_listeners: &mut HashMap, ) { if let Some(token) = tlm_listeners.remove(&request.uuid) { token.cancel(); @@ -76,7 +77,7 @@ async fn handle_websocket_message( data: &Arc, request: WebsocketRequest, tx: &Sender, - tlm_listeners: &mut HashMap, + tlm_listeners: &mut HashMap, ) { match request { WebsocketRequest::RegisterTlmListener(request) => { @@ -110,7 +111,7 @@ async fn handle_websocket_incoming( data: &Arc, session: &mut Session, tx: &Sender, - tlm_listeners: &mut HashMap, + tlm_listeners: &mut HashMap, ) -> anyhow::Result { match msg { Ok(AggregatedMessage::Close(_)) => Ok(false), @@ -130,7 +131,7 @@ async fn handle_websocket_incoming( } } -pub async fn websocket_connect( +async fn websocket_connect( req: HttpRequest, stream: web::Payload, data: web::Data>, diff --git a/server/src/http/websocket/request.rs b/server/src/http/websocket/request.rs index 6eaa984..58c3385 100644 --- a/server/src/http/websocket/request.rs +++ b/server/src/http/websocket/request.rs @@ -1,15 +1,16 @@ use derive_more::From; use serde::{Deserialize, Serialize}; +use uuid::Uuid; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RegisterTlmListenerRequest { - pub uuid: String, + pub uuid: Uuid, pub minimum_separation_ms: u32, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct UnregisterTlmListenerRequest { - pub uuid: String, + pub uuid: Uuid, } #[derive(Debug, Clone, Serialize, Deserialize, From)] diff --git a/server/src/http/websocket/response.rs b/server/src/http/websocket/response.rs index f056572..f2302c3 100644 --- a/server/src/http/websocket/response.rs +++ b/server/src/http/websocket/response.rs @@ -1,10 +1,11 @@ use crate::telemetry::data_item::TelemetryDataItem; use derive_more::From; use serde::{Deserialize, Serialize}; +use uuid::Uuid; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TlmValueResponse { - pub uuid: String, + pub uuid: Uuid, pub value: Option, } diff --git a/server/src/lib.rs b/server/src/lib.rs index 0fdf332..1b5a026 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -53,7 +53,7 @@ pub async fn setup() -> anyhow::Result<()> { let cmd = Arc::new(CommandManagementService::new()); - let grpc_server = grpc::setup(cancellation_token.clone(), tlm.clone(), cmd.clone())?; + let grpc_server = grpc::setup(cancellation_token.clone(), cmd.clone())?; let panel_service = PanelService::new(sqlite.clone()); diff --git a/server/src/telemetry/data_item.rs b/server/src/telemetry/data_item.rs index 8c7e482..c5b2588 100644 --- a/server/src/telemetry/data_item.rs +++ b/server/src/telemetry/data_item.rs @@ -1,8 +1,8 @@ -use crate::telemetry::data_value::TelemetryDataValue; +use api::data_value::DataValue; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TelemetryDataItem { - pub value: TelemetryDataValue, + pub value: DataValue, pub timestamp: String, } diff --git a/server/src/telemetry/definition.rs b/server/src/telemetry/definition.rs index 95deb36..fb17fa0 100644 --- a/server/src/telemetry/definition.rs +++ b/server/src/telemetry/definition.rs @@ -1,13 +1,10 @@ -use crate::core::TelemetryDataType; -use crate::telemetry::data_type::tlm_data_type_deserializer; -use crate::telemetry::data_type::tlm_data_type_serializer; +use api::data_type::DataType; use serde::{Deserialize, Serialize}; +use uuid::Uuid; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TelemetryDefinition { - pub uuid: String, + pub uuid: Uuid, pub name: String, - #[serde(serialize_with = "tlm_data_type_serializer")] - #[serde(deserialize_with = "tlm_data_type_deserializer")] - pub data_type: TelemetryDataType, + pub data_type: DataType, } diff --git a/server/src/telemetry/history.rs b/server/src/telemetry/history.rs index 78b36a2..ab5802f 100644 --- a/server/src/telemetry/history.rs +++ b/server/src/telemetry/history.rs @@ -1,10 +1,10 @@ -use crate::core::TelemetryDataType; use crate::serialization::file_ext::{ReadExt, WriteExt}; use crate::telemetry::data::TelemetryData; use crate::telemetry::data_item::TelemetryDataItem; -use crate::telemetry::data_value::TelemetryDataValue; use crate::telemetry::definition::TelemetryDefinition; use anyhow::{anyhow, ensure, Context}; +use api::data_type::DataType; +use api::data_value::DataValue; use chrono::{DateTime, DurationRound, SecondsFormat, TimeDelta, Utc}; use log::{error, info}; use std::cmp::min; @@ -44,7 +44,7 @@ fn update_next_from( } struct SegmentData { - values: Vec, + values: Vec, timestamps: Vec>, } @@ -66,7 +66,7 @@ impl HistorySegmentRam { } } - fn insert(&self, value: TelemetryDataValue, timestamp: DateTime) { + fn insert(&self, value: DataValue, timestamp: DateTime) { if timestamp < self.start || timestamp >= self.end { return; } @@ -121,7 +121,7 @@ impl HistorySegmentRam { next_from, ); result.push(TelemetryDataItem { - value: data.values[i].clone(), + value: data.values[i], timestamp: t.to_rfc3339_opts(SecondsFormat::Millis, true), }); } @@ -196,9 +196,9 @@ impl HistorySegmentFile { // Write all the values for value in &data.values { match value { - TelemetryDataValue::Float32(value) => file.write_data::(*value)?, - TelemetryDataValue::Float64(value) => file.write_data::(*value)?, - TelemetryDataValue::Boolean(value) => file.write_data::(*value)?, + DataValue::Float32(value) => file.write_data::(*value)?, + DataValue::Float64(value) => file.write_data::(*value)?, + DataValue::Boolean(value) => file.write_data::(*value)?, } } @@ -215,10 +215,7 @@ impl HistorySegmentFile { }) } - fn load_to_ram( - mut self, - telemetry_data_type: TelemetryDataType, - ) -> anyhow::Result { + fn load_to_ram(mut self, telemetry_data_type: DataType) -> anyhow::Result { let mut segment_data = SegmentData { values: Vec::with_capacity(self.length as usize), timestamps: Vec::with_capacity(self.length as usize), @@ -281,7 +278,7 @@ impl HistorySegmentFile { from: DateTime, to: DateTime, maximum_resolution: TimeDelta, - telemetry_data_type: TelemetryDataType, + telemetry_data_type: DataType, ) -> anyhow::Result<(DateTime, Vec)> { self.file_position = 0; self.file.seek(SeekFrom::Start(0))?; @@ -334,22 +331,19 @@ impl HistorySegmentFile { self.read_date_time() } - fn read_telemetry_item( - &mut self, - telemetry_data_type: TelemetryDataType, - ) -> anyhow::Result { + fn read_telemetry_item(&mut self, telemetry_data_type: DataType) -> anyhow::Result { match telemetry_data_type { - TelemetryDataType::Float32 => { + DataType::Float32 => { self.file_position += 4; - Ok(TelemetryDataValue::Float32(self.file.read_data::()?)) + Ok(DataValue::Float32(self.file.read_data::()?)) } - TelemetryDataType::Float64 => { + DataType::Float64 => { self.file_position += 8; - Ok(TelemetryDataValue::Float64(self.file.read_data::()?)) + Ok(DataValue::Float64(self.file.read_data::()?)) } - TelemetryDataType::Boolean => { + DataType::Boolean => { self.file_position += 1; - Ok(TelemetryDataValue::Boolean(self.file.read_data::()?)) + Ok(DataValue::Boolean(self.file.read_data::()?)) } } } @@ -357,12 +351,12 @@ impl HistorySegmentFile { fn get_telemetry_item( &mut self, index: u64, - telemetry_data_type: TelemetryDataType, - ) -> anyhow::Result { + telemetry_data_type: DataType, + ) -> anyhow::Result { let item_length = match telemetry_data_type { - TelemetryDataType::Float32 => 4, - TelemetryDataType::Float64 => 8, - TelemetryDataType::Boolean => 1, + DataType::Float32 => 4, + DataType::Float64 => 8, + DataType::Boolean => 1, }; let desired_position = Self::HEADER_LENGTH + self.length * Self::TIMESTAMP_LENGTH + index * item_length; @@ -429,7 +423,7 @@ impl TelemetryHistory { history_segment_ram: HistorySegmentRam, ) -> JoinHandle<()> { let mut path = service.data_root_folder.clone(); - path.push(&self.data.definition.uuid); + path.push(self.data.definition.uuid.as_hyphenated().to_string()); spawn_blocking(move || { match HistorySegmentFile::save_to_disk(path, history_segment_ram) { // Immediately drop the segment - now that we've saved it to disk we don't need to keep it in memory @@ -450,7 +444,7 @@ impl TelemetryHistory { start: DateTime, ) -> JoinHandle> { let mut path = service.data_root_folder.clone(); - path.push(&self.data.definition.uuid); + path.push(self.data.definition.uuid.as_hyphenated().to_string()); spawn_blocking(move || HistorySegmentFile::open(path, start)) } @@ -458,7 +452,7 @@ impl TelemetryHistory { &self, start: DateTime, service: &TelemetryHistoryService, - telemetry_data_type: TelemetryDataType, + telemetry_data_type: DataType, ) -> HistorySegmentRam { let ram = self .get_disk_segment(service, start) @@ -480,7 +474,7 @@ impl TelemetryHistory { pub async fn insert( &self, service: &TelemetryHistoryService, - value: TelemetryDataValue, + value: DataValue, timestamp: DateTime, ) { let segments = self.segments.read().await; @@ -531,7 +525,7 @@ impl TelemetryHistory { pub fn insert_sync( history: Arc, service: Arc, - value: TelemetryDataValue, + value: DataValue, timestamp: DateTime, ) { tokio::spawn(async move { @@ -579,7 +573,7 @@ impl TelemetryHistory { .unwrap(); let mut path = telemetry_history_service.data_root_folder.clone(); - path.push(&self.data.definition.uuid); + path.push(self.data.definition.uuid.as_hyphenated().to_string()); let mut start = start; while start < end { diff --git a/server/src/telemetry/management_service.rs b/server/src/telemetry/management_service.rs index d819cc7..63e8c81 100644 --- a/server/src/telemetry/management_service.rs +++ b/server/src/telemetry/management_service.rs @@ -1,7 +1,13 @@ -use crate::core::{TelemetryDefinitionRequest, Uuid}; use crate::telemetry::data::TelemetryData; +use crate::telemetry::data_item::TelemetryDataItem; use crate::telemetry::definition::TelemetryDefinition; use crate::telemetry::history::{TelemetryHistory, TelemetryHistoryService}; +use anyhow::bail; +use api::data_type::DataType; +use api::data_value::DataValue; +use api::request::{TelemetryDefinitionRequest, TelemetryEntry}; +use api::response::TelemetryDefinitionResponse; +use chrono::SecondsFormat; use log::{error, info, warn}; use papaya::{HashMap, HashMapRef, LocalGuard}; use std::fs; @@ -12,12 +18,13 @@ use std::sync::Arc; use std::time::Duration; use tokio::sync::Mutex; use tokio::time::sleep; +use uuid::Uuid; const RELEASED_ATTEMPTS: usize = 5; pub struct TelemetryManagementService { - uuid_index: HashMap, - tlm_data: HashMap>, + uuid_index: HashMap, + tlm_data: HashMap>, telemetry_history_service: Arc, metadata_file: Arc>, } @@ -49,8 +56,8 @@ impl TelemetryManagementService { // Skip invalid entries match serde_json::from_str::(line) { Ok(tlm_def) => { - let _ = uuid_index.insert(tlm_def.name.clone(), tlm_def.uuid.clone()); - let _ = tlm_data.insert(tlm_def.uuid.clone(), Arc::new(tlm_def.into())); + let _ = uuid_index.insert(tlm_def.name.clone(), tlm_def.uuid); + let _ = tlm_data.insert(tlm_def.uuid, Arc::new(tlm_def.into())); } Err(err) => { error!("Failed to parse metadata entry {err}"); @@ -79,23 +86,20 @@ impl TelemetryManagementService { pub fn register( &self, telemetry_definition_request: TelemetryDefinitionRequest, - ) -> anyhow::Result { + ) -> anyhow::Result { let uuid_index = self.uuid_index.pin(); let tlm_data = self.tlm_data.pin(); - let uuid = uuid_index - .get_or_insert_with(telemetry_definition_request.name.clone(), || { - Uuid::random().value - }) - .clone(); + let uuid = + *uuid_index.get_or_insert_with(telemetry_definition_request.name.clone(), Uuid::new_v4); let inserted = tlm_data.try_insert( - uuid.clone(), + uuid, Arc::new( TelemetryDefinition { - uuid: uuid.clone(), + uuid, name: telemetry_definition_request.name.clone(), - data_type: telemetry_definition_request.data_type(), + data_type: telemetry_definition_request.data_type, } .into(), ), @@ -129,7 +133,38 @@ impl TelemetryManagementService { }); } - Ok(uuid) + Ok(TelemetryDefinitionResponse { uuid }) + } + + pub fn add_tlm_item(&self, tlm_item: TelemetryEntry) -> anyhow::Result<()> { + let tlm_management_pin = self.pin(); + let Some(tlm_data) = tlm_management_pin.get_by_uuid(&tlm_item.uuid) else { + bail!("Telemetry Item Not Found"); + }; + + let expected_type = match &tlm_item.value { + DataValue::Float32(_) => DataType::Float32, + DataValue::Float64(_) => DataType::Float64, + DataValue::Boolean(_) => DataType::Boolean, + }; + if expected_type != tlm_data.data.definition.data_type { + bail!("Data Type Mismatch"); + }; + + let _ = tlm_data.data.data.send_replace(Some(TelemetryDataItem { + value: tlm_item.value, + timestamp: tlm_item + .timestamp + .to_rfc3339_opts(SecondsFormat::Millis, true), + })); + TelemetryHistory::insert_sync( + tlm_data.clone(), + self.history_service(), + tlm_item.value, + tlm_item.timestamp, + ); + + Ok(()) } pub fn get_by_name(&self, name: &String) -> Option { @@ -138,7 +173,7 @@ impl TelemetryManagementService { self.get_by_uuid(uuid) } - pub fn get_by_uuid(&self, uuid: &String) -> Option { + pub fn get_by_uuid(&self, uuid: &Uuid) -> Option { let tlm_data = self.tlm_data.pin(); tlm_data .get(uuid) @@ -200,11 +235,11 @@ impl TelemetryManagementService { } pub struct TelemetryManagementServicePin<'a> { - tlm_data: HashMapRef<'a, String, Arc, RandomState, LocalGuard<'a>>, + tlm_data: HashMapRef<'a, Uuid, Arc, RandomState, LocalGuard<'a>>, } impl<'a> TelemetryManagementServicePin<'a> { - pub fn get_by_uuid(&'a self, uuid: &String) -> Option<&'a Arc> { + pub fn get_by_uuid(&'a self, uuid: &Uuid) -> Option<&'a Arc> { self.tlm_data.get(uuid) } } diff --git a/server/src/telemetry/mod.rs b/server/src/telemetry/mod.rs index f95ccc7..d5272b6 100644 --- a/server/src/telemetry/mod.rs +++ b/server/src/telemetry/mod.rs @@ -1,7 +1,6 @@ pub mod data; pub mod data_item; pub mod data_type; -pub mod data_value; pub mod definition; pub mod history; pub mod management_service;