Improved reloading

src/engine/main.rs

@@ -20,18 +20,5 @@ async fn main() -> tokio::io::Result<()> {
     terminal_server??;
     broadcaster??;
 
-    let strategy = engine.strategy.strategy_handle.lock().await.take();
-    let risk = engine.strategy.risk_handle.lock().await.take();
-
-    drop(engine);
-
-    if let Some(strategy) = strategy {
-        strategy.into_future().await??;
-    }
-
-    if let Some(risk) = risk {
-        risk.into_future().await??;
-    }
-
     Ok(())
 }

src/engine/store/plugin.rs

@@ -19,23 +19,26 @@ use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     process::{Child, ChildStdout, Command},
     sync::Mutex,
-    task::JoinHandle,
 };
 
 use crate::{engine::Engine, store::pulse_plugin};
 
 #[derive(Debug)]
-pub struct Plugin<S: PulseWire, R: PulseWire> {
+pub struct Plugin<S: PulseWire, R: PulseWire, M: for<'de> Deserialize<'de>> {
+    pub manifest: Mutex<M>,
     pub stdout: Mutex<ChildStdout>,
     pub process: Mutex<Child>,
+
     pub _p: (PhantomData<S>, PhantomData<R>),
 }
 
-impl<S: PulseWire, R: PulseWire> Plugin<S, R> {
-    pub fn new(mut child: Child) -> Self {
+impl<S: PulseWire, R: PulseWire, M: for<'de> Deserialize<'de>> Plugin<S, R, M> {
+    pub fn new(mut child: Child, manifest: M) -> Self {
         Self {
             stdout: Mutex::new(child.stdout.take().expect("Failed to obtain child stdout")),
             process: Mutex::new(child),
+            manifest: Mutex::new(manifest),
+
             _p: (PhantomData, PhantomData),
         }
     }
@@ -74,13 +77,13 @@ impl<S: PulseWire, R: PulseWire> Plugin<S, R> {
         Ok(())
     }
 
-    pub async fn reload(&self, mut child: Child) -> tokio::io::Result<()> {
+    pub async fn reload(&self, mut child: Child, manifest: M) -> 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");
-
+        *self.manifest.lock().await = manifest;
         *process = child;
 
         Ok(())
@@ -107,11 +110,8 @@ pub fn get_manifest_plugin_pair<'de, M: Deserialize<'de>>(
 
 #[derive(Debug)]
 pub struct StrategyPair {
-    pub strategy: Plugin<StrategyEngineMessage, StrategyMessage>,
-    pub risk: Plugin<RiskEngineMessage, RiskMessage>,
-
-    pub strategy_manifest: Mutex<StrategyManifest>,
-    pub risk_manifest: Mutex<RiskManifest>,
+    pub strategy: Plugin<StrategyEngineMessage, StrategyMessage, StrategyManifest>,
+    pub risk: Plugin<RiskEngineMessage, RiskMessage, RiskManifest>,
 }
 
 impl StrategyPair {
@@ -132,10 +132,8 @@ impl StrategyPair {
         )?;
 
         Ok(Arc::new(Self {
-            strategy: Plugin::new(strategy),
-            risk: Plugin::new(risk),
-            strategy_manifest: Mutex::new(strategy_manifest),
-            risk_manifest: Mutex::new(risk_manifest),
+            strategy: Plugin::new(strategy, strategy_manifest),
+            risk: Plugin::new(risk, risk_manifest),
         }))
     }
 }
@@ -143,10 +141,6 @@ impl StrategyPair {
 #[derive(Debug)]
 pub struct StrategyEngine {
     pub pair: Mutex<Arc<StrategyPair>>,
-
-    pub strategy_handle: Mutex<Option<JoinHandle<tokio::io::Result<()>>>>,
-    pub risk_handle: Mutex<Option<JoinHandle<tokio::io::Result<()>>>>,
-
     pub engine: Weak<Engine>,
 }
 
@@ -154,8 +148,6 @@ impl StrategyEngine {
     pub async fn new(strategy_id: &str, risk_id: &str) -> tokio::io::Result<Self> {
         Ok(Self {
             pair: Mutex::new(StrategyPair::new(strategy_id, risk_id).await?),
-            strategy_handle: Mutex::new(None),
-            risk_handle: Mutex::new(None),
             engine: Weak::new(),
         })
     }
@@ -209,33 +201,26 @@ impl StrategyEngine {
     pub async fn spawn(self: &Arc<Self>) {
         let engine = self.clone();
 
-        *self.strategy_handle.lock().await =
-            Some(tokio::spawn(async move { engine.run_strategy().await }));
+        tokio::spawn(async move { engine.run_strategy().await });
 
         let engine = self.clone();
 
-        *self.risk_handle.lock().await = Some(tokio::spawn(async move { engine.run_risk().await }));
+        tokio::spawn(async move { engine.run_risk().await });
     }
 
     pub async fn reload_strategy(self: &Arc<Self>, id: &str) -> tokio::io::Result<()> {
         {
-            if let Some(handle) = &*self.strategy_handle.lock().await {
-                handle.abort();
-            }
-
             let pair = self.pair.lock().await;
 
-            let strategy = pulse_plugin(id)?;
+            let plugin = pulse_plugin(id)?;
 
             let (child, manifest) = get_manifest_plugin_pair(
-                &strategy,
-                &strategy.join("strategy.bash"),
-                &fs::read(strategy.join("strategy.toml")).await?,
+                &plugin,
+                &plugin.join("strategy.bash"),
+                &fs::read(plugin.join("strategy.toml")).await?,
             )?;
 
-            pair.risk.reload(child).await?;
-
-            *pair.strategy_manifest.lock().await = manifest;
+            pair.strategy.reload(child, manifest).await?;
         }
 
         self.spawn().await;

src/engine/terminal.rs

@@ -66,8 +66,8 @@ impl TerminalServer {
         self.send_to(
             id,
             pulse_wire::terminal::TerminalServerMessage::StrategyUpdated(Strategy {
-                strategy: pair.strategy_manifest.lock().await.clone(),
-                risk: pair.risk_manifest.lock().await.clone(),
+                strategy: pair.strategy.manifest.lock().await.clone(),
+                risk: pair.risk.manifest.lock().await.clone(),
                 mode: Mode::Auto,
                 state: ItemState::Running,
                 cooldown: TimeFrame::M15,