多个 actix-web 客户端请求 - 预期 struct actix_web::Error found ()

Question

我有这个代码：

extern crate actix;
extern crate actix_web;
extern crate env_logger;
extern crate futures; // 0.1.25
extern crate tokio; // 0.1.17
use futures::future::ok as fut_ok;
use futures::Future;
use tokio::runtime::Builder;

use std::sync::{Arc, Mutex};
extern crate serde_json;

type Error = ();

fn questions_data(
    val: i32,
) -> Box<Future<Item = serde_json::Value, Error = actix_web::error::Error>> {
    let f = std::fs::read_to_string("auth_token").unwrap();
    let token = f.trim();
    use actix_web::{client, HttpMessage};
    use std::time::Duration;
    Box::new(
        client::ClientRequest::get(
            "https://jsonplaceholder.typicode.com/todos/".to_owned() + &val.to_string(),
        )
        .header(
            actix_web::http::header::AUTHORIZATION,
            "Bearer ".to_owned() + token,
        )
        .finish()
        .unwrap()
        .send()
        .timeout(Duration::from_secs(30))
        .map_err(actix_web::error::Error::from) // <- convert SendRequestError to an Error
        .and_then(|resp| {
            resp.body().limit(67_108_864).from_err().and_then(|body| {
                let resp: serde_json::Value = serde_json::from_slice(&body).unwrap();
                fut_ok(resp)
            })
        }),
    )
}

fn main() {
    let num_workers = 8;

    let mut core = Builder::new().core_threads(num_workers).build().unwrap();

    let results = Arc::new(Mutex::new(Vec::new()));
    for n in 1..100 {
        let res = results.clone();
        core.spawn(questions_data(n).map(move |n| {
            res.lock().unwrap().push(n);
        }));
    }
    core.shutdown_on_idle().wait().unwrap();
    let data = results.lock().unwrap();
    println!("{:?}", *data);
}

[dependencies]
futures = "0.1.25"
tokio-core = "0.1.17"
futures-util = "0.2.1"
tokio = "0.1.11"
rand = "0.6.0"
actix-web = "0.7.14"
actix = "0.7.6"
env_logger = "0.6.0"
serde_json = "1.0.33"

当我cargo run它时出现错误：

error[E0271]: type mismatch resolving `<std::boxed::Box<futures::Map<std::boxed::Box<dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>>, [closure@src/main.rs:52:51: 54:10 res:_]>> as futures::Future>::Error == ()`
  --> src/main.rs:52:14
   |
52 |         core.spawn(Box::new(questions_data(n).map(move |n| {
   |              ^^^^^ expected struct `actix_web::Error`, found ()
   |
   = note: expected type `actix_web::Error`
              found type `()`

error[E0277]: `dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>` cannot be sent between threads safely
  --> src/main.rs:52:14
   |
52 |         core.spawn(Box::new(questions_data(n).map(move |n| {
   |              ^^^^^ `dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>` cannot be sent between threads safely
   |
   = help: the trait `std::marker::Send` is not implemented for `dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>>`
   = note: required because it appears within the type `std::boxed::Box<dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>>`
   = note: required because it appears within the type `futures::Map<std::boxed::Box<dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>>, [closure@src/main.rs:52:51: 54:10 res:_]>`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<futures::Map<std::boxed::Box<dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>>, [closure@src/main.rs:52:51: 54:10 res:_]>>`
   = note: required because it appears within the type `std::boxed::Box<futures::Map<std::boxed::Box<dyn futures::Future<Item=serde_json::Value, Error=actix_web::Error>>, [closure@src/main.rs:52:51: 54:10 res:_]>>`

类似的代码，无需运行 actix-web 客户端，工作 https://play.rust-lang.org/?version=stable&mode=debug&edition=2015&gist=e81185a73fcb40a3426875573c78accd

编辑：

可能类似于map_err(|_| ())，但不知道如何应用：

• https://users.rust-lang.org/t/type-error-when-combining-streams/21619
• Type mismatch resolving the error type when forwarding messages from a futures channel to a WebSocket Sink

Answer 1

正如rustc所说，这里有两个错误：

1. 首先是关于错误类型不匹配的问题。正如您已经注意到的，它可以使用map_err 修复：

   questions_data(n)
       .map(move |n| {
           res.lock().unwrap().push(n);
       })
       .map_err(|e| println!("an error occurred: {}", e))
   

2. Second（关于 Send 标记）有点令人困惑。基本上这意味着tokio::Runtime::spawn 希望它的论点未来也是Send + 'static，因为 tokio 可以将它移动到另一个线程。但是你的 future 不能在线程之间安全地发送，因为它在 actix_web::client::ClientRequest 类型中包含不可发送类型 (dyn futures::Stream<Item=bytes::bytes::Bytes, Error=actix_web::Error> + 'static)。

修复第二个错误的最简单方法可能是使用actix::spawn 函数而不是tokio::Runtime::spawn 方法：它的参数只需要'static 标记。

在这些更改之后，您的程序至少可以编译。