From 8c13344f8f97bc67b8fcde68ce14da438f7c66ba Mon Sep 17 00:00:00 2001
From: Lucas Fernandes Nogueira <lucasfernandesnog@gmail.com>
Date: Sun, 27 Jun 2021 10:00:43 -0300
Subject: [PATCH] fix(core): command events not firing consistently (#2082)

---
 .changes/fix-command-event-channel.md |  5 ++
 core/tauri/src/api/process/command.rs | 74 ++++++++++++++-------------
 core/tauri/src/endpoints/shell.rs     |  5 +-
 3 files changed, 45 insertions(+), 39 deletions(-)
 create mode 100644 .changes/fix-command-event-channel.md

diff --git a/.changes/fix-command-event-channel.md b/.changes/fix-command-event-channel.md
new file mode 100644
index 000000000..54710cd56
--- /dev/null
+++ b/.changes/fix-command-event-channel.md
@@ -0,0 +1,5 @@
+---
+"tauri": patch
+---
+
+Fixes `api::process::Command` events not firing consistently.
diff --git a/core/tauri/src/api/process/command.rs b/core/tauri/src/api/process/command.rs
index 9006cb52c..407b17943 100644
--- a/core/tauri/src/api/process/command.rs
+++ b/core/tauri/src/api/process/command.rs
@@ -7,7 +7,8 @@ use std::{
   io::{BufRead, BufReader, Write},
   path::PathBuf,
   process::{Command as StdCommand, Stdio},
-  sync::{Arc, Mutex},
+  sync::{Arc, Mutex, RwLock},
+  thread::spawn,
 };
 
 #[cfg(unix)]
@@ -18,8 +19,7 @@ use std::os::windows::process::CommandExt;
 #[cfg(windows)]
 const CREATE_NO_WINDOW: u32 = 0x0800_0000;
 
-use crate::async_runtime::{channel, spawn, Receiver, RwLock};
-use futures::{future, FutureExt};
+use crate::async_runtime::{channel, spawn as spawn_task, Receiver};
 use os_pipe::{pipe, PipeWriter};
 use serde::Serialize;
 use shared_child::SharedChild;
@@ -243,56 +243,58 @@ impl Command {
 
     let tx_ = tx.clone();
     let guard_ = guard.clone();
-    let stdout_task = async move {
-      let _lock = guard_.read().await;
+    spawn(move || {
+      let _lock = guard_.read().unwrap();
       let reader = BufReader::new(stdout_reader);
       for line in reader.lines() {
-        let _ = match line {
-          Ok(line) => tx_.send(CommandEvent::Stdout(line)).await,
-          Err(e) => tx_.send(CommandEvent::Error(e.to_string())).await,
-        };
+        let tx_ = tx_.clone();
+        spawn_task(async move {
+          let _ = match line {
+            Ok(line) => tx_.send(CommandEvent::Stdout(line)).await,
+            Err(e) => tx_.send(CommandEvent::Error(e.to_string())).await,
+          };
+        });
       }
-    };
+    });
 
     let tx_ = tx.clone();
     let guard_ = guard.clone();
-    let stderr_task = async move {
-      let _lock = guard_.read().await;
+    spawn(move || {
+      let _lock = guard_.read().unwrap();
       let reader = BufReader::new(stderr_reader);
       for line in reader.lines() {
-        let _ = match line {
-          Ok(line) => tx_.send(CommandEvent::Stderr(line)).await,
-          Err(e) => tx_.send(CommandEvent::Error(e.to_string())).await,
-        };
+        let tx_ = tx_.clone();
+        spawn_task(async move {
+          let _ = match line {
+            Ok(line) => tx_.send(CommandEvent::Stderr(line)).await,
+            Err(e) => tx_.send(CommandEvent::Error(e.to_string())).await,
+          };
+        });
       }
-    };
+    });
 
-    let terminated_task = async move {
+    spawn(move || {
       let _ = match child_.wait() {
         Ok(status) => {
-          guard.write().await;
+          let _l = guard.write().unwrap();
           commands().lock().unwrap().remove(&child_.id());
-          tx.send(CommandEvent::Terminated(TerminatedPayload {
-            code: status.code(),
-            #[cfg(windows)]
-            signal: None,
-            #[cfg(unix)]
-            signal: status.signal(),
-          }))
-          .await
+          spawn_task(async move {
+            tx.send(CommandEvent::Terminated(TerminatedPayload {
+              code: status.code(),
+              #[cfg(windows)]
+              signal: None,
+              #[cfg(unix)]
+              signal: status.signal(),
+            }))
+            .await
+          });
         }
         Err(e) => {
-          guard.write().await;
-          tx.send(CommandEvent::Error(e.to_string())).await
+          let _l = guard.write().unwrap();
+          spawn_task(async move { tx.send(CommandEvent::Error(e.to_string())).await });
         }
       };
-    };
-
-    spawn(future::join_all(vec![
-      stdout_task.boxed(),
-      stderr_task.boxed(),
-      terminated_task.boxed(),
-    ]));
+    });
 
     Ok((
       rx,
diff --git a/core/tauri/src/endpoints/shell.rs b/core/tauri/src/endpoints/shell.rs
index 3f4ffdc6b..5ad0432ee 100644
--- a/core/tauri/src/endpoints/shell.rs
+++ b/core/tauri/src/endpoints/shell.rs
@@ -102,9 +102,8 @@ impl Cmd {
           let pid = child.pid();
           command_childs().lock().unwrap().insert(pid, child);
 
-          // TODO: for some reason using `crate::async_runtime::spawn` and `rx.recv().await` doesn't work here, see issue #1935
-          std::thread::spawn(move || {
-            while let Some(event) = rx.blocking_recv() {
+          crate::async_runtime::spawn(async move {
+            while let Some(event) = rx.recv().await {
               if matches!(event, crate::api::process::CommandEvent::Terminated(_)) {
                 command_childs().lock().unwrap().remove(&pid);
               }