From c05f457e62f25a5d2c639e8a993273c9567c2eb2 Mon Sep 17 00:00:00 2001
From: sisungo <sisungo@icloud.com>
Date: Sun, 16 Jun 2024 12:08:36 +0800
Subject: [PATCH] fix: `airupfx_io::line_piper` removes `\n`, `\r` or `\0` from
 output

`airupfx_io::line_piper` by design should call callback with a byte
array which contains `\n`, `\r` or `\0` if the line ends with one of
them. However, current implementation removes them.
---
 airupfx/airupfx-io/src/line_piper.rs |  6 +--
 airupfx/airupfx-process/src/lib.rs   | 60 ++++++++++++++++++----------
 airupfx/airupfx-process/src/unix.rs  |  7 +++-
 airupfx/airupfx-signal/src/unix.rs   |  8 ++++
 4 files changed, 56 insertions(+), 25 deletions(-)

diff --git a/airupfx/airupfx-io/src/line_piper.rs b/airupfx/airupfx-io/src/line_piper.rs
index cffb617..f3afe7b 100644
--- a/airupfx/airupfx-io/src/line_piper.rs
+++ b/airupfx/airupfx-io/src/line_piper.rs
@@ -65,7 +65,7 @@ impl<R: AsyncRead + Unpin + Send + 'static> LinePiperEntity<R> {
                 }
                 position += count;
                 if let Some(pos) = &buf[..position].iter().position(|x| b"\n\r\0".contains(x)) {
-                    break *pos;
+                    break pos + 1;
                 }
                 assert!(position <= 4096);
                 if position == 4096 {
@@ -79,11 +79,11 @@ impl<R: AsyncRead + Unpin + Send + 'static> LinePiperEntity<R> {
 }
 
 #[derive(Clone)]
-struct ChannelCallback {
+pub struct ChannelCallback {
     tx: mpsc::Sender<Vec<u8>>,
 }
 impl ChannelCallback {
-    fn new(tx: mpsc::Sender<Vec<u8>>) -> Self {
+    pub fn new(tx: mpsc::Sender<Vec<u8>>) -> Self {
         Self { tx }
     }
 }
diff --git a/airupfx/airupfx-process/src/lib.rs b/airupfx/airupfx-process/src/lib.rs
index 3703710..b3e4f3d 100644
--- a/airupfx/airupfx-process/src/lib.rs
+++ b/airupfx/airupfx-process/src/lib.rs
@@ -10,12 +10,7 @@ cfg_if::cfg_if! {
 }
 
 use airupfx_io::line_piper::Callback as LinePiperCallback;
-use std::{
-    convert::Infallible,
-    ffi::OsString,
-    ops::{Deref, DerefMut},
-    path::PathBuf,
-};
+use std::{convert::Infallible, ffi::OsString, path::PathBuf};
 
 /// Returns `true` if supervising `forking` services are supported on the system.
 pub fn is_forking_supervisable() -> bool {
@@ -345,20 +340,26 @@ impl Command {
     }
 
     #[inline]
-    pub async fn spawn(&self) -> std::io::Result<Child> {
-        Ok(sys::spawn(self).await?.into())
+    pub fn stdin(&mut self, new: Stdio) -> &mut Self {
+        self.env.stdin(new);
+        self
     }
-}
-impl Deref for Command {
-    type Target = CommandEnv;
 
-    fn deref(&self) -> &Self::Target {
-        &self.env
+    #[inline]
+    pub fn stdout(&mut self, new: Stdio) -> &mut Self {
+        self.env.stdout(new);
+        self
     }
-}
-impl DerefMut for Command {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.env
+
+    #[inline]
+    pub fn stderr(&mut self, new: Stdio) -> &mut Self {
+        self.env.stderr(new);
+        self
+    }
+
+    #[inline]
+    pub async fn spawn(&self) -> std::io::Result<Child> {
+        Ok(sys::spawn(self).await?.into())
     }
 }
 
@@ -375,14 +376,33 @@ impl From<sys::WaitError> for WaitError {
 mod tests {
     #[tokio::test]
     async fn spawn_and_wait() {
-        let wait = crate::Command::new("true")
+        let spawn_wait = move |x| async move {
+            crate::Command::new(x)
+                .spawn()
+                .await
+                .unwrap()
+                .wait()
+                .await
+                .unwrap()
+        };
+
+        assert!(spawn_wait("true").await.is_success());
+        assert!(!spawn_wait("false").await.is_success());
+    }
+
+    #[tokio::test]
+    async fn stdout_capture() {
+        let (tx, mut rx) = tokio::sync::mpsc::channel(2);
+        let callback = airupfx_io::line_piper::ChannelCallback::new(tx);
+        crate::Command::new("echo")
+            .arg("Hello, world!")
+            .stdout(crate::Stdio::Callback(Box::new(callback)))
             .spawn()
             .await
             .unwrap()
             .wait()
             .await
             .unwrap();
-
-        assert!(wait.is_success());
+        assert_eq!(rx.recv().await.as_deref(), Some(&b"Hello, world!\n"[..]));
     }
 }
diff --git a/airupfx/airupfx-process/src/unix.rs b/airupfx/airupfx-process/src/unix.rs
index f6214aa..e01e6ab 100644
--- a/airupfx/airupfx-process/src/unix.rs
+++ b/airupfx/airupfx-process/src/unix.rs
@@ -327,7 +327,7 @@ pub(crate) async fn command_to_std(
     result
         .stdout(command.env.stdout.to_std().await?)
         .stderr(command.env.stderr.to_std().await?)
-        .stdin(command.stdin.to_std().await?);
+        .stdin(command.env.stdin.to_std().await?);
     if command.env.setsid {
         result.setsid();
     }
@@ -359,7 +359,10 @@ pub(crate) fn command_login(env: &mut CommandEnv, name: &str) -> std::io::Result
 }
 
 pub(crate) async fn spawn(cmd: &crate::Command) -> std::io::Result<Child> {
-    Ok(Child::from_std(cmd, command_to_std(cmd).await?.spawn()?))
+    Ok(Child::from_std(
+        &cmd.env,
+        command_to_std(cmd).await?.spawn()?,
+    ))
 }
 
 pub type WaitError = std::convert::Infallible;
diff --git a/airupfx/airupfx-signal/src/unix.rs b/airupfx/airupfx-signal/src/unix.rs
index 3af40d8..5a511db 100644
--- a/airupfx/airupfx-signal/src/unix.rs
+++ b/airupfx/airupfx-signal/src/unix.rs
@@ -118,3 +118,11 @@ extern "C" fn fatal_error_handler(signum: libc::c_int) {
         std::hint::spin_loop();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    #[tokio::test]
+    async fn ignore() {
+        super::ignore(super::SIGUSR1).unwrap();
+    }
+}