rust - 如何将hyper的Body流转换为Result >？

Question

我正在将代码更新为 super 和 future 的最新版本，但是我尝试过的所有内容都遗漏了某种或多种实现的特征。
一个不工作的例子playground为此...

extern crate futures; // 0.3.5
extern crate hyper; // 0.13.6

use futures::{future, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
use hyper::body;

fn get_body_as_vec<'a>(b: body::Body) -> future::BoxFuture<'a, Result<Vec<String>, hyper::Error>> {
    let f = b.and_then(|bytes| {
        let s = std::str::from_utf8(&bytes).expect("sends no utf-8");
        let mut lines: Vec<String> = Vec::new();
        for l in s.lines() {
            lines.push(l.to_string());
        }
        future::ok(lines)
    });

    Box::pin(f)
}

这将产生错误:

error[E0277]: the trait bound `futures::stream::AndThen<hyper::Body, futures::future::Ready<std::result::Result<std::vec::Vec<std::string::String>, hyper::Error>>, [closure@src/lib.rs:8:24: 15:6]>: futures::Future` is not satisfied
  --> src/lib.rs:17:5
   |
17 |     Box::pin(f)
   |     ^^^^^^^^^^^ the trait `futures::Future` is not implemented for `futures::stream::AndThen<hyper::Body, futures::future::Ready<std::result::Result<std::vec::Vec<std::string::String>, hyper::Error>>, [closure@src/lib.rs:8:24: 15:6]>`
   |
   = note: required for the cast to the object type `dyn futures::Future<Output = std::result::Result<std::vec::Vec<std::string::String>, hyper::Error>> + std::marker::Send`

我无法创造一个兼容的 future 。 Body是一个流，我找不到实现所需特征的任何“转换器”功能。
在hyper 0.12中，我使用了 concat2()。

Answer 1

从reference of and_then:

Note that this function consumes the receiving stream and returns awrapped version of it.

To process the entire stream and return a single future representingsuccess or error, use try_for_each instead.

是的，您的 f仍然是Stream， try_for_each将按引用建议工作，但是 try_fold是将字节表示为向量中的行的更好选择，但作为 @Shepmaster points in the comment;，如果我们直接将块转换为UTF-8，则可能会丢失完整性来自响应的多字节字符。
由于数据的一致性，最简单的解决方案可能是在转换为UTF-8之前收集所有字节。

use futures::{future, FutureExt, TryStreamExt};
use hyper::body;

fn get_body_as_vec<'a>(b: body::Body) -> future::BoxFuture<'a, Result<Vec<String>>> {
    let f = b
        .try_fold(vec![], |mut vec, bytes| {
            vec.extend_from_slice(&bytes);
            future::ok(vec)
        })
        .map(|x| {
            Ok(std::str::from_utf8(&x?)?
                .lines()
                .map(ToString::to_string)
                .collect())
        });

    Box::pin(f)
}

Playground

您可以使用hyper Body中的 channel测试多个块的行为。这是我在大块场景中创建的行分区，可以很好地与上面的代码一起使用，但是如果直接处理大块，则会失去一致性。

let (mut sender, body) = body::Body::channel();

tokio::spawn(async move {
    sender
        .send_data("Line1\nLine2\nLine3\nLine4\nLine5".into())
        .await;
    sender
        .send_data("next bytes of Line5\nLine6\nLine7\nLine8\n----".into())
        .await;
});

println!("{:?}", get_body_as_vec(body).await);

Playground(成功方案)

Playground(失败方案:“Line5的下一个字节”将是
表示为Vec中的新行)

注意:因为 std::error:Error和 hyper::Error都实现了它，所以我一直使用 FromUtf8Error作为返回类型，但是您仍然可以将 expect策略与 hyper::Error一起使用。