Могу ли я использовать Tokio Core и Hyper Client в Rocket?

Question

Я хочу сделать клиентский запрос с Hyper 0.11 после того, как получу входящий запрос от пользователя, с которым я работаю с помощью ракеты, и использую как можно меньше ресурсов. Если я не могу использовать Core или ClientЯ должен был бы создавать их каждый раз, когда Rocket запускает запрос. Это потребовало бы слишком много ресурсов в высокопроизводительной среде для создания Core и Client каждый раз, когда сервер отвечает на запрос. поэтому я использую State сделать объект доступным для всех "возгораний". В Hyper 0.10.13 это работает, но в Hyper 0.11 с Tokio я получаю массивные сообщения об ошибках.

Cargo.toml:

rocket = "0.3.3"
rocket_codegen = "0.3.3"
hyper = "0.11"
futures = "0.1"
tokio-core = "0.1"

main.rs

#![feature(plugin, decl_macro)]
#![plugin(rocket_codegen)]
extern crate futures;
extern crate hyper;
extern crate rocket;
extern crate tokio_core;
use rocket::State;
use hyper::*;
use hyper::header::{ContentLength, ContentType};
use futures::{Future, Stream};
use tokio_core::reactor::Core;

fn main() {
    let mut core = Core::new().unwrap();
    let client = Client::new(&core.handle());
    rocket::ignite()
        .mount("/", routes![index])
        .manage(client.clone())
        .launch();
}

#[post("/", data = "<var>")]
fn index(var: String, client: State<hyper::Client<tokio_core::reactor::Handle>>) -> &'static str {
    let json = r#"{"library":"hyper"}"#;
    let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
    let mut req = Request::new(Method::Post, uri);
    req.headers_mut().set(ContentType::json());
    req.headers_mut().set(ContentLength(json.len() as u64));
    req.set_body(json);
    let post = client.request(req).and_then(|res| {
        //println!("POST: {}", res.status());
        res.body().concat2()
    });
}

Я получаю сообщения об ошибках, такие как:

error[E0277]: the trait bound `std::rc::Weak<std::cell::RefCell<tokio_core::reactor::Inner>>: std::marker::Send` is not satisfied in `hyper::Client<tokio_core::reactor::Handle>`
  --> src/main.rs:23:1
   |
23 | / fn index(var: String, client: State<hyper::Client<tokio_core::reactor::Handle>>) -> &'static str {
24 | |     let json = r#"{"library":"hyper"}"#;
25 | |     let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
26 | |     let mut req = Request::new(Method::Post, uri);
...  |
33 | |     });
34 | | }
   | |_^ `std::rc::Weak<std::cell::RefCell<tokio_core::reactor::Inner>>` cannot be sent between threads safely
   |
   = help: within `hyper::Client<tokio_core::reactor::Handle>`, the trait `std::marker::Send` is not implemented for `std::rc::Weak<std::cell::RefCell<tokio_core::reactor::Inner>>`
   = note: required because it appears within the type `tokio_core::reactor::Handle`
   = note: required because it appears within the type `hyper::Client<tokio_core::reactor::Handle>`
   = note: required by `rocket::State`

error[E0277]: the trait bound `std::rc::Weak<std::cell::RefCell<tokio_core::reactor::Inner>>: std::marker::Sync` is not satisfied in `hyper::Client<tokio_core::reactor::Handle>`
  --> src/main.rs:23:1
   |
23 | / fn index(var: String, client: State<hyper::Client<tokio_core::reactor::Handle>>) -> &'static str {
24 | |     let json = r#"{"library":"hyper"}"#;
25 | |     let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
26 | |     let mut req = Request::new(Method::Post, uri);
...  |
33 | |     });
34 | | }
   | |_^ `std::rc::Weak<std::cell::RefCell<tokio_core::reactor::Inner>>` cannot be shared between threads safely
   |
   = help: within `hyper::Client<tokio_core::reactor::Handle>`, the trait `std::marker::Sync` is not implemented for `std::rc::Weak<std::cell::RefCell<tokio_core::reactor::Inner>>`
   = note: required because it appears within the type `tokio_core::reactor::Handle`
   = note: required because it appears within the type `hyper::Client<tokio_core::reactor::Handle>`
   = note: required by `rocket::State`

error[E0277]: the trait bound `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>>>: std::marker::Send` is not satisfied in `hyper::Client<tokio_core::reactor::Handle>`
  --> src/main.rs:23:1
   |
23 | / fn index(var: String, client: State<hyper::Client<tokio_core::reactor::Handle>>) -> &'static str {
24 | |     let json = r#"{"library":"hyper"}"#;
25 | |     let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
26 | |     let mut req = Request::new(Method::Post, uri);
...  |
33 | |     });
34 | | }
   | |_^ `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>>>` cannot be sent between threads safely
   |
   = help: within `hyper::Client<tokio_core::reactor::Handle>`, the trait `std::marker::Send` is not implemented for `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>>>`
   = note: required because it appears within the type `hyper::client::pool::Pool<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>`
   = note: required because it appears within the type `hyper::client::Dispatch<hyper::Body>`
   = note: required because it appears within the type `hyper::Client<tokio_core::reactor::Handle>`
   = note: required by `rocket::State`

