ref(daemon): switch to process-stream

Author: kkharji

Cargo.lock

@@ -6,16 +6,27 @@ edition = "2021"
 [features]
 default     = [ ]
 xcodegen    = [ "async", "dirs" ]
-daemon      = [ "serial", "compilation", "logging", "async", "watcher", "proc", "xcodegen", "parity-tokio-ipc", "nvim-rs", "strum", "simctl" ]
-server      = [ "serial", "logging", "dirs", "bsp-server", "url", "wax", "shell-words", "compilation", "strum" ]
-lua         = [ "mlua", "serial", "strum", "simctl" ]
+daemon      = [ 
+  "serial", 
+  "compilation", 
+  "logging", 
+  "async", 
+  "watcher",
+  "proc", 
+  "xcodegen", 
+  "parity-tokio-ipc", 
+  "nvim-rs", 
+  "simctl", 
+  "process-stream" 
+  ]
+server      = [ "serial", "logging", "dirs", "bsp-server", "url", "wax", "shell-words", "compilation" ]
+lua         = [ "mlua", "serial", "simctl" ]
 serial      = [ "serde", "serde_json", "serde_yaml" ]
 compilation = [ "serial", "lazy_static", "shell-words", "xcode" ]
 logging     = [ "tracing", "tracing-appender", "tracing-subscriber" ]
-async       = [ "tokio", "async-trait", "futures", "async-stream", "tokio-stream" ]
+async       = [ "tokio", "async-trait", "async-stream" ]
 proc        = [ "libproc" ]
 watcher     = [ "dirs", "notify", "wax" ]
-jsonrpc     = [ "serial" ]
 xcode       = [ "xcodebuild" ]
 
 [lib]
@@ -46,13 +57,13 @@ tap                   = "1.0.1"
 serde                 = { version = "1.0",    features = ["derive"], optional = true }
 serde_json            = { version = "1.0.79", optional = true }
 serde_yaml            = { version = "0.8.23", optional = true }
+strum                 = { version = "0.24.0",  features = ["derive"] }
 # Async Runtime Feature
 tokio                 = { version = "1.17.0", features = ["full"], optional = true }
 tokio-util            = { version = "0.7.1",  features = ["codec"], optional = true }
 async-trait           = { version = "0.1.52", optional = true }
-tokio-stream          = { version = "0.1.8", features = ["io-util"], optional = true  }
 async-stream          = { version = "0.3.3", optional = true }
-futures               = { version = "0.3.21", optional = true } 
+process-stream        = { version  = "0.1.3", optional = true }
 # Logging Feature
 tracing               = { version = "0.1.32", optional = true }
 tracing-subscriber    = { version = "0.3.9",  features = ["env-filter"], optional = true}
@@ -73,7 +84,6 @@ libproc               = { version = "0.12.0", optional = true }
 url                   = { version = "2.2.2",  features = ["serde"], optional = true }
 nvim-rs               = { version = "0.4.0", optional = true, features = ["use_tokio"] }
 parity-tokio-ipc      = { version = "0.9.0", optional = true }
-strum                 = { version = "0.24.0",  features = ["derive"], optional  = true }
 simctl                = { path = "../../../sources/simctl/", optional = true }
 thiserror = "1.0.31"
 libc = "0.2.126"

Cargo.toml

@@ -133,15 +133,16 @@ use {
         xcode::fresh_build,
         xcodegen,
     },
+    process_stream::StreamExt,
     std::path::PathBuf,
     tokio::{
         fs::{metadata, File},
         io::AsyncWriteExt,
         sync::OwnedMutexGuard,
     },
-    tokio_stream::StreamExt,
 };
 
