diff --git a/clover-hub/src/server/appd/mod.rs b/clover-hub/src/server/appd/mod.rs index 4bc0cb9..871803e 100644 --- a/clover-hub/src/server/appd/mod.rs +++ b/clover-hub/src/server/appd/mod.rs @@ -7,7 +7,10 @@ pub mod docker; pub mod ipc; pub mod models; -use crate::server::appd::ipc::handle_ipc; +use crate::{ + server::appd::ipc::handle_ipc, + utils::configure_zenoh, +}; use self::docker::init_app; use bollard::{ @@ -48,124 +51,136 @@ pub async fn appd_main( info!("Connected to docker on {}!", docker_path.clone()); let docker = Arc::new(docker_conn); - let mut zenoh_config = zenoh::Config::default(); - - zenoh_config.insert_json5("connect/endpoints", "tcp/localhost:6699"); - zenoh_config - .insert_json5( + let config_ret = configure_zenoh(vec![ + ("connect/endpoints", "[\"tcp/localhost:6699\"]"), + ( "timestamping/enabled", r#"{ router: true, peer: true, client: true }"#, - ) - .unwrap(); - - debug!("Connecting to Zenoh..."); - let session = Arc::new(zenoh::open(zenoh_config).await.unwrap()); - debug!("Connected to Zenoh!"); - - let status_publisher = session - .declare_publisher(format!("{MODULE_EVT_ID}/status")) - .cache(CacheConfig::default().max_samples(1)) - .await - .unwrap(); - - let ipc_token = cancellation_tokens.0.clone(); - let ipc_session = session.clone(); - let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); - - let init_session = session.clone(); - let init_store = Arc::new(store.clone()); - let init_docker = docker.clone(); - cancellation_tokens - .0 - .run_until_cancelled(async move { - let mut init_apps = init_store.applications.lock().await; - let mut apps_initialized = 0; - - if init_apps.len() == 0 { - info!("No pre-configured applications to initialize."); - } else { - for (id, spec) in init_apps.iter_mut() { - match init_app(init_docker.clone(), id, spec).await { - Ok(_) => { - apps_initialized += 1; - } - Err(_e) => { - error!( - "Failed to initialize application {} ({})!", - spec.name.clone(), - id.clone() - ); - } - } - - // Update the application state. - init_store - .applications - .lock() + ), + ("mode", "\"client\""), + ]); + + match config_ret { + Ok(zenoh_config) => { + debug!("Connecting to Zenoh..."); + + match zenoh::open(zenoh_config).await { + Ok(raw_session) => { + let session = Arc::new(raw_session); + debug!("Connected to Zenoh!"); + + let status_publisher = session + .declare_publisher(format!("{MODULE_EVT_ID}/status")) + .cache(CacheConfig::default().max_samples(1)) .await - .insert(id.clone(), spec.clone()); - } - } - - if apps_initialized != init_apps.len() { - warn!( - "Initialized {} apps out of {}!", - apps_initialized, - init_apps.len() - ); - status_publisher - .put("ready:incomplete") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - } else { - if apps_initialized != 0 { - info!("Initialized all {} apps!", apps_initialized); - } - status_publisher - .put("ready") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - } - - info!("AppDaemon Ready!"); - }) - .await; - - let cleanup_store = store.clone(); - let cleanup_docker = docker.clone(); - let cleanup_token = cancellation_tokens.0.clone(); - tokio::select! { - _ = cleanup_token.cancelled() => { - ipc_handle.abort(); - - info!("Cleaning up applications..."); - - let mut cleanup_apps = cleanup_store.applications.lock().await; - let mut apps_removed = 0; - - if cleanup_apps.len() == 0 { - info!("No applications to remove."); - } else { - for (id, spec) in cleanup_apps.iter_mut() { - match remove_app(cleanup_docker.clone(), id, spec).await { - Ok(_) => { apps_removed += 1; }, - Err(_e) => { - error!("Failed to remove application {} ({})!", spec.name.clone(), id.clone()); + .unwrap(); + + let ipc_token = cancellation_tokens.0.clone(); + let ipc_session = session.clone(); + let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); + + let init_session = session.clone(); + let init_store = Arc::new(store.clone()); + let init_docker = docker.clone(); + cancellation_tokens + .0 + .run_until_cancelled(async move { + let mut init_apps = init_store.applications.lock().await; + let mut apps_initialized = 0; + + if init_apps.len() == 0 { + info!("No pre-configured applications to initialize."); + } else { + for (id, spec) in init_apps.iter_mut() { + match init_app(init_docker.clone(), id, spec).await { + Ok(_) => { + apps_initialized += 1; + } + Err(_e) => { + error!( + "Failed to initialize application {} ({})!", + spec.name.clone(), + id.clone() + ); + } + } + + // Update the application state. + init_store + .applications + .lock() + .await + .insert(id.clone(), spec.clone()); + } + } + + if apps_initialized != init_apps.len() { + warn!( + "Initialized {} apps out of {}!", + apps_initialized, + init_apps.len() + ); + status_publisher + .put("ready:incomplete") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + } else { + if apps_initialized != 0 { + info!("Initialized all {} apps!", apps_initialized); + } + status_publisher + .put("ready") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + } + + info!("AppDaemon Ready!"); + }) + .await; + + let cleanup_store = store.clone(); + let cleanup_docker = docker.clone(); + let cleanup_token = cancellation_tokens.0.clone(); + tokio::select! { + _ = cleanup_token.cancelled() => { + ipc_handle.abort(); + + info!("Cleaning up applications..."); + + let mut cleanup_apps = cleanup_store.applications.lock().await; + let mut apps_removed = 0; + + if cleanup_apps.len() == 0 { + info!("No applications to remove."); + } else { + for (id, spec) in cleanup_apps.iter_mut() { + match remove_app(cleanup_docker.clone(), id, spec).await { + Ok(_) => { apps_removed += 1; }, + Err(_e) => { + error!("Failed to remove application {} ({})!", spec.name.clone(), id.clone()); + } + } + + // Update the application state. + cleanup_store.applications.lock().await.insert(id.clone(), spec.clone()); + } + } + + if apps_removed != cleanup_apps.len() { + warn!("Only removed {} apps out of {}!", apps_removed, cleanup_apps.len()); + } + + std::mem::drop(store); } } - - // Update the application state. - cleanup_store.applications.lock().await.insert(id.clone(), spec.clone()); + } + Err(err) => { + error!("FATAL: Failed to connect to zenoh due to:\n{err}"); } } - - if apps_removed != cleanup_apps.len() { - warn!("Only removed {} apps out of {}!", apps_removed, cleanup_apps.len()); - } - - std::mem::drop(store); - - cancellation_tokens.1.cancel(); + } + Err(err) => { + error!("FATAL: Failed to write configuration for zenoh connection due to:\n{err}"); } } } @@ -177,5 +192,7 @@ pub async fn appd_main( } } + cancellation_tokens.1.cancel(); + info!("AppD has stopped!"); } diff --git a/clover-hub/src/server/inference_engine/mod.rs b/clover-hub/src/server/inference_engine/mod.rs index 5b04a8e..fc62bd8 100644 --- a/clover-hub/src/server/inference_engine/mod.rs +++ b/clover-hub/src/server/inference_engine/mod.rs @@ -21,7 +21,10 @@ use zenoh_ext::{ CacheConfig, }; -use crate::server::inference_engine::ipc::handle_ipc; +use crate::{ + server::inference_engine::ipc::handle_ipc, + utils::configure_zenoh, +}; use super::warehouse::config::models::Config; @@ -50,53 +53,67 @@ pub async fn inference_engine_main( ) { info!("Starting Inference Engine..."); - let mut zenoh_config = zenoh::Config::default(); - - zenoh_config.insert_json5("connect/endpoints", "tcp/localhost:6699"); - zenoh_config - .insert_json5( + let config_ret = configure_zenoh(vec![ + ("connect/endpoints", "[\"tcp/localhost:6699\"]"), + ( "timestamping/enabled", r#"{ router: true, peer: true, client: true }"#, - ) - .unwrap(); - - debug!("Connecting to Zenoh..."); - let session = Arc::new(zenoh::open(zenoh_config).await.unwrap()); - debug!("Connected to Zenoh!"); - - let status_publisher = session - .declare_publisher(format!("{MODULE_EVT_ID}/status")) - .cache(CacheConfig::default().max_samples(1)) - .await - .unwrap(); - - let ipc_token = cancellation_tokens.0.clone(); - let ipc_session = session.clone(); - let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); - - cancellation_tokens - .0 - .run_until_cancelled(async move { - status_publisher - .put("ready") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - - info!("InferenceEngine Ready!"); - }) - .await; - - let cleanup_token = cancellation_tokens.0.clone(); - tokio::select! { - _ = cleanup_token.cancelled() => { - ipc_handle.abort(); - - info!("Cleaning up networks..."); - // TODO: Clean up registered networks when server is shutting down. - - cancellation_tokens.1.cancel(); + ), + ("mode", "\"client\""), + ]); + + match config_ret { + Ok(zenoh_config) => { + debug!("Connecting to Zenoh..."); + + match zenoh::open(zenoh_config).await { + Ok(raw_session) => { + let session = Arc::new(raw_session); + debug!("Connected to Zenoh!"); + + let status_publisher = session + .declare_publisher(format!("{MODULE_EVT_ID}/status")) + .cache(CacheConfig::default().max_samples(1)) + .await + .unwrap(); + + let ipc_token = cancellation_tokens.0.clone(); + let ipc_session = session.clone(); + let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); + + cancellation_tokens + .0 + .run_until_cancelled(async move { + status_publisher + .put("ready") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + + info!("InferenceEngine Ready!"); + }) + .await; + + let cleanup_token = cancellation_tokens.0.clone(); + tokio::select! { + _ = cleanup_token.cancelled() => { + ipc_handle.abort(); + + info!("Cleaning up networks..."); + // TODO: Clean up registered networks when server is shutting down. + } + } + } + Err(err) => { + error!("FATAL: Failed to connect to zenoh due to:\n{err}"); + } + } + } + Err(err) => { + error!("FATAL: Failed to write configuration for zenoh connection due to:\n{err}"); } } + cancellation_tokens.1.cancel(); + info!("Inference Engine has stopped!"); } diff --git a/clover-hub/src/server/modman/gestures/mod.rs b/clover-hub/src/server/modman/gestures/mod.rs index 04003db..1e08aad 100644 --- a/clover-hub/src/server/modman/gestures/mod.rs +++ b/clover-hub/src/server/modman/gestures/mod.rs @@ -1,8 +1,8 @@ -// TODO: Manifests specify bodies? Allow globs. +// TODO: Manifests specify bodies. Allow globs. // TODO: Each area has default applicable gestures /* Default area specification - - com.reboot-codes.clover.CORE.human + - com.reboot-codes.clover.CORE.humanoid - torso - core - back diff --git a/clover-hub/src/server/modman/mod.rs b/clover-hub/src/server/modman/mod.rs index 7c9b413..e0543c2 100644 --- a/clover-hub/src/server/modman/mod.rs +++ b/clover-hub/src/server/modman/mod.rs @@ -37,9 +37,12 @@ use zenoh_ext::{ CacheConfig, }; -use crate::server::modman::{ - ipc::handle_ipc, - models::Module, +use crate::{ + server::modman::{ + ipc::handle_ipc, + models::Module, + }, + utils::configure_zenoh, }; pub const MODULE_EVT_ID: &str = "com/reboot-codes/clover/hub/modman"; @@ -52,181 +55,195 @@ pub async fn modman_main( ) { info!("Starting ModMan..."); - let mut zenoh_config = zenoh::Config::default(); - - zenoh_config.insert_json5("connect/endpoints", "tcp/localhost:6699"); - zenoh_config - .insert_json5( + let config_ret = configure_zenoh(vec![ + ("connect/endpoints", "[\"tcp/localhost:6699\"]"), + ( "timestamping/enabled", r#"{ router: true, peer: true, client: true }"#, - ) - .unwrap(); - - debug!("Connecting to Zenoh..."); - let session = Arc::new(zenoh::open(zenoh_config).await.unwrap()); - debug!("Connected to Zenoh!"); - - let status_publisher = session - .declare_publisher(format!("{MODULE_EVT_ID}/status")) - .cache(CacheConfig::default().max_samples(1)) - .await - .unwrap(); - - let ipc_token = cancellation_tokens.0.clone(); - let ipc_session = session.clone(); - let ipc_store = store.clone(); - let ipc_handle = tokio::task::spawn(handle_ipc(ipc_store, ipc_token, ipc_session)); - - let bus_store = Arc::new(store.clone()); - let bus_token = cancellation_tokens.0.clone(); - let bus_session = session.clone(); - let bus_handle = tokio::task::spawn(async move { - tokio::select! { - _ = bus_token.cancelled() => { - debug!("bus_handle exited"); - }, - _ = start_busses(bus_store, bus_session) => {} - } - }); - - let init_store = Arc::new(store.clone()); - let init_results = cancellation_tokens - .0 - .run_until_cancelled(async move { - let config = init_store.config.lock().await; - let static_modules = &config.modman.static_modules; - let static_components = &config.modman.static_components; - let mut modules_to_init: HashMap = { - let mut hashmap = HashMap::new(); - let modules = init_store.modules.lock().await; - - for (module_id, module_config) in modules.iter() { - hashmap.insert(module_id.clone(), module_config.clone()); - } - - hashmap - }; - let mut modules_initalized: usize = 0; - - debug!("Checking for statically defined modules to init..."); - if static_modules.len() > 0 { - debug!( - "Adding {} statically defined module(s) to init queue...", - static_modules.len() - ); - for (module_id, module) in static_modules { - modules_to_init.insert(module_id.clone(), module.clone()); - } - } else { - debug!("No statically defined modules to put into store, skipping!"); - } - - debug!("Checking for statically defined components to init..."); - if static_components.len() > 0 { - debug!( - "Adding {} statically defined components(s) to init queue...", - static_components.len() - ); - let mut components = init_store.components.lock().await; - - for (component_id, component) in static_components { - components.insert( - component_id.clone(), - Arc::new((component.0.clone(), component.1.clone())), - ); - } - } else { - debug!("No statically defined components to put into store, skipping!"); - } - - let total_modules = modules_to_init.len(); - - drop(config); - - info!("Initalizing modules..."); - if total_modules > 0 { - // Initialize modules that were registered already via configuration and persistence. - for (id, module) in modules_to_init.iter() { - let (initialized, _components_initialized) = - init_module(&init_store, id.clone(), module.clone()).await; - - if initialized { - modules_initalized += 1; + ), + ("mode", "\"client\""), + ]); + + match config_ret { + Ok(zenoh_config) => { + debug!("Connecting to Zenoh..."); + + match zenoh::open(zenoh_config).await { + Ok(raw_session) => { + let session = Arc::new(raw_session); + debug!("Connected to Zenoh!"); + + let status_publisher = session + .declare_publisher(format!("{MODULE_EVT_ID}/status")) + .cache(CacheConfig::default().max_samples(1)) + .await + .unwrap(); + + let ipc_token = cancellation_tokens.0.clone(); + let ipc_session = session.clone(); + let ipc_store = store.clone(); + let ipc_handle = tokio::task::spawn(handle_ipc(ipc_store, ipc_token, ipc_session)); + + let bus_store = Arc::new(store.clone()); + let bus_token = cancellation_tokens.0.clone(); + let bus_session = session.clone(); + let bus_handle = tokio::task::spawn(async move { + tokio::select! { + _ = bus_token.cancelled() => { + debug!("bus_handle exited"); + }, + _ = start_busses(bus_store, bus_session) => {} + } + }); + + let init_store = Arc::new(store.clone()); + let init_results = cancellation_tokens + .0 + .run_until_cancelled(async move { + let config = init_store.config.lock().await; + let static_modules = &config.modman.static_modules; + let static_components = &config.modman.static_components; + let mut modules_to_init: HashMap = { + let mut hashmap = HashMap::new(); + let modules = init_store.modules.lock().await; + + for (module_id, module_config) in modules.iter() { + hashmap.insert(module_id.clone(), module_config.clone()); + } + + hashmap + }; + let mut modules_initalized: usize = 0; + + debug!("Checking for statically defined modules to init..."); + if static_modules.len() > 0 { + debug!( + "Adding {} statically defined module(s) to init queue...", + static_modules.len() + ); + for (module_id, module) in static_modules { + modules_to_init.insert(module_id.clone(), module.clone()); + } + } else { + debug!("No statically defined modules to put into store, skipping!"); + } + + debug!("Checking for statically defined components to init..."); + if static_components.len() > 0 { + debug!( + "Adding {} statically defined components(s) to init queue...", + static_components.len() + ); + let mut components = init_store.components.lock().await; + + for (component_id, component) in static_components { + components.insert( + component_id.clone(), + Arc::new((component.0.clone(), component.1.clone())), + ); + } + } else { + debug!("No statically defined components to put into store, skipping!"); + } + + let total_modules = modules_to_init.len(); + + drop(config); + + info!("Initalizing modules..."); + if total_modules > 0 { + // Initialize modules that were registered already via configuration and persistence. + for (id, module) in modules_to_init.iter() { + let (initialized, _components_initialized) = + init_module(&init_store, id.clone(), module.clone()).await; + + if initialized { + modules_initalized += 1; + } + } + } else { + info!("No static modules to initialize."); + } + + (modules_initalized, total_modules) + }) + .await; + + if let Some((modules_initalized, total_modules)) = init_results { + if modules_initalized != total_modules { + warn!("Initalized {modules_initalized} out of {total_modules} module(s)!"); + status_publisher + .put("ready:incomplete") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + } else { + if modules_initalized != 0 { + info!("Initalized all {modules_initalized} module(s)"); + } + status_publisher + .put("ready") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + } + + info!("ModMan Ready!"); } - } - } else { - info!("No static modules to initialize."); - } - - (modules_initalized, total_modules) - }) - .await; - - if let Some((modules_initalized, total_modules)) = init_results { - if modules_initalized != total_modules { - warn!("Initalized {modules_initalized} out of {total_modules} module(s)!"); - status_publisher - .put("ready:incomplete") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - } else { - if modules_initalized != 0 { - info!("Initalized all {modules_initalized} module(s)"); - } - status_publisher - .put("ready") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - } - - info!("ModMan Ready!"); - } - let mod_clean_token = cancellation_tokens.0.clone(); - tokio::select! { - _ = mod_clean_token.cancelled() => { - bus_handle.abort(); - ipc_handle.abort(); - drop(status_publisher); - - info!("Cleaning up modules..."); - - // TODO: Add override cancellation token to force stop! - - let modules_snapshot: Vec<(String, Module)> = { - let modules = store.modules.lock().await; - debug!("done waiting for lock"); - modules - .iter() - .filter(|(_, m)| m.initialized) - .map(|(id, m)| (id.clone(), m.clone())) - .collect() - // lock drops here - }; - - let total = modules_snapshot.len(); - let mut modules_deinitalized: usize = 0; - - if total > 0 { - for (id, module) in modules_snapshot { - let (de_initialized, _) = deinit_module(&store, id, module).await; - if de_initialized { modules_deinitalized += 1; } + let mod_clean_token = cancellation_tokens.0.clone(); + tokio::select! { + _ = mod_clean_token.cancelled() => { + bus_handle.abort(); + ipc_handle.abort(); + drop(status_publisher); + + info!("Cleaning up modules..."); + + // TODO: Add override cancellation token to force stop! + + let modules_snapshot: Vec<(String, Module)> = { + let modules = store.modules.lock().await; + debug!("done waiting for lock"); + modules + .iter() + .filter(|(_, m)| m.initialized) + .map(|(id, m)| (id.clone(), m.clone())) + .collect() + // lock drops here + }; + + let total = modules_snapshot.len(); + let mut modules_deinitalized: usize = 0; + + if total > 0 { + for (id, module) in modules_snapshot { + let (de_initialized, _) = deinit_module(&store, id, module).await; + if de_initialized { modules_deinitalized += 1; } + } + + if modules_deinitalized != total { + warn!("Deinitalized {} out of {} module(s)!", modules_deinitalized, total); + } else { + info!("Deinitalized all {} module(s)", modules_deinitalized); + } + } else { + debug!("No modules to deinit."); + } + + std::mem::drop(store); + } + } } - - if modules_deinitalized != total { - warn!("Deinitalized {} out of {} module(s)!", modules_deinitalized, total); - } else { - info!("Deinitalized all {} module(s)", modules_deinitalized); + Err(err) => { + error!("FATAL: Failed to connect to zenoh due to:\n{err}"); } - } else { - debug!("No modules to deinit."); } - - std::mem::drop(store); - - cancellation_tokens.1.cancel(); + } + Err(err) => { + error!("FATAL: Failed to write configuration for zenoh connection due to:\n{err}") } } + cancellation_tokens.1.cancel(); + info!("ModMan has stopped!"); } diff --git a/clover-hub/src/server/renderer/mod.rs b/clover-hub/src/server/renderer/mod.rs index 4b47447..58a243f 100644 --- a/clover-hub/src/server/renderer/mod.rs +++ b/clover-hub/src/server/renderer/mod.rs @@ -16,7 +16,10 @@ use crate::server::{ modman::MODULE_EVT_ID as MODMAN_EVT_ID, renderer::ipc::handle_ipc, }; -use crate::utils::RecvSync; +use crate::utils::{ + configure_zenoh, + RecvSync, +}; use queues::*; use std::sync::Arc; use std::sync::Mutex as StdMutex; @@ -64,200 +67,215 @@ pub async fn renderer_main( ) { info!("Starting Renderer..."); - let mut zenoh_config = zenoh::Config::default(); - - zenoh_config.insert_json5("connect/endpoints", "tcp/localhost:6699"); - zenoh_config - .insert_json5( + let config_ret = configure_zenoh(vec![ + ("connect/endpoints", "[\"tcp/localhost:6699\"]"), + ( "timestamping/enabled", r#"{ router: true, peer: true, client: true }"#, - ) - .unwrap(); - - debug!("Connecting to Zenoh..."); - let session = Arc::new(zenoh::open(zenoh_config).await.unwrap()); - debug!("Connected to Zenoh!"); - - let status_publisher = session - .declare_publisher(format!("{MODULE_EVT_ID}/status")) - .cache(CacheConfig::default().max_samples(1)) - .await - .unwrap(); - - let ipc_token = cancellation_tokens.0.clone(); - let ipc_session = session.clone(); - let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); - - let display_registration_queue = Arc::new(StdMutex::new(queue![])); - - let (bevy_cancel_tx, bevy_cancel_rx) = std::sync::mpsc::channel(); - let system_ui_ipc = SystemUIIPC { - exit_channel: RecvSync(bevy_cancel_rx), - display_registration_queue: display_registration_queue.clone(), - }; - - // TODO: Add this as a CLI/Config option! - let custom_bevy_ipc = system_ui_ipc; - let bevy_thread = std::thread::spawn(move || system_ui_main(custom_bevy_ipc, Some(false))); - - // let display_handles = Arc::new(HashMap::new()); - - let init_session = session.clone(); - cancellation_tokens - .0 - .run_until_cancelled(async move { - debug!("Waiting for ModMan to be ready..."); - - let modman_status_subscriber = init_session - .declare_subscriber(format!("{MODMAN_EVT_ID}/status")) - .history(HistoryConfig::default().detect_late_publishers()) - .recovery(RecoveryConfig::default().heartbeat()) - .await - .unwrap(); - - let modman_ready = loop { - match tokio::time::timeout( - std::time::Duration::from_millis(500), - modman_status_subscriber.recv_async(), - ) - .await - { - Ok(Ok(sample)) => { - let status = sample - .payload() - .try_to_string() - .unwrap_or_else(|e| e.to_string().into()); - debug!("ModMan Status: {status}"); - break status.to_string(); - } - Ok(Err(e)) => { - error!("Subscriber channel error: {e}"); - break "error".to_string(); - } - Err(_) => { - debug!("Timed out waiting for ModMan status, querying cache directly..."); - match init_session.get(format!("{MODMAN_EVT_ID}/status")).await { - Ok(replies) => { - if let Ok(reply) = replies.recv_async().await { - if let Ok(sample) = reply.into_result() { + ), + ("mode", "\"client\""), + ]); + + match config_ret { + Ok(zenoh_config) => { + debug!("Connecting to Zenoh..."); + + match zenoh::open(zenoh_config).await { + Ok(raw_session) => { + let session = Arc::new(raw_session); + debug!("Connected to Zenoh!"); + + let status_publisher = session + .declare_publisher(format!("{MODULE_EVT_ID}/status")) + .cache(CacheConfig::default().max_samples(1)) + .await + .unwrap(); + + let ipc_token = cancellation_tokens.0.clone(); + let ipc_session = session.clone(); + let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); + + let display_registration_queue = Arc::new(StdMutex::new(queue![])); + + let (bevy_cancel_tx, bevy_cancel_rx) = std::sync::mpsc::channel(); + let system_ui_ipc = SystemUIIPC { + exit_channel: RecvSync(bevy_cancel_rx), + display_registration_queue: display_registration_queue.clone(), + }; + + // TODO: Add this as a CLI/Config option! + let custom_bevy_ipc = system_ui_ipc; + let bevy_thread = + std::thread::spawn(move || system_ui_main(custom_bevy_ipc, Some(false))); + + // let display_handles = Arc::new(HashMap::new()); + + let init_session = session.clone(); + cancellation_tokens + .0 + .run_until_cancelled(async move { + debug!("Waiting for ModMan to be ready..."); + + let modman_status_subscriber = init_session + .declare_subscriber(format!("{MODMAN_EVT_ID}/status")) + .history(HistoryConfig::default().detect_late_publishers()) + .recovery(RecoveryConfig::default().heartbeat()) + .await + .unwrap(); + + let modman_ready = loop { + match tokio::time::timeout( + std::time::Duration::from_millis(500), + modman_status_subscriber.recv_async(), + ) + .await + { + Ok(Ok(sample)) => { let status = sample .payload() .try_to_string() .unwrap_or_else(|e| e.to_string().into()); - debug!("ModMan Status (from cache query): {status}"); + debug!("ModMan Status: {status}"); break status.to_string(); } + Ok(Err(e)) => { + error!("Subscriber channel error: {e}"); + break "error".to_string(); + } + Err(_) => { + debug!("Timed out waiting for ModMan status, querying cache directly..."); + match init_session.get(format!("{MODMAN_EVT_ID}/status")).await { + Ok(replies) => { + if let Ok(reply) = replies.recv_async().await { + if let Ok(sample) = reply.into_result() { + let status = sample + .payload() + .try_to_string() + .unwrap_or_else(|e| e.to_string().into()); + debug!("ModMan Status (from cache query): {status}"); + break status.to_string(); + } + } + debug!("No cached ModMan status yet, retrying..."); + } + Err(e) => { + error!("Failed to query ModMan status: {e}"); + break "error".to_string(); + } + } + } } - debug!("No cached ModMan status yet, retrying..."); - } - Err(e) => { - error!("Failed to query ModMan status: {e}"); - break "error".to_string(); - } - } - } - } - }; + }; - drop(modman_status_subscriber); + drop(modman_status_subscriber); - if modman_ready == "error" { - error!("Failed to get ModMan ready status!"); - status_publisher - .put("ready:incomplete") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - return; - } - - info!("Requesting displays to setup in SystemUI from ModMan..."); - - let displays_payload = loop { - match init_session - .get(format!( - "{MODMAN_EVT_ID}/components/by-type/video/displays/all" - )) - .await - { - Ok(reply_fifo) => match reply_fifo.recv_async().await { - Ok(reply) => match reply.result() { - Ok(sample) => { - break Ok( - sample - .payload() - .try_to_string() - .unwrap_or_else(|e| e.to_string().into()) - .to_string(), - ); + if modman_ready == "error" { + error!("Failed to get ModMan ready status!"); + status_publisher + .put("ready:incomplete") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + return; } - Err(e) => { - break Err( - e.payload() - .try_to_string() - .unwrap_or_else(|e| e.to_string().into()) - .to_string(), - ); + + info!("Requesting displays to setup in SystemUI from ModMan..."); + + let displays_payload = loop { + match init_session + .get(format!( + "{MODMAN_EVT_ID}/components/by-type/video/displays/all" + )) + .await + { + Ok(reply_fifo) => match reply_fifo.recv_async().await { + Ok(reply) => match reply.result() { + Ok(sample) => { + break Ok( + sample + .payload() + .try_to_string() + .unwrap_or_else(|e| e.to_string().into()) + .to_string(), + ); + } + Err(e) => { + break Err( + e.payload() + .try_to_string() + .unwrap_or_else(|e| e.to_string().into()) + .to_string(), + ); + } + }, + Err(e) => { + debug!("displays/get queryable not ready yet ({e}), retrying in 100ms..."); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + }, + Err(e) => { + error!("Unable to setup Querier:\n{e}"); + break Err(e.to_string()); + } + } + }; + + match displays_payload { + Ok(payload) => { + debug!("Got displays: {:?}", payload); + status_publisher + .put("ready") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + } + Err(payload) => { + error!("Got error payload:\n{payload}"); + status_publisher + .put("ready:incomplete") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + } } - }, - Err(e) => { - debug!("displays/get queryable not ready yet ({e}), retrying in 100ms..."); - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - } - }, - Err(e) => { - error!("Unable to setup Querier:\n{e}"); - break Err(e.to_string()); - } - } - }; - match displays_payload { - Ok(payload) => { - debug!("Got displays: {:?}", payload); - status_publisher - .put("ready") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - } - Err(payload) => { - error!("Got error payload:\n{payload}"); - status_publisher - .put("ready:incomplete") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - } - } + info!("Renderer Ready!"); + }) + .await; - info!("Renderer Ready!"); - }) - .await; + let cleanup_token = cancellation_tokens.0.clone(); + tokio::select! { + _ = cleanup_token.cancelled() => { + ipc_handle.abort(); - let cleanup_token = cancellation_tokens.0.clone(); - tokio::select! { - _ = cleanup_token.cancelled() => { - ipc_handle.abort(); + info!("Cleaning up displays..."); - info!("Cleaning up displays..."); + let _ = bevy_cancel_tx.send(bevy::prelude::AppExit::Success); - let _ = bevy_cancel_tx.send(bevy::prelude::AppExit::Success); + info!("Waiting for Bevy to shut down..."); + match bevy_thread.join() { + Ok(_) => { + info!("Bevy thread exited cleanly"); + } + Err(e) => { + error!("Bevy thread panicked during shutdown: {:?}", e); + } + } + + // TODO: Clean up registered displays when server is shutting down. - info!("Waiting for Bevy to shut down..."); - match bevy_thread.join() { - Ok(_) => { - info!("Bevy thread exited cleanly"); + std::mem::drop(store); + } + } } - Err(e) => { - error!("Bevy thread panicked during shutdown: {:?}", e); + Err(err) => { + error!("FATAL: Failed to connect to zenoh due to:\n{err}"); } } - - // TODO: Clean up registered displays when server is shutting down. - - std::mem::drop(store); - - cancellation_tokens.1.cancel(); + } + Err(err) => { + error!("FATAL: Failed to write configuration for zenoh connection due to:\n{err}") } } + cancellation_tokens.1.cancel(); + info!("Renderer has stopped!"); } diff --git a/clover-hub/src/server/warehouse/mod.rs b/clover-hub/src/server/warehouse/mod.rs index 8b79baa..70bb3a7 100644 --- a/clover-hub/src/server/warehouse/mod.rs +++ b/clover-hub/src/server/warehouse/mod.rs @@ -41,6 +41,7 @@ use zenoh_ext::{ }; use crate::server::warehouse::ipc::handle_ipc; +use crate::utils::configure_zenoh; /// The primary startup Error enum. /// Warehouse will return this enum in [`setup_warehouse`]. @@ -231,105 +232,119 @@ pub async fn warehouse_main( ) { info!("Starting Warehouse..."); - let mut zenoh_config = zenoh::Config::default(); - - zenoh_config.insert_json5("connect/endpoints", "tcp/localhost:6699"); - zenoh_config - .insert_json5( + let config_ret = configure_zenoh(vec![ + ("connect/endpoints", "[\"tcp/localhost:6699\"]"), + ( "timestamping/enabled", r#"{ router: true, peer: true, client: true }"#, - ) - .unwrap(); - - debug!("Connecting to Zenoh..."); - let session = Arc::new(zenoh::open(zenoh_config).await.unwrap()); - debug!("Connected to Zenoh!"); - - let status_publisher = session - .declare_publisher(format!("{MODULE_EVT_ID}/status")) - .cache(CacheConfig::default().max_samples(1)) - .await - .unwrap(); - - let ipc_token = cancellation_tokens.0.clone(); - let ipc_session = session.clone(); - let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); - - // TODO: Move to Zenoh's persistent storage - let db_raw = Database::connect(format!( - "sqlite://{}?mode=rwc", - store.config.lock().await.data_dir.join("/db.sqlite") - )) - .await; - - let init_store = Arc::new(store.clone()); - let init_tokens = cancellation_tokens.clone(); - cancellation_tokens - .0 - .run_until_cancelled(async move { - match db_raw { - Ok(db) => { - init_store.config.lock().await.db = Some(Arc::new(db)); - } - Err(e) => { - error!("Failed to access db file, due to:\n{}", e); - init_tokens.0.cancel(); - } - } - - status_publisher - .put("ready") - .await - .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); - - info!("Warehouse Ready!"); - }) - .await; - - let cleanup_token = cancellation_tokens.0.clone(); - tokio::select! { - _ = cleanup_token.cancelled() => { - ipc_handle.abort(); - - info!("Buttoning up storage..."); - // TODO: Lock db and clean up when done. - debug!("Writing Config File..."); - let config = store.config.lock().await; - match fs::File::open(config.data_dir.join("/config.json")).await { - Ok(mut config_file) => { - debug!("Config file opened!"); - - match config_file - .write_all( - serde_json_lenient::to_string_pretty::(&config.clone() as &Config) - .unwrap() - .as_bytes(), - ) + ), + ("mode", "\"client\""), + ]); + + match config_ret { + Ok(zenoh_config) => { + debug!("Connecting to Zenoh..."); + + match zenoh::open(zenoh_config).await { + Ok(raw_session) => { + let session = Arc::new(raw_session); + debug!("Connected to Zenoh!"); + + let status_publisher = session + .declare_publisher(format!("{MODULE_EVT_ID}/status")) + .cache(CacheConfig::default().max_samples(1)) .await - { - Ok(_) => { - debug!("Wrote config from store state!"); - } - Err(e) => { - error!("Failed to write config file"); - // TODO! + .unwrap(); + + let ipc_token = cancellation_tokens.0.clone(); + let ipc_session = session.clone(); + let ipc_handle = tokio::task::spawn(handle_ipc(ipc_token, ipc_session)); + + // TODO: Move to Zenoh's persistent storage + let db_raw = Database::connect(format!( + "sqlite://{}?mode=rwc", + store.config.lock().await.data_dir.join("/db.sqlite") + )) + .await; + + let init_store = Arc::new(store.clone()); + let init_tokens = cancellation_tokens.clone(); + cancellation_tokens + .0 + .run_until_cancelled(async move { + match db_raw { + Ok(db) => { + init_store.config.lock().await.db = Some(Arc::new(db)); + } + Err(e) => { + error!("Failed to access db file, due to:\n{}", e); + init_tokens.0.cancel(); + } + } + + status_publisher + .put("ready") + .await + .unwrap_or_else(|e| error!("Failed to publish status due to:\n{e}")); + + info!("Warehouse Ready!"); + }) + .await; + + let cleanup_token = cancellation_tokens.0.clone(); + tokio::select! { + _ = cleanup_token.cancelled() => { + ipc_handle.abort(); + + info!("Buttoning up storage..."); + // TODO: Lock db and clean up when done. + debug!("Writing Config File..."); + let config = store.config.lock().await; + match fs::File::open(config.data_dir.join("/config.json")).await { + Ok(mut config_file) => { + debug!("Config file opened!"); + + match config_file + .write_all( + serde_json_lenient::to_string_pretty::(&config.clone() as &Config) + .unwrap() + .as_bytes(), + ) + .await + { + Ok(_) => { + debug!("Wrote config from store state!"); + } + Err(e) => { + error!("Failed to write config file"); + // TODO! + } + } + + std::mem::drop(config_file); + } + Err(e) => { + error!("Unable to open config file"); + // TODO! + } + } + + std::mem::drop(config); + std::mem::drop(store); } } - - std::mem::drop(config_file); } - Err(e) => { - error!("Unable to open config file"); - // TODO! + Err(err) => { + error!("FATAL: Failed to connect to zenoh due to:\n{err}"); } } - - std::mem::drop(config); - std::mem::drop(store); - - cancellation_tokens.1.cancel(); + } + Err(err) => { + error!("FATAL: Failed to write configuration for zenoh connection due to:\n{err}") } } + cancellation_tokens.1.cancel(); + info!("Warehouse has stopped!"); } diff --git a/clover-hub/src/utils.rs b/clover-hub/src/utils.rs index b0e17bd..9a7f7de 100644 --- a/clover-hub/src/utils.rs +++ b/clover-hub/src/utils.rs @@ -114,3 +114,30 @@ where }, } } + +pub fn configure_zenoh( + config_steps: Vec<(&str, &str)>, +) -> Result> { + let mut zenoh_config = zenoh::Config::default(); + + let mut config_ret = None; + + for config_step in config_steps { + match zenoh_config.insert_json5(config_step.0, config_step.1) { + Ok(_) => {} + Err(err) => { + config_ret = Some(Err(err)); + break; + } + } + } + + match config_ret { + Some(err) => { + return err; + } + None => { + return Ok(zenoh_config); + } + } +} diff --git a/core/modules/2-phase-tail/README.md b/core/modules/2-phase-tail/README.md new file mode 100644 index 0000000..3b8a6d9 --- /dev/null +++ b/core/modules/2-phase-tail/README.md @@ -0,0 +1 @@ +# C.O.R.E. 2-Phase Tail diff --git a/core/modules/3-phase-tail/module.clover.json b/core/modules/2-phase-tail/module.clover.jsonc similarity index 63% rename from core/modules/3-phase-tail/module.clover.json rename to core/modules/2-phase-tail/module.clover.jsonc index 92f8156..8b7bf80 100644 --- a/core/modules/3-phase-tail/module.clover.json +++ b/core/modules/2-phase-tail/module.clover.jsonc @@ -1,10 +1,8 @@ { // Display name when running through config flows - "name": "C.O.R.E. 3-Phase Tail", - "location": [ - "hips", - "external" - ], + "name": "C.O.R.E. 2-Phase Tail", + "location": "@core.humanoid.pelvis.waist", + "internal": false, // define all components. they will be available as the next url segment after com.reboot-codes.clover.CORE.3-phase-tail.components "components": { // EG: com.reboot-codes.clover.CORE.3-phase-tail.components.phase-1-motor-1 for this component @@ -14,26 +12,16 @@ // Data type, in this case, each message must contain 2 floats. "input": "vec2d", // Data type, this phase has a position reporting function and will send a 2D vector back regularly - "output": "vec2d" + "output": "vec2d", }, "phase-2": { "type": "movement", "input": "vec2d", "output": "vec2d" }, - "phase-3": { - "type": "movement", - "input": "vec2d", - "output": "vec2d" - }, "tip-light": { - "type": "sensor", - "input": "RGBA", - "output": { - // Clover has to request updates for this output from the module explicitly. - "timing": "on-request", - "format": "RGBA" - } + "type": "indicator", + "input": "rgba", } } } diff --git a/core/modules/blinkie/README.md b/core/modules/blinkie/README.md new file mode 100644 index 0000000..aacdbb6 --- /dev/null +++ b/core/modules/blinkie/README.md @@ -0,0 +1,3 @@ +# C.O.R.E. Blinkie Example + +One of the simplest possible modules, a single dev board with a neopixel on it, connected via USB (UART) to Clover. diff --git a/core/modules/blinkie/module.clover.jsonc b/core/modules/blinkie/module.clover.jsonc new file mode 100644 index 0000000..6d3869e --- /dev/null +++ b/core/modules/blinkie/module.clover.jsonc @@ -0,0 +1,10 @@ +{ + "name": "C.O.R.E. Blinkie Example", + "location": "@core.humanoid.chest", + "components": { + "light": { + "type": "indicator", + "input": "rgba" + } + } +}