如何在 MPSC 接收器中使用 Hyper 客户端？答案

【问题标题】：How do I use a Hyper client within a MPSC receiver?如何在 MPSC 接收器中使用 Hyper 客户端？
【发布时间】：2018-03-05 10:48:36
【问题描述】：

我有一个应用程序可以在回调中从队列中拉出项目。回调在一个单独的线程中，所以我不能直接在回调中做任何事情。队列中的项目包括 URL 和要发布到 URL 的消息正文。根据响应，我要么从队列中删除该项目，要么将其重新加入队列。

这是我得到的：

let (mut sender, receiver) = mpsc::channel(1000);
let mut core = Core::new().expect("Error creating core");
let client = Client::new(&core.handle());

// Consumer callback closure
let consumer = move |_: &mut Channel, deliver: basic::Deliver, headers: basic::BasicProperties, data: Vec<u8>|
{
     let body = match String::from_utf8(data) {
        Ok(body) => body,
        Err(error) => {
            error!("Error parsing message body as UTF8: {}", error);
            return;
        }
    };
    if let Err(error) = sender.try_send(QueueData {
        deliver: deliver,
        headers: headers,
        data: body,
    }) {
        error!("Error sending queue data over channel: {}", error);
    }
};

let work = receiver.for_each(|queue_data| {
    let message = match Message::parse(&queue_data.data, &queue_data.headers) {
        Ok(message) => message,
        Err(error) => return err(Box::new(error) as Box<Error>),
    };

    let uri = match Uri::from_str(&message.body.destination) {
        Ok(uri) => uri,
        Err(error) => return err(Box::new(error) as Box<Error>),
    };

    let mut request: Request = Request::new(Method::Post, uri);
    request.headers_mut().set(ContentType::json());
    request.headers_mut().set(ContentLength(queue_data.data.len() as u64));
    request.set_body(queue_data.data.clone().into_bytes());

    client.request(request)
});

这给了我以下错误：

error[E0271]: type mismatch resolving `<futures::FutureResult<(), std::boxed::Box<std::error::Error>> as futures::IntoFuture>::Error == ()`
   --> src/main.rs:116:29
    |
116 |         let work = receiver.for_each(|queue_data| {
    |                             ^^^^^^^^ expected struct `std::boxed::Box`, found ()
    |
    = note: expected type `std::boxed::Box<std::error::Error>`
               found type `()`

error[E0308]: mismatched types
   --> src/main.rs:132:13
    |
132 |             client.request(request)
    |             ^^^^^^^^^^^^^^^^^^^^^^^ expected struct `futures::FutureResult`, found struct `hyper::client::FutureResponse`
    |
    = note: expected type `futures::FutureResult<(), std::boxed::Box<std::error::Error>>`
               found type `hyper::client::FutureResponse`

这是有道理的，但我找不到任何将FutureResponse 转换为FutureResult 的方法。我在这里错过了什么吗？我不能在这样的未来使用 Hyper 客户端吗？

【问题讨论】：

标签： rust hyper

【解决方案1】：

经过大量挖掘，找到了答案。

首先，我需要使用map 而不是for_each，因为for_each 不希望返回任何内容。其次，我需要将返回值包装在futures::future::ok 中，所以我更新后的代码如下所示：

let work = receiver.map(|queue_data| {
    let message = match Message::parse(&queue_data.data, &queue_data.headers) {
        Ok(message) => message,
        Err(error) => return err(Box::new(error) as Box<Error>),
    };

    let uri = match Uri::from_str(&message.body.destination) {
        Ok(uri) => uri,
        Err(error) => return err(Box::new(error) as Box<Error>),
    };

    let mut request: Request = Request::new(Method::Post, uri);
    request.headers_mut().set(ContentType::json());
    request.headers_mut().set(ContentLength(queue_data.data.len() as u64));
    request.set_body(queue_data.data.clone().into_bytes());

    ok(client.request(request))
});

不过，我不得不说，那个错误信息并不是最清楚的......

【讨论】：