How to read subprocess output asynchronously

I want to implement a futures::Stream for reading and parsing the standard output of a child subprocess.

What I'm doing at the moment:

spawn subprocess and obtain its stdout via std::process methods: let child = Command::new(...).stdout(Stdio.pipe()).spawn().expect(...)

add AsyncRead and BufRead to stdout:

let stdout = BufReader::new(tokio_io::io::AllowStdIo::new(
    child.stdout.expect("Failed to open stdout"),
));

declare a wrapper struct for stdout:

struct MyStream<Io: AsyncRead + BufRead> {
    io: Io,
}

implement Stream:

impl<Io: AsyncRead + BufRead> Stream for MyStream<Io> {
    type Item = Message;
    type Error = Error;

    fn poll(&mut self) -> Poll<Option<Message>, Error> {
        let mut line = String::new();
        let n = try_nb!(self.io.read_line(&mut line));
        if n == 0 {
            return Ok(None.into());
        }
        //...read & parse further
    }
}

The problem is that AllowStdIo doesn't make ChildStdout magically asynchronous and the self.io.read_line call still blocks.

I guess I need to pass something different instead of Stdio::pipe() to have it asynchronous, but what? Or is there a different solution for that?

This question is different from What is the best approach to encapsulate blocking I/O in future-rs? because I want to get asynchronous I/O for the specific case of a subprocess and not solve the problem of encapsulation of synchronous I/O.

Update: I'm using tokio = "0.1.3" to leverage its runtime feature and using tokio-process is not an option at the moment (https://github.com/alexcrichton/tokio-process/issues/27)

The tokio-process crate provides you with a CommandExt trait that allows you to spawn a command asynchronously.

The resulting Child has a getter for ChildStdout which implements Read and is non-blocking.

Wrapping tokio_process::ChildStdout into AllowStdIo as you did in your example should make it work!