+#[cfg(feature = "daemon")]
 pub async fn update_compilation_file(root: &path::PathBuf) -> Result<()> {
     // TODO(build): Ensure that build successed. check for Exit Code
     let steps = fresh_build(&root).await?.collect::<Vec<Step>>().await;

src/compile.rs

@@ -7,17 +7,14 @@ use crate::{
     nvim::BufferDirection,
     state::State,
     types::{Client, Platform},
-    util::{
-        fmt,
-        process::{self, Output},
-    },
+    util::fmt,
     Error, Result,
 };
 use {
+    process_stream::{Process, ProcessItem, StreamExt},
     tap::Pipe,
     tokio::{sync::OwnedMutexGuard, task::JoinHandle},
-    tokio_stream::StreamExt,
-    xcodebuild::{parser::BuildSettings, runner},
+    xcodebuild::parser::BuildSettings,
 };
 
 pub struct Runner {
@@ -51,11 +48,11 @@ impl Runner {
 
         tokio::spawn(async move {
             let program = settings.path_to_output_binary()?;
-            let mut stream = runner::run(&program).await?;
+            let mut stream = Process::new(&program).stream()?;
 
             tracing::debug!("Running binary {program:?}");
 
-            use xcodebuild::runner::ProcessUpdate::*;
+            use ProcessItem::*;
             // NOTE: This is required so when neovim exist this should also exit
             while let Some(update) = stream.next().await {
                 let state = DAEMON_STATE.clone();
@@ -65,20 +62,15 @@ impl Runner {
 
                 // NOTE: NSLog get directed to error by default which is odd
                 match update {
-                    Stdout(msg) => {
-                        logger.log(msg).await?;
-                    }
-                    Error(msg) | Stderr(msg) => {
-                        logger.log(format!("[Error]  {msg}")).await?;
-                    }
+                    Output(msg) => logger.log(msg).await?,
+                    Error(msg) => logger.log(format!("[Error] {msg}")).await?,
                     Exit(ref code) => {
                         let success = code == "0";
                         let msg = fmt::as_section(if success {
                             "".into()
                         } else {
                             format!("Panic {code}")
                         });
-
                         logger.log(msg).await?;
                         logger.set_status_end(success, true).await?;
                     }
@@ -121,42 +113,37 @@ impl Runner {
             runner
         };
 
-        let child = runner.launch(logger).await?;
-        let mut stream = process::stream(child)?;
+        let mut launcher = runner.launch(logger).await?;
+        let mut stream = launcher.stream()?;
         // TODO(daemon): ensure Simulator.app is running
 
         logger.log(fmt::separator()).await?;
 
-        // TODO(nvim): Close ran simctl process on exit.
         tokio::spawn(async move {
             while let Some(output) = stream.next().await {
                 let state = DAEMON_STATE.clone();
                 let state = state.lock().await;
                 let mut logger = match state.clients.get(&client.pid) {
                     Ok(nvim) => nvim.new_unamed_logger(),
-                    Err(e) => {
-                        tracing::error!("SimDevice runner: {e}");
+                    Err(_) => {
+                        tracing::info!("Nvim Instance closed, closing runner ..");
+                        launcher.kill().await;
                         break;
                     }
                 };
 
+                use ProcessItem::*;
                 match output {
-                    Output::Out(msg) => {
+                    Output(msg) => {
                         if !msg.contains("ignoring singular matrix") {
                             logger.log(msg).await?;
                         }
                     }
-                    Output::Err(msg) => {
+                    Error(msg) => {
                         logger.log(format!("[Error] {msg}")).await?;
                     }
-                    Output::Exit(status) => {
-                        if let Ok(Some(code)) = status {
-                            logger.log(format!("[Exit] {code}")).await?;
-                        } else {
-                            logger
-                                .log(format!("[Error] Unable to get exit code"))
-                                .await?;
-                        }
+                    Exit(code) => {
+                        logger.log(format!("[Exit] {code}")).await?;
                         break;
                     }
                 };

src/runner.rs

@@ -1,7 +1,7 @@
 use crate::{nvim::Logger, types::SimDevice, Error, Result};
-use std::{path::PathBuf, process::Stdio};
+use process_stream::Process;
+use std::path::PathBuf;
 use tap::Pipe;
-use tokio::process::{Child, Command};
 
 /// SimDevice ruuner
 pub struct SimDeviceRunner {
@@ -35,20 +35,19 @@ impl SimDeviceRunner {
         Ok(())
     }
 
-    pub async fn launch<'a>(&self, logger: &mut Logger<'a>) -> Result<Child> {
+    pub async fn launch<'a>(&self, logger: &mut Logger<'a>) -> Result<Process> {
         logger.log(self.launching_msg()).await?;
-        let process = Command::new("xcrun")
-            .arg("simctl")
-            .arg("launch")
-            .arg("--terminate-running-process")
-            .arg("--console")
-            .arg(&self.device.udid)
-            .arg(&self.app_id)
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .stdin(Stdio::null())
-            .kill_on_drop(true)
-            .spawn()?;
+        let mut process = Process::new("xcrun");
+
+        process.args(&[
+            "simctl",
+            "launch",
+            "--terminate-running-process",
+            "--console",
+        ]);
+        process.arg(&self.device.udid);
+        process.arg(&self.app_id);
+        process.kill_on_drop(true);
 
         logger.log(self.connected_msg()).await?;
         Ok(process)

src/runner/simctl.rs

@@ -10,6 +10,3 @@ pub mod serde;
 pub mod tracing;
 
 pub mod fmt;
-
-#[cfg(feature = "daemon")]
-pub mod process;