refactor body limit, for reuse in server

Signed-off-by: Ian Jackson <ijackson@chiark.greenend.org.uk>

diff --git a/src/bin/client.rs b/src/bin/client.rs
index 9c2f297985e500dd3931c5a146800e761b7524bf..29b6ef9dff1c66dda86b7236d14a3373145e4053 100644 (file)
--- a/src/bin/client.rs
+++ b/src/bin/client.rs
@@ -142,9 +142,10 @@ fn submit_request<'r, 'c:'r, C:HCC>(
     let r = async { tokio::time::timeout( c.ic.effective_http_timeout, async {
       let resp = resp.await.context("make request")?;
       let status = resp.status();
-      let resp = resp.into_body();
+      let mut resp = resp.into_body();
       let max_body = c.ic.max_batch_down.sat() + MAX_BATCH_DOWN_RESP_OVERHEAD;
-      let resp = read_limited_body(max_body, resp).await?;
+      let resp = read_limited_bytes(max_body, &mut resp).await
+        .context("fetching response body")?;
 
       if ! status.is_success() {
         throw!(anyhow!("HTTP error status={} body={:?}",

diff --git a/src/utils.rs b/src/utils.rs
index 2271fd47cb2b0a97c7136bacd58d4ddbd63d1b8f..b84d2f20f54c0e0dee058c5a8ed00ab81876fbcc 100644 (file)
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -16,19 +16,28 @@ impl<T,E> Result<T,E> where AE: From<E> {
   }
 }
 
-#[throws(AE)]
-pub async fn read_limited_body<S,E>(limit: usize, mut stream: S) -> Box<[u8]>
-where S: futures::Stream<Item=Result<hyper::body::Bytes,E>> + Unpin,
+#[derive(Error,Debug)]
+pub enum ReadLimitedError {
+  #[error("maximum size {limit} exceeded")]
+  Truncated { limit: usize },
+
+  #[error("HTTP error {0}")]
+  Hyper(#[from] hyper::Error),
+}
+
+#[throws(ReadLimitedError)]
+pub async fn read_limited_bytes<S>(limit: usize, stream: &mut S) -> Box<[u8]>
+where S: futures::Stream<Item=Result<hyper::body::Bytes,hyper::Error>>
+         + Debug + Unpin,
       // we also require that the Stream is cancellation-safe
-      E: std::error::Error + Sync + Send + 'static,
 {
   let mut accum = vec![];
   while let Some(item) = stream.next().await {
-    let b = item.context("HTTP error fetching response body")?;
-    if accum.len() + b.len() > limit {
-      throw!(anyhow!("maximum response body size {} exceeded", limit));
-    }
+    let b = item?;
     accum.extend(b);
+    if accum.len() > limit {
+      throw!(ReadLimitedError::Truncated { limit })
+    }
   }
   accum.into()
 }