single-user service manager
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500pub mod service;pub mod task;
use crate::apogee_core_v1;use crate::apogee_core_v1::apogee_service_logs_v1;use crate::{condition, ecs};use bevy_ecs::prelude::*;use std::collections::{BTreeMap, HashSet, VecDeque};use std::{process, sync, time};
#[derive(Message)]pub struct ServiceOutput { pub stream: apogee_core_v1::LogStream, pub process: ProcessId, pub line: sync::Arc<str>,}
pub fn store_service_output( mut reader: MessageReader<ServiceOutput>, mut services: Query< (&mut Logs, &Process, &LogObjects), (With<service::Service>, Without<task::Task>), >, mut tasks: Query< (&mut Logs, &Process, &LogObjects), (With<task::Task>, Without<service::Service>), >,) { for message in reader.read() { if let Some((mut logs, process, log_objects)) = services .iter_mut() .find(|(_, process, _)| process.id == message.process) .or_else(|| { tasks .iter_mut() .find(|(_, process, _)| process.id == message.process) }) { logs.push(process.id, sync::Arc::clone(&message.line), message.stream); for log_object in &log_objects.0 { log_object.send_entry(process.id.0, message.stream, &message.line); } } }}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub struct ProcessId(pub u32);
#[derive(Component, Debug, Default, PartialEq, Eq)]pub struct LogObjects(Vec<apogee_service_logs_v1::ApogeeServiceLogsV1>);
#[derive(Component, Default)]pub struct Timestamp(pub u64);
pub fn unix_timestamp() -> u64 { time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |duration| duration.as_secs())}
#[derive(Component)]pub struct Name(pub sync::Arc<str>);
#[derive(Component)]pub struct Description(pub String);
#[derive(Component)]pub struct Command(pub String);
#[derive(Component)]pub struct Args(pub Vec<String>);
#[derive(Component)]pub struct Conditions(pub Vec<Entity>);
#[derive(Component)]pub struct Running(pub Entity);
#[derive(Component)]pub struct Environment(pub BTreeMap<String, String>);
pub struct LogEntry { process: ProcessId, message: sync::Arc<str>, stream: apogee_core_v1::LogStream,}
#[derive(Component, Default)]pub struct Logs(pub VecDeque<LogEntry>);
impl Logs { pub fn push( &mut self, process: ProcessId, line: sync::Arc<str>, stream: apogee_core_v1::LogStream, ) { if self.0.len() == 10 { self.0.pop_front(); }
self.0.push_back(LogEntry { process, message: line, stream, }); }}
#[derive(Component)]pub struct Process { pub id: ProcessId, pub child: process::Child, pub token: calloop::RegistrationToken, pub notify_token: Option<calloop::RegistrationToken>,}
impl Process { pub fn remove_notify_source(&mut self, loop_handle: &ecs::LoopHandle) { if let Some(token) = self.notify_token.take() { loop_handle.0.remove(token); } }
pub fn remove_sources(&mut self, loop_handle: &ecs::LoopHandle) { loop_handle.0.remove(self.token); self.remove_notify_source(loop_handle); }}
fn wire_restart_delay(delay: u64) -> u32 { u32::try_from(delay).unwrap_or_else(|_| { log::warn!( "restart delay {delay} exceeds the Hyprwire uint range; advertising {}", u32::MAX ); u32::MAX })}
fn wire_timestamp(timestamp: u64) -> u32 { u32::try_from(timestamp).unwrap_or(u32::MAX)}
#[allow(clippy::panic)]fn condition_name(entity: Entity, names: &Query<&condition::Name>) -> sync::Arc<str> { sync::Arc::clone( &names .get(entity) .unwrap_or_else(|error| panic!("condition {entity:?} is missing its name: {error}")) .0, )}
fn condition_names(entities: &[Entity], names: &Query<&condition::Name>) -> Vec<sync::Arc<str>> { entities .iter() .map(|entity| condition_name(*entity, names)) .collect()}
pub fn sync_objects( mut removed_processes: RemovedComponents<Process>, names: Query<&condition::Name>, mut services: Query< ( Entity, Ref<Name>, Ref<Description>, Ref<Command>, Ref<Args>, Ref<Conditions>, Ref<service::Ready>, Ref<Running>, Ref<Environment>, Ref<service::RestartDelay>, Ref<apogee_core_v1::RestartPolicy>, Ref<service::RestartCount>, Ref<apogee_core_v1::ServiceState>, (Ref<Timestamp>, Option<Ref<Process>>), &mut service::Objects, ), (With<service::Service>, Without<task::Task>), >, mut tasks: Query< ( Entity, Ref<Name>, Ref<Description>, Ref<Command>, Ref<Args>, Ref<Conditions>, Ref<task::Successful>, Ref<Running>, Ref<Environment>, Ref<apogee_core_v1::TaskState>, (Ref<Timestamp>, Option<Ref<Process>>), &mut task::Objects, ), (With<task::Task>, Without<service::Service>), >,) { let removed_processes = removed_processes.read().collect::<HashSet<_>>();
for ( entity, name, description, command, args, conditions, ready, running, environment, restart_delay, restart_policy, restart_count, state, (timestamp, process), mut objects, ) in &mut services { objects .bypass_change_detection() .0 .retain(|object| object.try_client().is_some()); if objects.0.is_empty() { continue; }
if objects.is_changed() { let condition_names = condition_names(&conditions.0, &names); let ready_name = condition_name(ready.0, &names); let running_name = condition_name(running.0, &names); let environment_names = environment.0.keys().map(String::as_str).collect::<Vec<_>>(); let environment_values = environment .0 .values() .map(String::as_str) .collect::<Vec<_>>();
for object in &objects.0 { object.send_name(&name.0); object.send_command(&command.0); object.send_user(rustix::process::getuid().to_string()); object.send_group(rustix::process::getgid().to_string()); object.send_args(&args.0); object.send_conditions(&condition_names); object.send_ready(&ready_name); object.send_running(&running_name); object.send_environment(&environment_names, &environment_values); object.send_restart_delay(wire_restart_delay(restart_delay.0)); object.send_restart_policy(*restart_policy); object.send_restart_count(restart_count.0); object.send_description(&description.0); object.send_state(*state); if let Some(process) = &process { object.send_pid(process.id.0); object.send_process_started(wire_timestamp(timestamp.0)); } else { object.send_process_stopped(); } object.send_done(); } continue; }
let resolved_conditions = conditions .is_changed() .then(|| condition_names(&conditions.0, &names)); let ready_name = ready.is_changed().then(|| condition_name(ready.0, &names)); let running_name = running .is_changed() .then(|| condition_name(running.0, &names)); let environment_names = environment .is_changed() .then(|| environment.0.keys().map(String::as_str).collect::<Vec<_>>()); let environment_values = environment.is_changed().then(|| { environment .0 .values() .map(String::as_str) .collect::<Vec<_>>() }); let process_changed = removed_processes.contains(&entity) || process .as_ref() .is_some_and(bevy_ecs::change_detection::DetectChanges::is_changed) || timestamp.is_changed();
for object in &objects.0 { if name.is_changed() { object.send_name(&name.0); } if description.is_changed() { object.send_description(&description.0); } if command.is_changed() { object.send_command(&command.0); } if args.is_changed() { object.send_args(&args.0); } if let Some(conditions) = &resolved_conditions { object.send_conditions(conditions); } if let Some(ready_name) = &ready_name { object.send_ready(ready_name); } if let Some(running_name) = &running_name { object.send_running(running_name); } if let (Some(names), Some(values)) = (&environment_names, &environment_values) { object.send_environment(names, values); } if restart_delay.is_changed() { object.send_restart_delay(wire_restart_delay(restart_delay.0)); } if restart_policy.is_changed() { object.send_restart_policy(*restart_policy); } if restart_count.is_changed() { object.send_restart_count(restart_count.0); } if state.is_changed() { object.send_state(*state); } if process_changed { if let Some(process) = &process { object.send_pid(process.id.0); object.send_process_started(wire_timestamp(timestamp.0)); } else { object.send_process_stopped(); } } object.send_done(); } }
for ( entity, name, description, command, args, conditions, successful, running, environment, state, (timestamp, process), mut objects, ) in &mut tasks { objects .bypass_change_detection() .0 .retain(|object| object.try_client().is_some()); if objects.0.is_empty() { continue; }
if objects.is_changed() { let condition_names = condition_names(&conditions.0, &names); let ready_name = condition_name(successful.0, &names); let running_name = condition_name(running.0, &names); let environment_names = environment.0.keys().map(String::as_str).collect::<Vec<_>>(); let environment_values = environment .0 .values() .map(String::as_str) .collect::<Vec<_>>();
for object in &objects.0 { object.send_name(&name.0); object.send_command(&command.0); object.send_user(rustix::process::getuid().to_string()); object.send_group(rustix::process::getgid().to_string()); object.send_args(&args.0); object.send_conditions(&condition_names); object.send_successful(&ready_name); object.send_running(&running_name); object.send_environment(&environment_names, &environment_values); object.send_description(&description.0); object.send_state(*state); if let Some(process) = &process { object.send_pid(process.id.0); object.send_process_started(wire_timestamp(timestamp.0)); } else { object.send_process_stopped(); } object.send_done(); } continue; }
let resolved_conditions = conditions .is_changed() .then(|| condition_names(&conditions.0, &names)); let ready_name = successful .is_changed() .then(|| condition_name(successful.0, &names)); let running_name = running .is_changed() .then(|| condition_name(running.0, &names)); let environment_names = environment .is_changed() .then(|| environment.0.keys().map(String::as_str).collect::<Vec<_>>()); let environment_values = environment.is_changed().then(|| { environment .0 .values() .map(String::as_str) .collect::<Vec<_>>() }); let process_changed = removed_processes.contains(&entity) || process .as_ref() .is_some_and(bevy_ecs::change_detection::DetectChanges::is_changed) || timestamp.is_changed();
for object in &objects.0 { if name.is_changed() { object.send_name(&name.0); } if description.is_changed() { object.send_description(&description.0); } if command.is_changed() { object.send_command(&command.0); } if args.is_changed() { object.send_args(&args.0); } if let Some(conditions) = &resolved_conditions { object.send_conditions(conditions); } if let Some(ready_name) = &ready_name { object.send_successful(ready_name); } if let Some(running_name) = &running_name { object.send_running(running_name); } if let (Some(names), Some(values)) = (&environment_names, &environment_values) { object.send_environment(names, values); } if state.is_changed() { object.send_state(*state); } if process_changed { if let Some(process) = &process { object.send_pid(process.id.0); object.send_process_started(wire_timestamp(timestamp.0)); } else { object.send_process_stopped(); } } object.send_done(); } }}
#[derive(Event)]pub struct NewLogObject { entity: Entity, object: apogee_service_logs_v1::ApogeeServiceLogsV1,}
pub fn handle_new_log_object( new_log_object: On<NewLogObject>, mut services: Query<(&Logs, &mut LogObjects), (With<service::Service>, Without<task::Task>)>, mut tasks: Query<(&Logs, &mut LogObjects), (With<task::Task>, Without<service::Service>)>,) { let (logs, mut log_objects) = if let Ok(value) = services.get_mut(new_log_object.entity) { value } else if let Ok(value) = tasks.get_mut(new_log_object.entity) { value } else { return; };
for log in &logs.0 { new_log_object .object .send_entry(log.process.0, log.stream, &log.message); } log_objects.0.push(new_log_object.object.clone()); new_log_object.object.send_done();}
hyprwire::delegate_noop!(crate::Apogee: apogee_service_logs_v1::ApogeeServiceLogsV1);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum StopReason { Exited, Restarted, Removed,}