networking overhaul

- convert to axum - rework strand/braid connections - remove bui-backend
strawlab · Jan 22, 2024 · 8c6defc · 8c6defc
1 parent ebf1bc0
commit 8c6defc
Show file tree

Hide file tree

Showing 56 changed files with 2,385 additions and 2,356 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -27,6 +27,7 @@ members = [
     "braidz-viewer",
     "bui-backend-session",
     "bui-backend-session/demo",
+    "bui-backend-session/types",
     "build-util",
     "camcal",
     "led-box",
@@ -49,6 +50,7 @@ members = [
     "enum-iter",
     "env-tracing-logger",
     "env-tracing-logger/env-tracing-logger-sample",
+    "event-stream-types",
     "fastimage",
     "fastfreeimage",
     "fly-eye",
@@ -120,9 +122,7 @@ members = [
     "zip-or-dir/dir2zip",
 ]
 
-exclude = [
-    "led-box-firmware",
-]
+exclude = ["led-box-firmware"]
 
 [profile.release]
 debug = true

diff --git a/ads-webasm/Cargo.toml b/ads-webasm/Cargo.toml
@@ -12,7 +12,6 @@ js-sys = "0.3"
 gloo = "0.8.0"
 gloo-file = "0.2"
 wasm-bindgen = { version = "0.2.58" }
-http = "0.2"
 serde = "1.0"
 serde_yaml = "0.9"
 serde_derive = "1.0"
@@ -27,14 +26,14 @@ uuid = { version = "1.2.2", default-features = false, features = [
 ] } # add feature flag required for uuid crate
 csv = { version = "1.1", optional = true }
 
-bui-backend-types = "0.8"
 yew-tincture = "0.1"
 
 simple-obj-parse = { path = "../simple-obj-parse", optional = true }
 textured-tri-mesh = { path = "../textured-tri-mesh", optional = true }
 http-video-streaming-types = { path = "../http-video-streaming/http-video-streaming-types" }
 enum-iter = { path = "../enum-iter" }
 rust-cam-bui-types = { path = "../rust-cam-bui-types" }
+bui-backend-session-types = { path = "../bui-backend-session/types" }
 
 [dependencies.web-sys]
 version = "0.3"

diff --git a/ads-webasm/src/components/video_field.rs b/ads-webasm/src/components/video_field.rs
@@ -1,7 +1,7 @@
 use std::{cell::RefCell, rc::Rc};
 
 use crate::video_data::VideoData;
-use bui_backend_types;
+use bui_backend_session_types;
 use gloo::timers::callback::Timeout;
 use serde::{Deserialize, Serialize};
 use wasm_bindgen::prelude::*;
@@ -29,7 +29,7 @@ pub struct ImData2 {
     pub draw_shapes: Vec<CanvasDrawableShape>,
     pub fno: u64,
     pub ts_rfc3339: String, // timestamp in RFC3339 format
-    pub ck: bui_backend_types::ConnectionKey,
+    pub ck: bui_backend_session_types::ConnectionKey,
     pub name: Option<String>,
 }
 

diff --git a/braid-http-session/Cargo.toml b/braid-http-session/Cargo.toml
@@ -2,19 +2,17 @@
 name = "braid-http-session"
 version = "0.1.0"
 edition = "2021"
-rust-version="1.60"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+rust-version = "1.60"
 
 [dependencies]
-log = "0.4"
+tracing = "0.1"
 thiserror = "1"
 hyper = "1.1"
-serde = {version="1.0",features=["derive"]}
+serde = { version = "1.0", features = ["derive"] }
 serde_json = "1"
 futures = "0.3"
 
-bui-backend-session = {path="../bui-backend-session"}
-flydra-types = { path = "../flydra-types", features=["with-dns"] }
+bui-backend-session = { path = "../bui-backend-session" }
+flydra-types = { path = "../flydra-types" }
 http-body-util = "0.1.0"
 bytes = "1.5.0"
diff --git a/braid-http-session/src/lib.rs b/braid-http-session/src/lib.rs
@@ -1,5 +1,5 @@
 use ::bui_backend_session::{future_session, InsecureSession};
-use log::{debug, error};
+use tracing::{debug, error};
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
@@ -47,33 +47,24 @@ impl MainbrainSession {
     async fn do_post(&mut self, bytes: Vec<u8>) -> Result<(), Error> {
         let body = body_from_buf(&bytes);
 
-        let resp = self.inner.post("callback", body).await?;
-
-        debug!("called do_post and got response: {:?}", resp);
-        if !resp.status().is_success() {
-            error!(
-                "error: POST response was not a success {}:{}",
-                file!(),
-                line!()
-            );
-            // TODO: return Err(_)?
-        };
+        debug!("calling mainbrain callback handler");
+        let _resp = self.inner.post("callback", body).await?;
         Ok(())
     }
 
     pub async fn get_remote_info(
         &mut self,
-        orig_cam_name: &flydra_types::RawCamName,
+        raw_cam_name: &flydra_types::RawCamName,
     ) -> Result<flydra_types::RemoteCameraInfoResponse, Error> {
         let path = format!(
-            "{}?camera={}",
-            flydra_types::REMOTE_CAMERA_INFO_PATH,
-            orig_cam_name.as_str()
+            "{}/{}",
+            flydra_types::braid_http::REMOTE_CAMERA_INFO_PATH,
+            flydra_types::braid_http::encode_cam_name(&raw_cam_name)
         );
 
         debug!(
             "Getting remote camera info for camera \"{}\".",
-            orig_cam_name.as_str()
+            raw_cam_name.as_str()
         );
 
         let resp = self.inner.get(&path).await?;
@@ -97,31 +88,10 @@ impl MainbrainSession {
         >(&data)?)
     }
 
-    pub async fn register_flydra_camnode(
+    pub async fn post_callback_message(
         &mut self,
-        msg: &flydra_types::RegisterNewCamera,
+        msg: flydra_types::BraidHttpApiCallback,
     ) -> Result<(), Error> {
-        debug!("register_flydra_camnode with message {:?}", msg);
-        let msg = flydra_types::HttpApiCallback::NewCamera(msg.clone());
-        Ok(self.send_message(msg).await?)
-    }
-
-    pub async fn update_image(
-        &mut self,
-        ros_cam_name: flydra_types::RosCamName,
-        current_image_png: flydra_types::PngImageData,
-    ) -> Result<(), Error> {
-        let msg = flydra_types::PerCam {
-            ros_cam_name,
-            inner: flydra_types::UpdateImage { current_image_png },
-        };
-
-        debug!("update_image with message {:?}", msg);
-        let msg = flydra_types::HttpApiCallback::UpdateCurrentImage(msg);
-        Ok(self.send_message(msg).await?)
-    }
-
-    pub async fn send_message(&mut self, msg: flydra_types::HttpApiCallback) -> Result<(), Error> {
         let bytes = serde_json::to_vec(&msg).unwrap();
         Ok(self.do_post(bytes).await?)
     }

diff --git a/braid-offline/src/lib.rs b/braid-offline/src/lib.rs
@@ -19,7 +19,7 @@ use flydra2::{
     NumberedRawUdpPoint, StreamItem,
 };
 use flydra_types::{
-    CamInfoRow, PerCamSaveData, RawCamName, RosCamName, SyncFno, TrackingParams,
+    CamInfoRow, PerCamSaveData, RawCamName, SyncFno, TrackingParams,
     FEATURE_DETECT_SETTINGS_DIRNAME, IMAGES_DIRNAME,
 };
 use groupby::{AscendingGroupIter, BufferedSortIter};
@@ -334,7 +334,7 @@ where
 
     let all_expected_cameras = recon
         .cam_names()
-        .map(|x| RosCamName::new(x.to_string()))
+        .map(|x| RawCamName::new(x.to_string()))
         .collect();
 
     let signal_all_cams_present = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
@@ -347,12 +347,6 @@ where
         signal_all_cams_synced,
     );
 
-    // Create `stream_cancel::Valve` for shutting everything down. Note this is
-    // `Clone`, so we can (and should) shut down everything with it. Here we let
-    // _quit_trigger drop when it goes out of scope. This is due to use in this
-    // offline context.
-    let (_quit_trigger, valve) = stream_cancel::Valve::new();
-
     let (frame_data_tx, frame_data_rx) = tokio::sync::mpsc::channel(10);
     let frame_data_rx = tokio_stream::wrappers::ReceiverStream::new(frame_data_rx);
     let save_empty_data2d = true;
@@ -368,7 +362,6 @@ where
         cam_manager.clone(),
         Some(recon.clone()),
         metadata_builder.clone(),
-        valve,
     )?;
 
     let images_dirname = data_src.path_starter().join(IMAGES_DIRNAME);
@@ -420,13 +413,13 @@ where
 
     let images_dirname = data_src.path_starter().join(IMAGES_DIRNAME);
 
-    let per_cam_data: BTreeMap<RosCamName, PerCamSaveData> = match images_dirname.list_paths() {
+    let per_cam_data: BTreeMap<RawCamName, PerCamSaveData> = match images_dirname.list_paths() {
         Ok(relnames) => relnames
             .iter()
             .map(|relname| {
                 assert_eq!(relname.extension(), Some(std::ffi::OsStr::new("png")));
-                let ros_cam_name =
-                    RosCamName::new(relname.file_stem().unwrap().to_str().unwrap().to_string());
+                let raw_cam_name =
+                    RawCamName::new(relname.file_stem().unwrap().to_str().unwrap().to_string());
 
                 let png_fname = data_src.path_starter().join(IMAGES_DIRNAME).join(relname);
                 let current_image_png = {
@@ -439,7 +432,7 @@ where
                 let mut current_feature_detect_settings_fname = data_src
                     .path_starter()
                     .join(FEATURE_DETECT_SETTINGS_DIRNAME)
-                    .join(format!("{}.toml", ros_cam_name.as_str()));
+                    .join(format!("{}.toml", raw_cam_name.as_str()));
 
                 let current_feature_detect_settings =
                     if current_feature_detect_settings_fname.exists() {
@@ -452,7 +445,7 @@ where
                     };
 
                 (
-                    ros_cam_name,
+                    raw_cam_name,
                     PerCamSaveData {
                         current_image_png: current_image_png.into(),
                         cam_settings_data: None,
@@ -471,18 +464,17 @@ where
     let mut cam_info_fname = data_src.path_starter();
     cam_info_fname.push(flydra_types::CAM_INFO_CSV_FNAME);
     let cam_info_file = open_maybe_gzipped(cam_info_fname)?;
-    let mut orig_camn_to_cam_name: BTreeMap<flydra_types::CamNum, RosCamName> = BTreeMap::new();
+    let mut orig_camn_to_cam_name: BTreeMap<flydra_types::CamNum, RawCamName> = BTreeMap::new();
     let rdr = csv::Reader::from_reader(cam_info_file);
     for row in rdr.into_deserialize::<CamInfoRow>() {
         let row = row?;
 
         let orig_cam_name = RawCamName::new(row.cam_id.to_string());
-        let ros_cam_name = RosCamName::new(row.cam_id.to_string());
-        let no_server = flydra_types::StrandCamHttpServerInfo::NoServer;
+        let no_server = flydra_types::BuiServerInfo::NoServer;
 
-        orig_camn_to_cam_name.insert(row.camn, ros_cam_name.clone());
+        orig_camn_to_cam_name.insert(row.camn, orig_cam_name.clone());
 
-        cam_manager.register_new_camera(&orig_cam_name, &no_server, &ros_cam_name);
+        cam_manager.register_new_camera(&orig_cam_name, &no_server);
     }
 
     {
@@ -672,21 +664,14 @@ where
 
     let expected_framerate = Some(fps as f32);
 
-    // let model_server_addr = opt.model_server_addr.clone();
-
-    let (_quit_trigger, valve) = stream_cancel::Valve::new();
     let (data_tx, data_rx) = tokio::sync::mpsc::channel(50);
 
     let _model_server = match &opt2.model_server_addr {
         Some(ref addr) => {
             let addr = addr.parse().unwrap();
             info!("send_pose server at {}", addr);
-            let info = flydra_types::StaticMainbrainInfo {
-                name: env!("CARGO_PKG_NAME").into(),
-                version: env!("CARGO_PKG_VERSION").into(),
-            };
             coord_processor.add_listener(data_tx);
-            Some(flydra2::new_model_server(data_rx, valve, &addr, info, rt_handle).await?)
+            Some(flydra2::new_model_server(data_rx, addr).await?)
         }
         None => None,
     };

diff --git a/braid-process-video/Cargo.toml b/braid-process-video/Cargo.toml
@@ -11,7 +11,6 @@ log = "0.4"
 env_logger = "0.10"
 tokio = {version="1.17", features=["macros","rt","tracing"]}
 tokio-stream = "0.1.8"
-stream-cancel = "0.8"
 anyhow = "1"
 chrono = {version="0.4.23", default-features=false, features=["serde", "clock", "std", "wasmbind"]}
 serde = {version="1", features=["derive"]}