我想将实现tokio::process::child::ChildStdout
的tokio::io::AsyncRead
的输出转发到实现futures::channel::mpsc::UnboundedSender<MyType>
的futures::sink::Sink
。
我正在使用一个定制的编解码器,该编解码器会生成MyType
的项目,但为了忠于MRE中的M,我将使用Tokio的LinesCodec
并针对这个问题说MyType = String
。
use futures::StreamExt; // 0.3.8
use tokio; // 1.0.1
use tokio_util; // 0.6.0
#[tokio::main]
pub async fn main() {
let (mut tx, rx) = futures::channel::mpsc::unbounded::<String>();
let mut process = tokio::process::Command::new("dmesg")
.arg("-w")
.stdout(std::process::Stdio::piped())
.spawn()
.unwrap();
let stdout = process.stdout.take().unwrap();
let codec = tokio_util::codec::LinesCodec::new();
let framed_read = tokio_util::codec::FramedRead::new(stdout, codec);
let forward = framed_read.forward(tx);
// read from the other end of the channel
tokio::spawn(async move {
while let Some(line) = rx.next().await {
eprintln!("{}", line);
}
});
//forward.await;
}
但是,编译器报告错误类型不匹配:error[E0271]: type mismatch resolving `<futures::channel::mpsc::UnboundedSender<String> as futures::Sink<String>>::Error == LinesCodecError`
--> src/main.rs:19:31
|
19 | let forward = framed_read.forward(tx);
| ^^^^^^^ expected struct `futures::channel::mpsc::SendError`, found enum `LinesCodecError`
假设我在这里没有做根本上错误的事情,如何正确处理/转换这些错误类型?似乎有一个similar question asked before,但似乎是相反的情况, future 0.1,我怀疑它可能已经过时了,因为Rust异步生态系统中的变化如此之快。
最佳答案
流中的项目可能会失败(LinesCodecError
),并且将值发送到 channel 可能会失败(SendError
),但是整个转发过程只会导致单个错误类型。
您可以使用 SinkExt::sink_err_into
和 TryStreamExt::err_into
将错误转换为兼容的统一类型。在这里,我选择了Box<dyn Error>
:
type Error = Box<dyn std::error::Error>;
let forward = framed_read.err_into::<Error>().forward(tx.sink_err_into::<Error>());
在许多情况下,您将创建一个自定义错误类型。您也可能不需要像上面的示例那样使用turbofish,因为类型推断可能会在某个时候开始。也可以看看:
关于rust - 从AsyncRead转发到 future 0.3 mpsc::UnboundedSender <T>时的错误处理,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/65703532/