16

我想使用 Hyper 的当前 master 分支编写一个服务器，它保存由 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()（似乎是 a Streamof Chunks）变成 a Vec<u8>？我想我必须以某种方式返回一个Future消耗Streama并将它变成一个单一的Vec<u8>，也许是fold(). 但我不知道该怎么做。

4

3 回答 3

31

Hyper 0.13为此提供了一个body::to_bytes功能。

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"))
}
于 2019-12-20T16:19:43.643 回答
16

我将简化问题，只返回字节总数，而不是回显整个流。

期货 0.3

超 0.13 +TryStreamExt::try_fold

如果您只想将所有数据作为一个巨大的斑点，请参阅euclio 的回答。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

从期货 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)
        }
    }
}

您也可以将其Bytes转换为Vec<u8>via entire_body.to_vec()，然后将其转换为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,

我们可以使用 aVec作为累加器。Body的Stream实现返回一个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.

也可以看看：

输出

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

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

警告

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

也可以看看：

于 2017-04-14T23:33:37.923 回答
-1

关于该主题的大多数答案都已过时或过于复杂。解决方案非常简单：

/*
    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 @euclio 回答。两种方法都是直截了当的！不要忘记unwrap妥善处理。

于 2020-11-06T14:02:57.323 回答