git.delta.rocks / fleet / refs/commits / 48d120aa4aef

1use std::ffi::OsStr;2use std::process::Stdio;3use std::sync::Mutex;45use bifrostlink::{Rpc, Rtt, WeakRpc};6use tokio::process::{Child, Command};78use remowt_link_shared::plugin::{Error, PluginEndpoints, PluginHost};9use remowt_link_shared::port::child_port;10use remowt_link_shared::{Address, BifConfig};11use tokio::select;1213pub fn serve(rpc: &mut Rpc<BifConfig>) {14	let host = Host {15		me: rpc.me(),16		rpc: rpc.clone().downgrade(),17		children: Mutex::new(Vec::new()),18	};19	PluginEndpoints(host).register_endpoints(rpc);20}2122struct Host {23	me: Address,24	rpc: WeakRpc<BifConfig>,25	children: Mutex<Vec<Child>>,26}2728impl Host {29	async fn spawn(&self, id: u16, path: impl AsRef<OsStr>) -> Result<(), Error> {30		let rpc = self.rpc.clone().upgrade().ok_or(Error::Gone)?;3132		let mut child = Command::new(path)33			.arg(id.to_string())34			.arg(serde_json::to_string(&self.me).expect("address serializes"))35			.stdin(Stdio::piped())36			.stdout(Stdio::piped())37			.kill_on_drop(true)38			.spawn()39			.map_err(|e| Error::Spawn(e.to_string()))?;40		let stdin = child.stdin.take().expect("stdin piped");41		let stdout = child.stdout.take().expect("stdout piped");4243		let addr = Address::Plugin(id);44		rpc.add_direct(addr.clone(), child_port(stdout, stdin), Rtt(0));4546		select! {47			e = rpc.wait_for_connection_to(addr) => {48				if e.is_err() {49					return Err(Error::ConnectionWaitError)50				}51			},52			_ = child.wait() => {53				return Err(Error::ConnectionWaitError)54			}55		};56		self.children.lock().expect("not poisoned").push(child);5758		Ok(())59	}60}6162impl PluginHost for Host {63	async fn load_plugin(&self, id: u16, name: String) -> Result<(), Error> {64		// TODO: Right now loads plugin next to the binary...65		// But with our CA addressed schema, the plugins should be located in content-addressed subdir...66		// Maybe it should just be scrapped in favor of load_plugin_path.67		if name.is_empty() || name == "." || name == ".." || name.contains(['/', '\0']) {68			return Err(Error::BadName);69		}70		let exe = std::env::current_exe().map_err(|e| Error::Spawn(e.to_string()))?;71		let dir = exe72			.parent()73			.ok_or_else(|| Error::Spawn("primary agent has no parent directory".to_owned()))?;74		self.spawn(id, dir.join(&name)).await75	}7677	async fn load_plugin_path(&self, id: u16, path: String) -> Result<(), Error> {78		if path.is_empty() || path.contains('\0') {79			return Err(Error::BadName);80		}81		self.spawn(id, path).await82	}83}