Datastore sync MVP (#61)

* Datastore sync MVP * tunng * Tuned GCS * Updated
victoryforphil · Nov 12, 2024 · 00e4765 · 00e4765
1 parent a0b89c3
commit 00e4765
Show file tree

Hide file tree

Showing 22 changed files with 118 additions and 78 deletions.
diff --git a/lil-gcs/Cargo.toml b/lil-gcs/Cargo.toml
@@ -20,7 +20,7 @@ victory-wtf = { workspace = true }
 lil-link = { path = "../lil-link" }
 serde_json = "1.0.132"
 polars = { version = "0.44.0", features = ["csv"] }
-tokio = "1.41.0"
+tokio = { version = "1.41.0", features = ["full"] }
 tokio-tungstenite = "0.24.0"
 tungstenite = "0.24.0"
 futures = "0.3.31"
diff --git a/lil-gcs/src/main.rs b/lil-gcs/src/main.rs
@@ -13,6 +13,9 @@ use std::sync::Mutex;
 use std::thread;
 use std::time::Duration;
 use std::time::{SystemTime, UNIX_EPOCH};
+use victory_data_store::database::listener::DataStoreListener;
+use victory_data_store::sync::adapters::tcp::tcp_client::TCPClient;
+use victory_data_store::sync::config::SyncConfig;
 
 use clap::Parser;
 use serde::Serialize;
@@ -21,10 +24,6 @@ use tracing::warn;
 use tracing::Level;
 use tracing_subscriber::fmt;
 
-use victory_broker::adapters::tcp::TCPClientAdapter;
-use victory_broker::adapters::tcp::TCPClientOptions;
-use victory_broker::node::sub_callback::SubCallback;
-use victory_broker::node::Node;
 use victory_data_store::database::Datastore;
 use victory_data_store::topics::TopicKey;
 use victory_wtf::Timepoint;
@@ -38,15 +37,23 @@ pub struct TCPNodeSubscriber {
     update: BTreeMap<String, String>,
 }
 
-impl SubCallback for TCPNodeSubscriber {
-    fn on_update(&mut self, datapoints: &victory_data_store::datapoints::DatapointMap) {
-        for (topic, datapoint) in datapoints.iter() {
-            self.map
-                .insert(topic.display_name(), format!("{:?}", datapoint.value));
+impl DataStoreListener for TCPNodeSubscriber {
+    fn on_datapoint(&mut self, datapoint: &victory_data_store::datapoints::Datapoint) {
+
+    }
 
-            self.update
-                .insert(topic.display_name(), format!("{:?}", datapoint.value));
-        }
+    fn on_raw_datapoint(&mut self, datapoint: &victory_data_store::datapoints::Datapoint) {
+        let topic = datapoint.topic.clone();
+        self.map
+            .insert(topic.display_name(), format!("{:?}", datapoint.value));
+
+        self.update
+            .insert(topic.display_name(), format!("{:?}", datapoint.value));
+    }
+
+
+    fn on_bucket_update(&mut self, bucket: &victory_data_store::buckets::BucketHandle) {
+
     }
 }
 
@@ -106,7 +113,7 @@ async fn main() {
     tokio::spawn(webserver::websocket_server_task(tcp_tx.clone(), ws_tx));
 
     fmt()
-        .with_max_level(Level::INFO)
+        .with_max_level(Level::DEBUG)
         .with_target(true)
         .pretty()
         .compact()
@@ -117,41 +124,49 @@ async fn main() {
 
     let args = SILArgs::parse();
 
-    let mut client = TCPClientAdapter::new(TCPClientOptions::from_url(&args.connection));
+    let mut client = TCPClient::new(args.connection.to_string()).await;
 
     while client.is_err() {
         info!("Failed to connect to server, retrying...");
         thread::sleep(Duration::from_secs_f32(1.0));
-        client = TCPClientAdapter::new(TCPClientOptions::from_url(&args.connection));
+        client = TCPClient::new(args.connection.to_string()).await;
     }
 
     let client = client.unwrap();
 
     let client_handle = Arc::new(Mutex::new(client));
     let datastore = Datastore::new().handle();
-    let mut node = Node::new(
-        "TCP Client".to_string(),
-        client_handle.clone(),
-        datastore.clone(),
-    );
 
-    let subscriber = TCPNodeSubscriber {
+    let topic_key = TopicKey::from_str("");
+
+    let sync_config = SyncConfig {
+        client_name: "GCS".to_string(),
+        subscriptions: vec![topic_key.display_name()],
+    };
+    datastore
+        .lock()
+        .unwrap()
+        .setup_sync(sync_config, client_handle);
+
+
+    let mut listener = TCPNodeSubscriber {
         map: BTreeMap::new(),
         update: BTreeMap::new(),
     };
 
-    let subscriber_handle = Arc::new(Mutex::new(subscriber));
+    let listener_handle = Arc::new(Mutex::new(listener));
 
-    let topic_key = TopicKey::from_str("");
-    node.add_sub_callback(topic_key, subscriber_handle.clone());
-    node.register();
+    datastore.lock().unwrap().add_listener(&topic_key, listener_handle.clone());
+
+
 
-    let subscriber_handle_clone = subscriber_handle.clone();
+    let subscriber_handle_clone = listener_handle.clone();
     let tcp_tx_clone = tcp_tx.clone();
 
-    let subscriber_handle_clone_ws = subscriber_handle.clone();
+    let subscriber_handle_clone_ws = listener_handle.clone();
     let tcp_tx_clone_ws = tcp_tx.clone();
 
+    let datastore_clone = datastore.clone();
     tokio::spawn(async move {
         loop {
             if let Some(ws_msg) = ws_rx.recv().await {
@@ -160,7 +175,7 @@ async fn main() {
                 // Need to parse from json or something a topic and a value. It would be great if I could parse a normal victory value.
 
                 // I thought this was a scope thing at some point.
-                let mut datastore = datastore.lock().expect("Failed to lock mutex");
+                let mut datastore = datastore_clone.lock().expect("Failed to lock mutex");
 
                 let (msg_topic, params) = parse_message(&ws_msg);
 
@@ -288,7 +303,7 @@ async fn main() {
     tokio::spawn(async move {
         loop {
             // thread::sleep(Duration::from_secs_f32(2.0));
-            tokio::time::sleep(Duration::from_secs_f32(0.25)).await;
+            tokio::time::sleep(Duration::from_millis(200)).await;
             {
                 let mut map = subscriber_handle_clone.lock().unwrap();
 
@@ -318,9 +333,9 @@ async fn main() {
             }
         }
     });
-
+    let datastore = datastore.clone();
     loop {
         tokio::time::sleep(Duration::from_millis(100)).await;
-        node.tick();
+        datastore.lock().unwrap().run_sync();
     }
 }
diff --git a/lil-launcher/Cargo.toml b/lil-launcher/Cargo.toml
@@ -48,4 +48,5 @@ victory-broker = { workspace = true }
 lil-rerun = { path = "../lil-rerun" }
 lil-quad = { path = "../lil-quad" }
 lil-link = { path = "../lil-link" }
+tokio = { version = "1.41.1", features = ["full"] }
 
diff --git a/lil-launcher/bin_tests/quad_idle.rs b/lil-launcher/bin_tests/quad_idle.rs
@@ -27,6 +27,8 @@ use victory_broker::adapters::tcp::TCPServerAdapter;
 use victory_broker::adapters::tcp::TCPServerOptions;
 use victory_commander::system::runner::BasherSysRunner;
 use victory_data_store::primitives::Primitives;
+use victory_data_store::sync::adapters::tcp::tcp_server::TcpSyncServer;
+use victory_data_store::sync::config::SyncConfig;
 use victory_data_store::topics::TopicKey;
 use victory_wtf::Timepoint;
 use victory_wtf::Timespan;
@@ -49,6 +51,10 @@ struct SILArgs {
     )]
     duration: f32,
 
+
+    #[clap(short, long, value_parser, help = "TCP Sync Server address")]
+    sync_server_address: String,
+
     #[clap(
         short,
         long,
@@ -58,8 +64,8 @@ struct SILArgs {
     )]
     arm_time: f32,
 }
-
-fn main() {
+#[tokio::main]
+async fn main() {
     fmt()
         .with_max_level(Level::INFO)
         .with_target(true)
@@ -74,13 +80,20 @@ fn main() {
     info!("Running 'quad_sil' with args: {:#?}", args);
 
     let mut runner = BasherSysRunner::new();
-    let server = TCPServerAdapter::new(TCPServerOptions {
-        port: 7001,
-        address: "0.0.0.0".to_string(),
-        update_interval: Timespan::new_hz(100.0),
-    });
+    let server = TcpSyncServer::new(&args.sync_server_address).await;
     let server_handle = Arc::new(Mutex::new(server));
-    runner.enable_pubsub(server_handle);
+
+    let topic_filter = TopicKey::from_str("cmd");
+
+    let sync_config = SyncConfig {
+        client_name: "Quad Idle".to_string(),
+        subscriptions: vec![topic_filter.display_name()],
+    };
+    runner.data_store
+        .lock()
+        .unwrap()
+        .setup_sync(sync_config, server_handle);
+
     runner.dt = Timespan::new_hz(args.hz as f64);
 
     runner.add_system(Arc::new(Mutex::new(

diff --git a/lil-launcher/bin_tests/quad_sil.rs b/lil-launcher/bin_tests/quad_sil.rs
@@ -21,10 +21,11 @@ use tracing::Level;
 use tracing_subscriber::fmt;
 
 use clap::Parser;
-use victory_broker::adapters::tcp::TCPServerAdapter;
-use victory_broker::adapters::tcp::TCPServerOptions;
+
 use victory_commander::system::runner::BasherSysRunner;
 use victory_data_store::primitives::Primitives;
+use victory_data_store::sync::adapters::tcp::tcp_server::TcpSyncServer;
+use victory_data_store::sync::config::SyncConfig;
 use victory_data_store::topics::TopicKey;
 use victory_wtf::Timepoint;
 use victory_wtf::Timespan;
@@ -35,6 +36,9 @@ struct SILArgs {
     #[clap(short, long, value_parser, help = "Mavlink connection string")]
     connection_string: String,
 
+    #[clap(short, long, value_parser, help = "TCP Sync Server address")]
+    sync_server_address: String,
+
     #[clap(long, value_parser, help = "Command Hz ", default_value = "25.0")]
     hz: f32,
 
@@ -56,10 +60,10 @@ struct SILArgs {
     )]
     arm_time: f32,
 }
-
-fn main() {
+#[tokio::main]
+async fn main() {
     fmt()
-        .with_max_level(Level::DEBUG)
+        .with_max_level(Level::INFO)
         .with_target(true)
         .pretty()
         .compact()
@@ -72,13 +76,20 @@ fn main() {
     info!("Running 'quad_sil' with args: {:#?}", args);
 
     let mut runner = BasherSysRunner::new();
-    let server = TCPServerAdapter::new(TCPServerOptions {
-        port: 7001,
-        address: "0.0.0.0".to_string(),
-        update_interval: Timespan::new_hz(100.0),
-    });
+    let server = TcpSyncServer::new(&args.sync_server_address).await;
     let server_handle = Arc::new(Mutex::new(server));
-    runner.enable_pubsub(server_handle);
+
+    let topic_filter = TopicKey::from_str("cmd");
+
+    let sync_config = SyncConfig {
+        client_name: "TCP Sync Server".to_string(),
+        subscriptions: vec![topic_filter.display_name()],
+    };
+    runner.data_store
+        .lock()
+        .unwrap()
+        .setup_sync(sync_config, server_handle);
+
     runner.dt = Timespan::new_hz(args.hz as f64);
 
     runner.add_system(Arc::new(Mutex::new(

diff --git a/lil-link/src/mavlink/processors/mod.rs b/lil-link/src/mavlink/processors/mod.rs
@@ -1,4 +1,4 @@
-use victory_data_store::database::DataView;
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use super::core::MavlinkMessageType;
 

diff --git a/lil-link/src/mavlink/processors/proc_attitude.rs b/lil-link/src/mavlink/processors/proc_attitude.rs
@@ -1,4 +1,4 @@
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use crate::{
     common::{

diff --git a/lil-link/src/mavlink/processors/proc_command_ack.rs b/lil-link/src/mavlink/processors/proc_command_ack.rs
@@ -1,5 +1,5 @@
 use log::info;
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use crate::{
     common::identifiers::{IDENT_BASE_LOG, IDENT_COMMAND_ACK},

diff --git a/lil-link/src/mavlink/processors/proc_ekf_health.rs b/lil-link/src/mavlink/processors/proc_ekf_health.rs
@@ -1,4 +1,4 @@
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use crate::{
     common::identifiers::{IDENT_BASE_STATUS, IDENT_STATUS_EKF, IDENT_STATUS_SENSORS},

diff --git a/lil-link/src/mavlink/processors/proc_heartbeat.rs b/lil-link/src/mavlink/processors/proc_heartbeat.rs
@@ -1,5 +1,5 @@
 use log::trace;
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use crate::{
     common::{

diff --git a/lil-link/src/mavlink/processors/proc_local_position.rs b/lil-link/src/mavlink/processors/proc_local_position.rs
@@ -1,4 +1,4 @@
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use crate::{
     common::{

diff --git a/lil-link/src/mavlink/processors/proc_param_value.rs b/lil-link/src/mavlink/processors/proc_param_value.rs
@@ -12,7 +12,7 @@ pub struct ParamValueProcessor;
 impl MavlinkMessageProcessor for ParamValueProcessor {
     fn on_mavlink_message(
         msg: crate::mavlink::core::MavlinkMessageType,
-        data_view: &mut victory_data_store::database::DataView,
+        data_view: &mut victory_data_store::database::view::DataView,
     ) -> Result<(), anyhow::Error> {
         let param_value_msg = match msg {
             MavlinkMessageType::PARAM_VALUE(param_value) => param_value,

diff --git a/lil-link/src/mavlink/processors/proc_status_text.rs b/lil-link/src/mavlink/processors/proc_status_text.rs
@@ -1,5 +1,5 @@
 use log::info;
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use crate::{
     common::identifiers::{IDENT_BASE_LOG, IDENT_STATUS_TEXT},

diff --git a/lil-link/src/mavlink/processors/proc_sys_status.rs b/lil-link/src/mavlink/processors/proc_sys_status.rs
@@ -1,4 +1,4 @@
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 
 use crate::{
     common::identifiers::*,

diff --git a/lil-link/src/mavlink/system.rs b/lil-link/src/mavlink/system.rs
@@ -5,7 +5,7 @@ use std::{
 
 use log::{debug, info};
 use victory_commander::system::System;
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 use victory_wtf::Timespan;
 
 use crate::common::types::{

diff --git a/lil-quad/src/systems/health_check.rs b/lil-quad/src/systems/health_check.rs
@@ -7,7 +7,7 @@ use lil_link::common::{
 use log::info;
 use serde::{Deserialize, Serialize};
 use victory_commander::system::System;
-use victory_data_store::{database::DataView, topics::TopicKey};
+use victory_data_store::{database::view::DataView, topics::TopicKey};
 use victory_wtf::Timepoint;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -59,9 +59,9 @@ impl System for HealthCheck {
 
     fn execute(
         &mut self,
-        inputs: &victory_data_store::database::DataView,
+        inputs: &victory_data_store::database::view::DataView,
         _dt: victory_wtf::Timespan,
-    ) -> victory_data_store::database::DataView {
+    ) -> victory_data_store::database::view::DataView {
         let ekf_status: QuadEkfStatus = inputs
             .get_latest(&TopicKey::from_str(&format!(
                 "{}/{}",