Rust 期货——将函数改编为 Sink答案

【问题标题】：Rust futures -- adapting a function as a SinkRust 期货——将函数改编为 Sink
【发布时间】：2020-04-10 21:28:57
【问题描述】：

我有一个类似于tokio connect example 的方法，它接受一个接收器：

pub async fn connect(
        addr: &SocketAddr,
        mut stdin: impl Stream<Item = Result<Request, io::Error>> + Unpin,
        mut stdout: impl Sink<Response, Error = io::Error> + Unpin,
    ) -> Result<(), Box<dyn Error>> {

是否有一种标准/简单的方法可以使函数适应接收器以进行打印和/或转换？

例如。类似：

connect(.., .., sink::from_function(|r| match r {
    Ok(response) => println!("received a response: {:?}", response),
    Err(e) => println!("error! {:?}", e);
})
.await;

【问题讨论】：

标签： rust rust-tokio

【解决方案1】：

您可以使用与.with() 方法（映射接收器的输入）链接的drain() 函数（它创建一个只丢弃所有项目的接收器）从函数创建接收器：

use futures::prelude::*;
use futures::sink::drain;

let sink = drain().with(|value| async move { // <-- note async block
    // do something with the input...

    // then return a result
    Ok(())
});

您还可以使用.with() 来检查或转换现有流，您只需确保从闭包返回的成功类型与您正在转换的流的输入相同。

Playground example

【讨论】：

谢谢，这很有帮助！不幸的是，在我的特定示例中，我似乎遇到了 Unpin 的问题。我有wrapped up the example。听起来它可以用pin_mut 宏解决？编译器错误为：within `impl core::future::future::Future`, the trait `std::marker::Unpin` is not implemented for `std::future::GenFuture<[static generator@src/main.rs:9:39: 13:10 value:_ _]>
@arbe 您可以通过将async 块包装在Box::pin 中来固定未来，这会将其固定在堆上：|value| Box::pin(async move { ... })。
@arbe 另外，在您的示例中，您在第 20 行有一个 Unpin 约束，在这种情况下这不是必需的。您可以删除该约束，这也将解决问题而无需将未来移动到堆上（并且只要它可以编译并且您没有使用 unsafe 仍然是安全的）。但是你只能在函数内部没有东西要求它被固定的情况下才能这样做，否则你可以做我上面描述的或者使用 pin_mut！在函数内部，是的。
在声明之后添加pin_mut!(sink); 解决了我的问题。再次感谢！
如果我的 FnMut 有一些状态，这是否有效？我想为每个项目写信给AsyncWriter。