tokio-process-stream

tokio-process-stream是一个简单库，将tokio::process包装为tokio::stream

当我们需要合并多个数据源并从单个入口点开始处理时，拥有进程流接口是有用的。

此库为futures::stream::Stream包装了tokio::process::Child。主要的结构是ProcessLineStream，它实现了该特性，每次产生一个Item枚举，每个包含底层进程的stdout（Item::Stdout）或stderr（Item::Stderr）中的一行，直到它退出。此时，流产生一个单独的Item::Done并结束。

示例用法

use tokio_process_stream::ProcessLineStream;
use tokio::process::Command;
use tokio_stream::StreamExt;
use std::error::Error;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut sleep_cmd = Command::new("sleep");
    sleep_cmd.args(&["1"]);
    let ls_cmd = Command::new("ls");

    let sleep_procstream = ProcessLineStream::try_from(sleep_cmd)?;
    let ls_procstream = ProcessLineStream::try_from(ls_cmd)?;
    let mut procstream = sleep_procstream.merge(ls_procstream);

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

    Ok(())
}

流式传输块

还可以使用ProcessChunkStream流式传输Item<Bytes>块。

use tokio_process_stream::{Item, ProcessChunkStream};
use tokio::process::Command;
use tokio_stream::StreamExt;
use std::error::Error;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut procstream: ProcessChunkStream = Command::new("/bin/sh")
        .arg("-c")
        .arg(r#"printf "1/2"; sleep 0.1; printf "\r2/2 done\n""#)
        .try_into()?;

    while let Some(item) = procstream.next().await {
        println!("{:?}", item);
    }
    Ok(())
}