Make use of body inspection rather than body mapping

http-signature-normalization-warp/src/digest/mod.rs

@@ -1,4 +1,4 @@
-use bytes::Buf;
+use bytes::BytesMut;
 use std::str::FromStr;
 use warp::{header, Filter, Rejection};
 
@@ -36,31 +36,30 @@ pub enum ParseDigestError {
 #[error("Could not parse request body")]
 pub struct ParseBodyError;
 
-pub fn verify<F>(
+pub fn verify(
     verifier: impl DigestVerify + Clone + Send + Sync + 'static,
-    filter: F,
-) -> impl Filter<Extract = F::Extract, Error = Rejection> + Clone
-where
-    F: Filter + Clone + Send + Sync + 'static,
-    F::Extract: warp::Reply,
-    F::Error: Into<Rejection>,
-{
-    filter.with(
-        warp::body::map_request_body(parse_digest_header(), move |body: hyper::body::Body, parts: (Vec<DigestPart>,)| {
+) -> impl Filter<Extract = ((),), Error = Rejection> + Clone {
+    parse_digest_header()
+        .and(warp::body::inspect_request_body(
+            BytesMut::new(),
+            move |mut acc, bytes| {
+                let bytes = bytes.clone();
+                async move {
+                    acc.extend_from_slice(&bytes);
+                    acc
+                }
+            },
+        ))
+        .and_then(move |parts: Vec<DigestPart>, bytes_mut: BytesMut| {
             let mut verifier = verifier.clone();
             async move {
-                let parts = parts.clone();
-                let buf = hyper::body::aggregate(body).await.ok()?;
-                let bytes: Vec<u8> = buf.bytes().to_owned();
-
-                if verifier.verify(&parts.0, &bytes)  {
-                    Some(bytes.into())
+                if verifier.verify(&parts, &bytes_mut.freeze()) {
+                    Ok(())
                 } else {
-                    None
+                    Err(warp::reject::custom(VerifyError))
                 }
             }
         })
-    )
 }
 
 fn parse_digest_header() -> impl Filter<Extract = (Vec<DigestPart>,), Error = Rejection> + Clone {