Cargo.lock

@@ -592,7 +592,7 @@ dependencies = [
 
 [[package]]
 name = "ntfy-run"
-version = "0.1.2"
+version = "0.1.0"
 dependencies = [
  "clap",
  "itertools",

Cargo.toml

@@ -1,19 +1,7 @@
 [package]
 name = "ntfy-run"
-version = "0.1.2"
+version = "0.1.0"
 edition = "2021"
-authors = ["Quantum <me@quantum5.ca>"]
-description = "ntfy-run is a tool to run a command, capture its output, and send it to a ntfy server."
-readme = "README.md"
-homepage = "https://github.com/quantum5/ntfy-run"
-repository = "https://github.com/quantum5/ntfy-run"
-license = "GPL-3.0-or-later"
-keywords = ["ntfy", "cron", "notifications", "utility"]
-categories = ["command-line-interface"]
-
-exclude = [
-    ".github/*"
-]
 
 [dependencies]
 clap = { version = "4.5.20", features = ["derive", "env"] }

README.md

@@ -16,13 +16,7 @@ sudo wget https://github.com/quantum5/ntfy-run/releases/latest/download/ntfy-run
 sudo chmod a+x /usr/local/bin/ntfy-run
 ```
 
-You can also install the latest release with `cargo`:
-
-```bash
-cargo install ntfy-run
-```
-
-Finally, you can build the latest Git version yourself:
+Alternatively, build it yourself:
 
 ```bash
 git clone https://github.com/quantum5/ntfy-run.git

src/main.rs

@@ -1,13 +1,12 @@
-use crate::runner::{CaptureError, CapturedOutput};
+use crate::runner::CaptureError;
 use clap::Parser;
-use std::process::exit;
+use runner::CapturedOutput;
 
 mod quote;
 mod runner;
-mod tap_stream;
 
 #[derive(Parser)]
-#[command(version, about)]
+/// Tool to run a command, capture its output, and send it to ntfy.
 struct Cli {
     /// URL of the ntfy server and topic, e.g. https://ntfy.sh/topic
     #[arg(short = 'n', long = "ntfy-url", env = "NTFY_URL", alias = "url")]
@@ -100,8 +99,7 @@ fn format_post_body(output: CapturedOutput) -> String {
     }];
 
     if !output.errors.is_empty() {
-        fragments.push("".to_string());
-        fragments.push("========== Errors ==========".to_string());
+        fragments.push("==================== Errors ====================".to_string());
         for error in &output.errors {
             fragments.push(match error {
                 CaptureError::Spawn(error) => format!("Spawn error: {}", error),
@@ -110,18 +108,19 @@ fn format_post_body(output: CapturedOutput) -> String {
                 CaptureError::Wait(error) => format!("Error while waiting for process: {}", error),
             });
         }
+        fragments.push("\n".to_string());
     }
 
     if !output.stdout.is_empty() {
-        fragments.push("".to_string());
-        fragments.push("========== STDOUT ==========".to_string());
-        fragments.push(String::from_utf8_lossy(output.stdout.trim_ascii_end()).into_owned());
+        fragments.push("==================== STDOUT ====================".to_string());
+        fragments.push(output.stdout);
+        fragments.push("\n".to_string());
     }
 
     if !output.stderr.is_empty() {
-        fragments.push("".to_string());
-        fragments.push("========== STDERR ==========".to_string());
-        fragments.push(String::from_utf8_lossy(output.stderr.trim_ascii_end()).into_owned());
+        fragments.push("==================== STDERR ====================".to_string());
+        fragments.push(output.stderr);
+        fragments.push("\n".to_string());
     }
 
     fragments.join("\n")
@@ -210,13 +209,7 @@ async fn main() {
     };
 
     match request.send().await.and_then(|r| r.error_for_status()) {
-        Ok(_) => exit(match status {
-            Some(code) => code.code().unwrap_or(255),
-            None => 255,
-        }),
-        Err(error) => {
-            eprintln!("Failed to send request to ntfy: {}", error);
-            exit(37)
-        }
+        Ok(_) => (),
+        Err(error) => eprintln!("Failed to send request to ntfy: {}", error),
     }
 }

src/runner.rs

@@ -1,18 +1,19 @@
-use crate::tap_stream::{ReadOrWrite, TapStream};
 use std::process::{ExitStatus, Stdio};
-use tokio::{io, process::Command, select};
+use tokio::io::{AsyncBufReadExt, BufReader};
+use tokio::process::Command;
+use tokio::select;
 
 pub enum CaptureError {
-    Spawn(io::Error),
-    Stdout(io::Error),
-    Stderr(io::Error),
-    Wait(io::Error),
+    Spawn(std::io::Error),
+    Stdout(std::io::Error),
+    Stderr(std::io::Error),
+    Wait(std::io::Error),
 }
 
 pub struct CapturedOutput {
     pub status: Option<ExitStatus>,
-    pub stdout: Vec<u8>,
-    pub stderr: Vec<u8>,
+    pub stdout: String,
+    pub stderr: String,
     pub errors: Vec<CaptureError>,
 }
 
@@ -37,50 +38,49 @@ pub async fn run_forward_and_capture(cmdline: &Vec<String>) -> CapturedOutput {
         Err(error) => {
             return CapturedOutput {
                 status: None,
-                stdout: vec![],
-                stderr: vec![],
+                stdout: "".to_string(),
+                stderr: "".to_string(),
                 errors: vec![CaptureError::Spawn(error)],
             }
         }
     };
 
-    let mut stdout_tap = TapStream::new(child.stdout.take().unwrap(), io::stdout());
-    let mut stderr_tap = TapStream::new(child.stderr.take().unwrap(), io::stderr());
+    let mut stdout_stream = BufReader::new(child.stdout.take().unwrap()).lines();
+    let mut stderr_stream = BufReader::new(child.stderr.take().unwrap()).lines();
 
-    let mut stdout = vec![];
-    let mut stderr = vec![];
+    let mut stdout_buffer = Vec::new();
+    let mut stderr_buffer = Vec::new();
     let mut errors = Vec::new();
 
-    let mut stdout_eof = false;
-    let mut stderr_eof = false;
-    let mut maybe_status: Option<ExitStatus> = None;
-
     let status = loop {
         select! {
-            result = stdout_tap.step(), if !stdout_eof => match result {
-                Ok(ReadOrWrite::Read(bytes)) => stdout.extend_from_slice(bytes),
-                Ok(ReadOrWrite::Written) => (),
-                Ok(ReadOrWrite::EOF) => stdout_eof = true,
+            line = stdout_stream.next_line() => match line {
+                Ok(Some(line)) => {
+                    println!("{}", line);
+                    stdout_buffer.push(line);
+                },
+                Ok(None) => (),
                 Err(error) => errors.push(CaptureError::Stdout(error)),
             },
-            result = stderr_tap.step(), if !stderr_eof => match result {
-                Ok(ReadOrWrite::Read(bytes)) => stderr.extend_from_slice(bytes),
-                Ok(ReadOrWrite::Written) => (),
-                Ok(ReadOrWrite::EOF) => stderr_eof = true,
+            line = stderr_stream.next_line() => match line {
+                Ok(Some(line)) => {
+                    eprintln!("{}", line);
+                    stderr_buffer.push(line);
+                },
+                Ok(None) => (),
                 Err(error) => errors.push(CaptureError::Stderr(error)),
             },
-            status = child.wait(), if maybe_status.is_none() => match status {
-                Ok(status) => maybe_status = Some(status),
+            status = child.wait() => match status {
+                Ok(status) => break status,
                 Err(error) => errors.push(CaptureError::Wait(error)),
-            },
-            else => break maybe_status.unwrap(),
+            }
         }
     };
 
     CapturedOutput {
         status: Some(status),
-        stdout,
-        stderr,
+        stdout: stdout_buffer.join("\n").to_string(),
+        stderr: stderr_buffer.join("\n").to_string(),
         errors,
     }
 }

src/tap_stream.rs

@@ -1,52 +0,0 @@
-use std::io;
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
-
-const BUF_SIZE: usize = 16384;
-
-pub struct TapStream<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> {
-    source: R,
-    target: W,
-    buffer: [u8; BUF_SIZE],
-    buf_start: usize,
-    buf_end: usize,
-}
-
-pub enum ReadOrWrite<'a> {
-    Read(&'a [u8]),
-    Written,
-    EOF,
-}
-
-impl<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> TapStream<R, W> {
-    pub fn new(source: R, target: W) -> TapStream<R, W> {
-        TapStream {
-            source,
-            target,
-            buffer: [0; BUF_SIZE],
-            buf_start: 0,
-            buf_end: 0,
-        }
-    }
-
-    pub async fn step(&mut self) -> io::Result<ReadOrWrite> {
-        if self.buf_start == self.buf_end {
-            let bytes = self.source.read(&mut self.buffer[..]).await?;
-            self.buf_start = 0;
-            self.buf_end = bytes;
-
-            if bytes == 0 {
-                Ok(ReadOrWrite::EOF)
-            } else {
-                Ok(ReadOrWrite::Read(&self.buffer[0..bytes]))
-            }
-        } else {
-            let bytes = self
-                .target
-                .write(&self.buffer[self.buf_start..self.buf_end])
-                .await?;
-
-            self.buf_start += bytes;
-            Ok(ReadOrWrite::Written)
-        }
-    }
-}