diff --git a/src/engine/store/plugin.rs b/src/engine/store/plugin.rs index 9ab4e0c..b891aae 100644 --- a/src/engine/store/plugin.rs +++ b/src/engine/store/plugin.rs @@ -1,5 +1,6 @@ use std::{ marker::PhantomData, + path::PathBuf, process::Stdio, sync::{Arc, Weak}, }; @@ -12,6 +13,7 @@ use pulse_wire::{ }, }; +use serde::Deserialize; use tokio::{ fs, io::{AsyncReadExt, AsyncWriteExt}, @@ -71,6 +73,36 @@ impl Plugin { Ok(()) } + + pub async fn reload(&self, mut child: Child) -> tokio::io::Result<()> { + let mut process = self.process.lock().await; + + process.kill().await?; + + *self.stdout.lock().await = child.stdout.take().expect("Failed to obtain child stdout"); + + *process = child; + + Ok(()) + } +} + +pub fn get_manifest_plugin_pair<'de, M: Deserialize<'de>>( + plugin_dir: &PathBuf, + plugin_path: &PathBuf, + manifest: &'de [u8], +) -> tokio::io::Result<(Child, M)> { + Ok(( + Command::new("bash") + .arg(plugin_path) + .current_dir(plugin_dir) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn()?, + toml::from_slice(manifest) + .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))?, + )) } #[derive(Debug)] @@ -87,35 +119,23 @@ impl StrategyPair { let strategy = pulse_plugin(strategy_id)?; let risk = pulse_plugin(risk_id)?; + let (strategy, strategy_manifest) = get_manifest_plugin_pair( + &strategy, + &strategy.join("strategy.bash"), + &fs::read(strategy.join("strategy.toml")).await?, + )?; + + let (risk, risk_manifest) = get_manifest_plugin_pair( + &risk, + &risk.join("risk.bash"), + &fs::read(risk.join("risk.toml")).await?, + )?; + Ok(Arc::new(Self { - strategy: Plugin::new( - Command::new("bash") - .arg(strategy.join("strategy.bash")) - .current_dir(&strategy) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::inherit()) - .spawn()?, - ), - risk: Plugin::new( - Command::new("bash") - .arg(risk.join("risk.bash")) - .current_dir(&risk) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::inherit()) - .spawn()?, - ), - strategy_manifest: Mutex::new( - toml::from_slice(&fs::read(strategy.join("strategy.toml")).await?).map_err( - |v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()), - )?, - ), - risk_manifest: Mutex::new( - toml::from_slice(&fs::read(risk.join("risk.toml")).await?).map_err(|v| { - tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()) - })?, - ), + strategy: Plugin::new(strategy), + risk: Plugin::new(risk), + strategy_manifest: Mutex::new(strategy_manifest), + risk_manifest: Mutex::new(risk_manifest), })) } } @@ -156,17 +176,21 @@ impl StrategyEngine { loop { match pair.strategy.recv().await? { + None => {} + Some(StrategyMessage::Log(log)) => { engine.terminal_server.log_raw(log).await?; } + + Some(StrategyMessage::Signal(signal)) => {} + + Some(StrategyMessage::SubscribeCandle { symbol, timeframe }) => {} + Some(StrategyMessage::RequestOHLC { symbol, timeframe, count, }) => {} - Some(StrategyMessage::Signal(signal)) => {} - Some(StrategyMessage::SubscribeCandle { symbol, timeframe }) => {} - None => {} } } } @@ -193,28 +217,25 @@ impl StrategyEngine { *self.risk_handle.lock().await = Some(tokio::spawn(async move { engine.run_risk().await })); } - pub async fn reload( - self: &Arc, - strategy_id: &str, - risk_id: &str, - ) -> tokio::io::Result<()> { + pub async fn reload_strategy(self: &Arc, id: &str) -> tokio::io::Result<()> { { if let Some(handle) = &*self.strategy_handle.lock().await { handle.abort(); } - if let Some(handle) = &*self.risk_handle.lock().await { - handle.abort(); - } + let pair = self.pair.lock().await; - { - let pair = self.pair.lock().await.clone(); + let strategy = pulse_plugin(id)?; - pair.strategy.process.lock().await.kill().await?; - pair.risk.process.lock().await.kill().await?; - } + let (child, manifest) = get_manifest_plugin_pair( + &strategy, + &strategy.join("strategy.bash"), + &fs::read(strategy.join("strategy.toml")).await?, + )?; - *self.pair.lock().await = StrategyPair::new(strategy_id, risk_id).await?; + pair.risk.reload(child).await?; + + *pair.strategy_manifest.lock().await = manifest; } self.spawn().await;