error[E0277]: the trait bound `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>>>: std::marker::Send` is not satisfied in `hyper::Client<tokio_core::reactor::Handle>`
  --> src/main.rs:23:1
   |
23 | / fn index(var: String, client: State<hyper::Client<tokio_core::reactor::Handle>>) -> &'static str {
24 | |     let json = r#"{"library":"hyper"}"#;
25 | |     let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
26 | |     let mut req = Request::new(Method::Post, uri);
...  |
33 | |     });
34 | | }
   | |_^ `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>>>` cannot be sent between threads safely
   |
   = help: within `hyper::Client<tokio_core::reactor::Handle>`, the trait `std::marker::Send` is not implemented for `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>>>`
   = note: required because it appears within the type `hyper::client::pool::Pool<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>`
   = note: required because it appears within the type `hyper::client::Dispatch<hyper::Body>`
   = note: required because it appears within the type `hyper::Client<tokio_core::reactor::Handle>`
   = note: required by `rocket::State`

error[E0277]: the trait bound `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>>>: std::marker::Sync` is not satisfied in `hyper::Client<tokio_core::reactor::Handle>`
  --> src/main.rs:23:1
   |
23 | / fn index(var: String, client: State<hyper::Client<tokio_core::reactor::Handle>>) -> &'static str {
24 | |     let json = r#"{"library":"hyper"}"#;
25 | |     let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
26 | |     let mut req = Request::new(Method::Post, uri);
...  |
33 | |     });
34 | | }
   | |_^ `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>>>` cannot be shared between threads safely
   |
   = help: within `hyper::Client<tokio_core::reactor::Handle>`, the trait `std::marker::Sync` is not implemented for `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>>>`
   = note: required because it appears within the type `hyper::client::pool::Pool<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::proto::MessageHead<hyper::proto::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>`
   = note: required because it appears within the type `hyper::client::Dispatch<hyper::Body>`
   = note: required because it appears within the type `hyper::Client<tokio_core::reactor::Handle>`
   = note: required by `rocket::State`

error[E0277]: the trait bound `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>>>: std::marker::Sync` is not satisfied in `hyper::Client<tokio_core::reactor::Handle>`
  --> src/main.rs:23:1
   |
23 | / fn index(var: String, client: State<hyper::Client<tokio_core::reactor::Handle>>) -> &'static str {
24 | |     let json = r#"{"library":"hyper"}"#;
25 | |     let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
26 | |     let mut req = Request::new(Method::Post, uri);
...  |
33 | |     });
34 | | }
   | |_^ `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>>>` cannot be shared between threads safely
   |
   = help: within `hyper::Client<tokio_core::reactor::Handle>`, the trait `std::marker::Sync` is not implemented for `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>>>`
   = note: required because it appears within the type `hyper::client::pool::Pool<std::cell::RefCell<futures::sync::mpsc::Sender<(hyper::proto::MessageHead<hyper::proto::RequestLine>, std::option::Option<hyper::Body>, futures::Sender<std::result::Result<hyper::Response, hyper::Error>>)>>>`
   = note: required because it appears within the type `hyper::client::Dispatch<hyper::Body>`
   = note: required because it appears within the type `hyper::Client<tokio_core::reactor::Handle>`
   = note: required by `rocket::State`

Что-то не так с моим состоянием? Пожалуйста, не прилагайте слишком много усилий для ответа. Если вы можете указать мне правильное направление, я могу работать сам. Также приветствуется другое решение, как решить эту проблему с помощью другой настройки.

Answer 1

Нет, вы не можете поделиться hyper::Client между нитями. Тем не менее, вы можете поделиться Tokio_core Remote:

#![feature(plugin, decl_macro)]
#![plugin(rocket_codegen)]
extern crate futures;
extern crate hyper;
extern crate rocket;
extern crate tokio_core;

use rocket::State;
use futures::{Future, Stream};
use tokio_core::reactor::Core;

fn main() {
    let core = Core::new().unwrap();
    rocket::ignite()
        .mount("/", routes![index])
        .manage(core.remote())
        .launch();
}

#[post("/", data = "<var>")]
fn index(var: String, remote: State<tokio_core::reactor::Remote>) -> String {
    remote.spawn(|handle| {
        use hyper::{Client, Request, Method};
        use hyper::header::{ContentLength, ContentType};

        let client = Client::new(handle);

        let json = r#"{"library":"hyper"}"#;
        let uri = "http://xxx.xxx.xxx.xxx".parse().unwrap();
        let mut req = Request::new(Method::Post, uri);
        req.headers_mut().set(ContentType::json());
        req.headers_mut().set(ContentLength(json.len() as u64));
        req.set_body(json);

        client.request(req)
            .and_then(|res| res.body().concat2())
            .map(|_| ())
            .map_err(|_| ())
    });

    var
}

Тогда вопрос переходит к вашему другому утверждению:

Я должен был бы создавать их каждый раз, когда Ракета зажигает запрос. Это потребовало бы слишком много ресурсов в высокопроизводительной среде

Я не уверен, на чем вы основываете такое решение. ПостроениеClient не делает много