rust - 不能在 mpsc::channel 上使用 Stream::take_while: bool: future 不满足

Question

<分区>

Answer 1

阅读并理解您尝试使用的方法的文档和函数签名:

fn take_while<P, R>(self, pred: P) -> TakeWhile<Self, P, R>
where
    P: FnMut(&Self::Item) -> R,
    R: IntoFuture<Item = bool, Error = Self::Error>,
    Self: Sized, 

take_while 接受一个闭包，该闭包返回某种必须可转换为 future 的类型； bool 不可转换为 future 。最简单的方法是通过 future::ok :

let thr = thread::spawn(|| {
    let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap();
    runtime.spawn({
        rx.take_while(|&x| future::ok(x))
            .for_each(|x| {
                println!("{}", x);
                future::ok(())
            })
    });

    runtime.run()
});

另见:

• The trait bound `(): futures::Future` is not satisfied when using TcpConnectionNew

But my problem also in joining future::lazy and rx.take_while

这与您询问的问题无关。再次，我们查看文档，这次是 Future::and_then :

fn and_then<F, B>(self, f: F) -> AndThen<Self, B, F>
where
    F: FnOnce(Self::Item) -> B,
    B: IntoFuture<Error = Self::Error>,
    Self: Sized, 

类似于take_while，它需要一个闭包并且闭包必须返回可以转换为 future 的东西。您的代码不提供闭包。

然后看Stream::into_future .这将返回 a type that implements Future并返回一个元组。元组中的第一项是流中的单个值，第二项是流本身，以允许获取更多值。

为了让所有项目和错误类型正确，我自由使用了 map(drop) 和 map_err(drop) — 你会想要做一些事情更适合您的数据和错误处理。

runtime.spawn({
    future::lazy(|| {
        println!("event loop started");
        Ok(())
    })
    .and_then(|_| {
        rx.take_while(|&x| future::ok(x))
            .into_future()
            .map(drop)
            .map_err(drop)
    })
    .map(drop)
});

---

真的，你应该只使用 oneshot channel ;这要简单得多:

use futures::{
    future::{self, Future},
    sync::oneshot,
};
use std::thread;

fn main() {
    let (tx, rx) = oneshot::channel();

    let thr = thread::spawn(|| {
        let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap();

        runtime.spawn({
            future::lazy(|| {
                println!("event loop started");
                Ok(())
            })
            .and_then(|_| rx.map_err(drop))
        });

        runtime.run()
    });

    let lines = ["hello", "goodbye", "exit"];
    for &line in &lines {
        if line == "exit" {
            tx.send(()).unwrap();
            break;
        }
    }

    thr.join().unwrap().unwrap();
}