如何阅读基于 Tokio 的 Hyper 请求的整个主体？

Question

我想使用 Hyper 的当前主分支编写一个服务器，该服务器保存一条由 POST 请求传递的消息，并将该消息发送到每个传入的 GET 请求。

我有这个，主要是从 Hyper 示例目录中复制的：

extern crate futures;
extern crate hyper;
extern crate pretty_env_logger;

use futures::future::FutureResult;

use hyper::{Get, Post, StatusCode};
use hyper::header::{ContentLength};
use hyper::server::{Http, Service, Request, Response};
use futures::Stream;

struct Echo {
    data: Vec<u8>,
}

impl Echo {
    fn new() -> Self {
        Echo {
            data: "text".into(),
        }
    }
}

impl Service for Echo {
    type Request = Request;
    type Response = Response;
    type Error = hyper::Error;
    type Future = FutureResult<Response, hyper::Error>;

    fn call(&self, req: Self::Request) -> Self::Future {
        let resp = match (req.method(), req.path()) {
            (&Get, "/") | (&Get, "/echo") => {
                Response::new()
                    .with_header(ContentLength(self.data.len() as u64))
                    .with_body(self.data.clone())
            },
            (&Post, "/") => {
                //self.data.clear(); // argh. &self is not mutable :(
                // even if it was mutable... how to put the entire body into it?
                //req.body().fold(...) ?
                let mut res = Response::new();
                if let Some(len) = req.headers().get::<ContentLength>() {
                    res.headers_mut().set(ContentLength(0));
                }
                res.with_body(req.body())
            },
            _ => {
                Response::new()
                    .with_status(StatusCode::NotFound)
            }
        };
        futures::future::ok(resp)
    }

}


fn main() {
    pretty_env_logger::init().unwrap();
    let addr = "127.0.0.1:12346".parse().unwrap();

    let server = Http::new().bind(&addr, || Ok(Echo::new())).unwrap();
    println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
    server.run().unwrap();
}

如何将 req.body()（似乎是 Chunks 的 Stream）变成 Vec<u8>？我假设我必须以某种方式 return 消耗 Stream 并将其变成单个 Vec<u8> 的 Future，也许 fold()。但我不知道该怎么做。

Answer 1

我打算将问题简化为 return 字节总数，而不是回显整个流。

期货 0.3

超级 0.13 + `TryStreamExt::try_fold`

如果您只想将所有数据作为一个巨大的 blob，请参阅 about hyper::body::to_bytes。

访问流允许更细粒度的控制：

use futures::TryStreamExt; // 0.3.7
use hyper::{server::Server, service, Body, Method, Request, Response}; // 0.13.9
use std::convert::Infallible;
use tokio; // 0.2.22

#[tokio::main]
async fn main() {
    let addr = "127.0.0.1:12346".parse().expect("Unable to parse address");

    let server = Server::bind(&addr).serve(service::make_service_fn(|_conn| async {
        Ok::<_, Infallible>(service::service_fn(echo))
    }));

    println!("Listening on http://{}.", server.local_addr());

    if let Err(e) = server.await {
        eprintln!("Error: {}", e);
    }
}

async fn echo(req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
    let (parts, body) = req.into_parts();
    match (parts.method, parts.uri.path()) {
        (Method::POST, "/") => {
            let entire_body = body
                .try_fold(Vec::new(), |mut data, chunk| async move {
                    data.extend_from_slice(&chunk);
                    Ok(data)
                })
                .await;

            entire_body.map(|body| {
                let body = Body::from(format!("Read {} bytes", body.len()));
                Response::new(body)
            })
        }
        _ => {
            let body = Body::from("Can only POST to /");
            Ok(Response::new(body))
        }
    }
}

不幸的是，Bytes 的当前实现不再兼容 TryStreamExt::try_concat，因此我们必须切换回折叠。

期货 0.1

超 0.12 + `Stream::concat2`

从 futures 0.1.14 开始，您可以使用 Stream::concat2 将所有数据粘在一起：

fn concat2(self) -> Concat2<Self>
where
    Self: Sized,
    Self::Item: Extend<<Self::Item as IntoIterator>::Item> + IntoIterator + Default,

use futures::{
    future::{self, Either},
    Future, Stream,
}; // 0.1.25

use hyper::{server::Server, service, Body, Method, Request, Response}; // 0.12.20

use tokio; // 0.1.14

fn main() {
    let addr = "127.0.0.1:12346".parse().expect("Unable to parse address");

    let server = Server::bind(&addr).serve(|| service::service_fn(echo));

    println!("Listening on http://{}.", server.local_addr());

    let server = server.map_err(|e| eprintln!("Error: {}", e));
    tokio::run(server);
}

