无法将 futures-util 板条箱与 Actix 一起使用，因为未实现特征 Future [重复]

Question

Scala 有一种方法可以通过 Futures.sequence 将一个可迭代的期货转换为单一的可迭代未来

我在 Rust 中搜索相同的内容并找到了 futures_util crate。我在从 Actix 示例编辑的程序中使用了这个 crate，但编译失败。

Cargo.toml

[package]
name = "actix"
version = "0.7.10"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actor framework for Rust"
readme = "README.md"
keywords = ["actor", "futures", "actix", "async", "tokio"]
homepage = "https://actix.rs"
repository = "https://github.com/actix/actix.git"
documentation = "https://docs.rs/actix/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]

[badges]
travis-ci = { repository = "actix/actix", branch = "master" }
appveyor = { repository = "fafhrd91/actix-n9e64" }
codecov = { repository = "actix/actix", branch = "master", service = "github" }

[lib]
name = "actix"
path = "src/lib.rs"

[workspace]
members = ["examples/chat"]

[features]
default = ["signal", "resolver"]

# dns resolver
resolver = ["trust-dns-resolver", "trust-dns-proto"]

# signal handling
signal = ["tokio-signal"]

[dependencies]
actix_derive = "0.3"

# io
bytes = "0.4"
futures = "0.1"
futures-util = "0.2.1"
tokio = "0.1.7"
tokio-io = "0.1"
tokio-codec = "0.1"
tokio-executor = "0.1"
tokio-reactor = "0.1"
tokio-tcp = "0.1"
tokio-timer = "0.2"

# other
log = "0.4"
fnv = "1.0.5"
failure = "0.1.1"
bitflags = "1.0"
smallvec = "0.6"
crossbeam-channel = "0.3"
parking_lot = "0.7"
uuid = { version = "0.7", features = ["v4"] }

# signal handling
tokio-signal = { version = "0.2", optional = true }

# dns resolver
trust-dns-proto = { version = "^0.5.0", optional = true }
trust-dns-resolver = { version = "^0.10.0", optional = true }

[target.'cfg(unix)'.dependencies]
libc = "0.2"

[profile.release]
lto = true
opt-level = 3
codegen-units = 1

代码：

extern crate actix;
extern crate futures;
extern crate tokio;
extern crate futures_util;

use actix::prelude::*;
use futures::Future;
use futures_util::future::*;
use std::time::{SystemTime, UNIX_EPOCH};

/// Define `Ping` message
struct Ping(usize);

impl Message for Ping {
    type Result = usize;
}

/// Actor
struct MyActor {
    count: usize,
}

/// Declare actor and its context
impl Actor for MyActor {
    type Context = Context<Self>;
}

/// Handler for `Ping` message
impl Handler<Ping> for MyActor {
    type Result = usize;

    fn handle(&mut self, msg: Ping, _: &mut Context<Self>) -> Self::Result {
        self.count += msg.0;
        self.count
    }
}

fn main() {
    // start system, this is required step
    System::run(|| {
        // start new actor
        let addr = MyActor { count: 10 }.start();

        let start = SystemTime::now();

        // send message and get future for result
        let res =  join_all((1..10).into_iter().map(|x| addr.send(Ping(x))));

        // handle() returns tokio handle
        tokio::spawn(
       res.map(|res| {
           let difference = start.duration_since(start)
                          .expect("SystemTime::duration_since failed");
           println!("Time taken: {:?}", difference);

           // stop system and exit
           System::current().stop();
       }).map_err(|_| ()),
        );
    });
}

即使该错误是有意义的，但我发现很难解决，因为 Actix 中的 Request 实现了 Future。我错过了任何进口吗？

error[E0277]: the trait bound `actix::prelude::Request<MyActor, Ping>: futures_core::future::Future` is not satisfied
  --> examples/ping.rs:47:20
   |
47 |         let res =  join_all((1..10).into_iter().map(|x| addr.send(Ping(x))));
   |                    ^^^^^^^^ the trait `futures_core::future::Future` is not implemented for `actix::prelude::Request<MyActor, Ping>`
   |
   = note: required because of the requirements on the impl of `futures_core::future::IntoFuture` for `actix::prelude::Request<MyActor, Ping>`
   = note: required by `futures_util::future::join_all`

error[E0277]: the trait bound `actix::prelude::Request<MyActor, Ping>: futures_core::future::Future` is not satisfied
  --> examples/ping.rs:47:20
   |
47 |         let res =  join_all((1..10).into_iter().map(|x| addr.send(Ping(x))));
   |                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `futures_core::future::Future` is not implemented for `actix::prelude::Request<MyActor, Ping>`
   |
   = note: required by `futures_util::future::JoinAll`

error[E0599]: no method named `map` found for type `futures_util::future::JoinAll<actix::prelude::Request<MyActor, Ping>>` in the current scope
  --> examples/ping.rs:55:12
   |
55 |        res.map(|res| {
   |            ^^^
   |
   = note: the method `map` exists but the following trait bounds were not satisfied:
           `futures_util::future::JoinAll<actix::prelude::Request<MyActor, Ping>> : futures_util::FutureExt`
           `&futures_util::future::JoinAll<actix::prelude::Request<MyActor, Ping>> : futures_util::FutureExt`
           `&mut futures_util::future::JoinAll<actix::prelude::Request<MyActor, Ping>> : futures_util::FutureExt`
           `&mut futures_util::future::JoinAll<actix::prelude::Request<MyActor, Ping>> : futures::Future`
           `&mut futures_util::future::JoinAll<actix::prelude::Request<MyActor, Ping>> : std::iter::Iterator`

Answer 1

在您的项目中，您使用futures-util 中包含的期货的join_all 功能。这个 crate 似乎与 futures 的 actix 版本冲突。

在actix 0.7.10

futures = "0.1"

在futures-util 0.2.1:

futures = "~0.1.15"

我建议你直接使用futures中的join_all：

[package]
name = "Battlefield Vietnam"
version = "0.0.1"

[dependencies]
actix = "0.7"
futures = "0.1"
tokio = "0.1.7"

extern crate actix;
extern crate futures;
extern crate tokio;

use actix::prelude::*;
use futures::future::*;
use futures::Future;
use std::time::SystemTime;

/// Define `Ping` message
struct Ping(usize);

impl Message for Ping {
    type Result = usize;
}

/// Actor
struct MyActor {
    count: usize,
}

/// Declare actor and its context
impl Actor for MyActor {
    type Context = Context<Self>;
}

/// Handler for `Ping` message
impl Handler<Ping> for MyActor {
    type Result = usize;

    fn handle(&mut self, msg: Ping, _: &mut Context<Self>) -> Self::Result {
        self.count += msg.0;
        self.count
    }
}

fn main() {
    // start system, this is required step
    System::run(|| {
        // start new actor
        let addr = MyActor { count: 10 }.start();

        let start = SystemTime::now();

        // send message and get future for result
        let res = join_all((1..10).into_iter().map(move |x| addr.send(Ping(x))));

        // handle() returns tokio handle
        tokio::spawn(
            res.map(move |res| {
                let difference = start
                    .duration_since(start)
                    .expect("SystemTime::duration_since failed");
                println!("Time taken: {:?}", difference);

                // stop system and exit
                System::current().stop();
            })
            .map_err(|_| ()),
        );
    });
}