Multiple actix-web client requests - expected struct actix_web::Error found ()

Question

I have this code:

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"

I get error when I cargo run it:

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:_]>>`

Similar code, without running actix-web client, works https://play.rust-lang.org/?version=stable&mode=debug&edition=2015&gist=e81185a73fcb40a3426875573c78accd

EDIT:

Maybe something like map_err(|_| ()) but don’t know how to apply that:

• 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

As rustc says there are two errors here:

1. First is about error types mismatch. It could be fixed with map_err as you already noticed:

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

2. Second (about Send marker) is somewhat more confusing. Basically it means that tokio::Runtime::spawn want its argument future to be also Send + 'static because tokio could move it to another thread. But your future cannot be send between threads safely because it contains non-sendable type (dyn futures::Stream<Item=bytes::bytes::Bytes, Error=actix_web::Error> + 'static) inside actix_web::client::ClientRequest type.

Probably the easiest way to fix the second error is to use actix::spawn function instead tokio::Runtime::spawn method: it requires only 'static marker for its argument.

After these changes your program compiles at least.