fn echo(req: Request<Body>) -> impl Future<Item = Response<Body>, Error = hyper::Error> {
    let (parts, body) = req.into_parts();

    match (parts.method, parts.uri.path()) {
        (Method::POST, "/") => {
            let entire_body = body.concat2();
            let resp = entire_body.map(|body| {
                let body = Body::from(format!("Read {} bytes", body.len()));
                Response::new(body)
            });
            Either::A(resp)
        }
        _ => {
            let body = Body::from("Can only POST to /");
            let resp = future::ok(Response::new(body));
            Either::B(resp)
        }
    }
}

您还可以通过 entire_body.to_vec() 将 Bytes 转换为 Vec<u8>，然后将其转换为 String。

另请参阅：

How do I convert a Vector of bytes (u8) to a string

超 0.11 + `Stream::fold`

类似于Iterator::fold，Stream::fold接受一个累加器（称为init）和一个对累加器和项目进行操作的函数从流。该函数的结果必须是与原始错误类型相同的另一个未来。总的结果本身就是一个未来。

fn fold<F, T, Fut>(self, init: T, f: F) -> Fold<Self, F, Fut, T>
where
    F: FnMut(T, Self::Item) -> Fut,
    Fut: IntoFuture<Item = T>,
    Self::Error: From<Fut::Error>,
    Self: Sized,

我们可以使用Vec作为累加器。 Body 的 Stream 实现 return 是 Chunk。这实现了 Deref<[u8]>，因此我们可以使用它来将每个块的数据附加到 Vec。

extern crate futures; // 0.1.23
extern crate hyper;   // 0.11.27

use futures::{Future, Stream};
use hyper::{
    server::{Http, Request, Response, Service}, Post,
};

fn main() {
    let addr = "127.0.0.1:12346".parse().unwrap();

    let server = Http::new().bind(&addr, || Ok(Echo)).unwrap();
    println!(
        "Listening on http://{} with 1 thread.",
        server.local_addr().unwrap()
    );
    server.run().unwrap();
}

struct Echo;

impl Service for Echo {
    type Request = Request;
    type Response = Response;
    type Error = hyper::Error;
    type Future = Box<futures::Future<Item = Response, Error = Self::Error>>;

    fn call(&self, req: Self::Request) -> Self::Future {
        match (req.method(), req.path()) {
            (&Post, "/") => {
                let f = req.body()
                    .fold(Vec::new(), |mut acc, chunk| {
                        acc.extend_from_slice(&*chunk);
                        futures::future::ok::<_, Self::Error>(acc)
                    })
                    .map(|body| Response::new().with_body(format!("Read {} bytes", body.len())));

                Box::new(f)
            }
            _ => panic!("Nope"),
        }
    }
}

您还可以将 Vec<u8> body 转换为 String。

另请参阅：

How do I convert a Vector of bytes (u8) to a string

输出

从命令行调用时，我们可以看到结果：

$ curl -X POST --data hello http://127.0.0.1:12346/
Read 5 bytes

警告

所有这些解决方案都允许恶意最终用户 POST 无限大小的文件，这会导致机器运行内存不足。根据预期用途，您可能希望对读取的字节数设置某种上限，可能会在某个断点处写入文件系统。

另请参阅：

Answer 2

Hyper 0.13 为此提供了 body::to_bytes function。

use hyper::body;
use hyper::{Body, Response};

pub async fn read_response_body(res: Response<Body>) -> Result<String, hyper::Error> {
    let bytes = body::to_bytes(res.into_body()).await?;
    Ok(String::from_utf8(bytes.to_vec()).expect("response was not valid utf-8"))
}

Answer 3

关于这个主题的大部分答案都已过时或过于复杂。解决方案非常简单：

/*
    WARNING for beginners!!! This use statement
    is important so we can later use .data() method!!!
*/
use hyper::body::HttpBody;

let my_vector: Vec<u8> = request.into_body().data().await.unwrap().unwrap().to_vec();
let my_string = String::from_utf8(my_vector).unwrap();

您也可以使用body::to_bytes 。两种方法都是straight-forward！不要忘记正确处理 unwrap。

如何阅读基于 Tokio 的 Hyper 请求的整个主体？

How do I read the entire body of a Tokio-based Hyper request?

rust

hyper

rust-tokio

期货 0.3

超级 0.13 + `TryStreamExt::try_fold`

期货 0.1

超 0.12 + `Stream::concat2`

超 0.11 + `Stream::fold`

输出

警告

如何阅读基于 Tokio 的 Hyper 请求的整个主体？

How do I read the entire body of a Tokio-based Hyper request?

rust

hyper

rust-tokio

期货 0.3

超级 0.13 + TryStreamExt::try_fold

期货 0.1

超 0.12 + Stream::concat2

超 0.11 + Stream::fold

输出

警告

超级 0.13 + `TryStreamExt::try_fold`

超 0.12 + `Stream::concat2`

超 0.11 + `Stream::fold`