Use process group for node runtime

src-tauri/tauri-plugin-plugin-runtime/src/lib.rs

@@ -1,9 +1,10 @@
 extern crate core;
 
-use tauri::plugin::{Builder, TauriPlugin};
-use tauri::{Manager, Runtime};
-
+use log::info;
 use crate::manager::PluginManager;
+use tauri::plugin::{Builder, TauriPlugin};
+use tauri::{Manager, RunEvent, Runtime, State};
+use tokio::sync::Mutex;
 
 pub mod manager;
 mod nodejs;
@@ -17,9 +18,20 @@ pub fn init<R: Runtime>() -> TauriPlugin<R> {
         .setup(|app, _| {
             tauri::async_runtime::block_on(async move {
                 let manager = PluginManager::new(&app).await;
-                app.manage(manager);
+                let manager_state = Mutex::new(manager);
+                app.manage(manager_state);
                 Ok(())
             })
         })
+        .on_event(|app, e| match e {
+            RunEvent::ExitRequested { code, .. } => {
+                tauri::async_runtime::block_on(async move {
+                    info!("Exiting plugin runtime because of app exit {:?}", code);
+                    let manager: State<Mutex<PluginManager>> = app.state();
+                    manager.lock().await.cleanup();
+                });
+            }
+            _ => {}
+        })
         .build()
 }

src-tauri/tauri-plugin-plugin-runtime/src/manager.rs

@@ -1,24 +1,27 @@
+use command_group::GroupChild;
 use log::{debug, info};
 use tauri::{AppHandle, Manager, Runtime};
-use tokio::sync::Mutex;
 use tonic::transport::Channel;
 
 use crate::nodejs::node_start;
 use crate::plugin_runtime::plugin_runtime_client::PluginRuntimeClient;
-use crate::plugin_runtime::{HookExportRequest, HookImportRequest, HookResponse, HookResponseFilterRequest};
+use crate::plugin_runtime::{
+    HookExportRequest, HookImportRequest, HookResponse, HookResponseFilterRequest,
+};
 
 pub struct PluginManager {
-    client: Mutex<PluginRuntimeClient<Channel>>,
+    client: PluginRuntimeClient<Channel>,
+    child: GroupChild,
 }
 
 impl PluginManager {
     pub async fn new<R: Runtime>(app_handle: &AppHandle<R>) -> PluginManager {
         let temp_dir = app_handle.path().temp_dir().unwrap();
 
-        let addr = node_start(app_handle, &temp_dir).await;
-        info!("Connecting to gRPC client at {addr}");
+        let start_resp = node_start(app_handle, &temp_dir).await;
+        info!("Connecting to gRPC client at {}", start_resp.addr);
 
-        let client = match PluginRuntimeClient::connect(addr.clone()).await {
+        let client = match PluginRuntimeClient::connect(start_resp.addr.clone()).await {
             Ok(v) => v,
             Err(err) => {
                 panic!("{}", err.to_string());
@@ -26,15 +29,19 @@ impl PluginManager {
         };
 
         PluginManager {
-            client: Mutex::new(client),
+            client,
+            child: start_resp.child,
         }
     }
 
-    pub async fn run_import(&self, data: &str) -> Result<HookResponse, String> {
+    pub fn cleanup(&mut self) {
+        info!("Cleaning up NodeJS process");
+        self.child.kill().unwrap();
+    }
+
+    pub async fn run_import(&mut self, data: &str) -> Result<HookResponse, String> {
         let response = self
             .client
-            .lock()
-            .await
             .hook_import(tonic::Request::new(HookImportRequest {
                 data: data.to_string(),
             }))
@@ -44,11 +51,9 @@ impl PluginManager {
         Ok(response.into_inner())
     }
 
-    pub async fn run_export_curl(&self, request: &str) -> Result<HookResponse, String> {
+    pub async fn run_export_curl(&mut self, request: &str) -> Result<HookResponse, String> {
         let response = self
             .client
-            .lock()
-            .await
             .hook_export(tonic::Request::new(HookExportRequest {
                 request: request.to_string(),
             }))
@@ -59,7 +64,7 @@ impl PluginManager {
     }
 
     pub async fn run_response_filter(
-        &self,
+        &mut self,
         filter: &str,
         body: &str,
         content_type: &str,
@@ -67,8 +72,6 @@ impl PluginManager {
         debug!("Running plugin filter");
         let response = self
             .client
-            .lock()
-            .await
             .hook_response_filter(tonic::Request::new(HookResponseFilterRequest {
                 filter: filter.to_string(),
                 body: body.to_string(),

src-tauri/tauri-plugin-plugin-runtime/src/nodejs.rs

@@ -1,13 +1,14 @@
 use std::path::PathBuf;
+use std::process::Command;
 use std::time::Duration;
 
+use command_group::{CommandGroup, GroupChild};
 use log::{debug, info};
 use rand::distributions::{Alphanumeric, DistString};
 use serde;
 use serde::Deserialize;
-use tauri::path::BaseDirectory;
 use tauri::{AppHandle, Manager, Runtime};
-use tauri_plugin_shell::process::CommandEvent;
+use tauri::path::BaseDirectory;
 use tauri_plugin_shell::ShellExt;
 use tokio::fs;
 
@@ -17,7 +18,12 @@ struct PortFile {
     port: i32,
 }
 
-pub async fn node_start<R: Runtime>(app: &AppHandle<R>, temp_dir: &PathBuf) -> String {
+pub struct StartResp {
+    pub addr: String,
+    pub child: GroupChild,
+}
+
+pub async fn node_start<R: Runtime>(app: &AppHandle<R>, temp_dir: &PathBuf) -> StartResp {
     let port_file_path = temp_dir.join(Alphanumeric.sample_string(&mut rand::thread_rng(), 10));
 
     let plugins_dir = app
@@ -42,26 +48,17 @@ pub async fn node_start<R: Runtime>(app: &AppHandle<R>, temp_dir: &PathBuf) -> S
         plugin_runtime_dir.to_string_lossy(),
     );
 
-    let (mut rx, _child) = app
+    let cmd = app
         .shell()
         .sidecar("yaaknode")
         .expect("yaaknode not found")
         .env("YAAK_GRPC_PORT_FILE_PATH", port_file_path.clone())
         .env("YAAK_PLUGINS_DIR", plugins_dir)
-        .args(&[plugin_runtime_dir.join("index.cjs")])
-        .spawn()
-        .expect("yaaknode failed to start");
+        .args(&[plugin_runtime_dir.join("index.cjs")]);
 
-    tauri::async_runtime::spawn(async move {
-        // read events such as stdout
-        while let Some(event) = rx.recv().await {
-            if let CommandEvent::Stdout(line) = event {
-                print!("{}", String::from_utf8_lossy(line.as_slice()));
-            } else if let CommandEvent::Stderr(line) = event {
-                print!("{}", String::from_utf8_lossy(line.as_slice()));
-            }
-        }
-    });
+    let child = Command::from(cmd)
+        .group_spawn()
+        .expect("yaaknode failed to start");
 
     let start = std::time::Instant::now();
     let port_file_contents = loop {
@@ -80,5 +77,7 @@ pub async fn node_start<R: Runtime>(app: &AppHandle<R>, temp_dir: &PathBuf) -> S
 
     let port_file: PortFile = serde_json::from_str(port_file_contents.as_str()).unwrap();
     info!("Started plugin runtime on :{}", port_file.port);
-    format!("http://localhost:{}", port_file.port)
+    let addr = format!("http://localhost:{}", port_file.port);
+
+    StartResp { addr, child }
 }