Improved reload

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<S: PulseWire, R: PulseWire> Plugin<S, R> {
 
         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(
+            strategy: Plugin::new(strategy),
-                Command::new("bash")
+            risk: Plugin::new(risk),
-                    .arg(strategy.join("strategy.bash"))
+            strategy_manifest: Mutex::new(strategy_manifest),
-                    .current_dir(&strategy)
+            risk_manifest: Mutex::new(risk_manifest),
-                    .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())
-                })?,
-            ),
         }))
     }
 }
@@ -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(
+    pub async fn reload_strategy(self: &Arc<Self>, id: &str) -> tokio::io::Result<()> {
-        self: &Arc<Self>,
-        strategy_id: &str,
-        risk_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 {
+            let pair = self.pair.lock().await;
-                handle.abort();
-            }
 
-            {
+            let strategy = pulse_plugin(id)?;
-                let pair = self.pair.lock().await.clone();
 
-                pair.strategy.process.lock().await.kill().await?;
+            let (child, manifest) = get_manifest_plugin_pair(
-                pair.risk.process.lock().await.kill().await?;
+                &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;