How to read stdout/err stream of continuous process with tokio (Rust) (and pass args back to it)?

Question

I've been playing around with Rust lately and I am trying to run a async command with tokio. My goal is to read stdout/err live and be able to pass messages to the process if needed. I do not depend on tokio, and if there is a better way of doing this, please advise.

I am using tokio process::Command to create a child process and than transform it into stream and read from it, but if the process is continuous I am getting no stdout/err. Here is what I have been playing around with (used python as continuously running process):

    use std::process::Stdio;
    use tokio_stream::StreamExt;
    use tokio_util::codec::{FramedRead, LinesCodec, FramedWrite};

    let mut process = tokio::process::Command::new("sh")
        .arg("-c")
        .arg("python3")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .stdin(Stdio::piped())
        .spawn()
        .unwrap();


    let stdout = FramedRead::new(process.stdout.take().unwrap(), LinesCodec::new())
        .map(|data| data.expect("fail on out!"));

    let stderr = FramedRead::new(process.stderr.take().unwrap(), LinesCodec::new())
        .map(|data| data.expect("fail on err!"));


    let mut stream = stdout.chain(stderr);

    while let Some(msg) = stream.next().await {
        println!("{:?}", msg);
    }

I also tried waiting for the process with output, where I get empty stdout and stderr - not entierly sure why. When I execute the process without piping stderr/out/in I get the usual initial responses from python.

Cargo.toml dependencies for ref:

tokio = {version="1.26", features = ["rt", "macros", "rt-multi-thread", "fs", "process", "io-std", "io-util"]}
tokio-util = { version="0.7.7", features = ["codec", "io"]}
tokio-stream = "0.1.12"

Answer 1

For anyone struggelling with reading lines as they come from the subprocess, here is the example from the tokio library:

use tokio::io::{BufReader, AsyncBufReadExt};
use tokio::process::Command;

use std::process::Stdio;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut cmd = Command::new("cat");

    // Specify that we want the command's standard output piped back to us.
    // By default, standard input/output/error will be inherited from the
    // current process (for example, this means that standard input will
    // come from the keyboard and standard output/error will go directly to
    // the terminal if this process is invoked from the command line).
    cmd.stdout(Stdio::piped());

    let mut child = cmd.spawn()
        .expect("failed to spawn command");

    let stdout = child.stdout.take()
        .expect("child did not have a handle to stdout");

    let mut reader = BufReader::new(stdout).lines();

    // Ensure the child process is spawned in the runtime so it can
    // make progress on its own while we await for any output.
    tokio::spawn(async move {
        let status = child.wait().await
            .expect("child process encountered an error");

        println!("child status was: {}", status);
    });

    while let Some(line) = reader.next_line().await? {
        println!("Line: {}", line);
    }

    Ok(())
}

https://docs.rs/tokio/latest/tokio/process/#examples

Answer 2

The issue was due to the processes I was using for testing. When tested with another process, does not check if it is called directly everything works correctly.

Information was provided in comment:

FWIW, Python also tries to be smart, and if it detects that stdin is not a TTY (?), it does not start in an interactive mode. Just try echo "foo\n" | python3 in a termial and you can observe that it does not print out a prompt. I would first recommend to edit your question to be a MCVE (currently imports are not listed, dependencies seem to be missing, e.g. map() is not found). 2. Try to get reading stdout/stderr working with a non-interactive program (one that does not switch behavior depending on whether a TTY is attached or not). – justinas