Merge pull request #2 from jiacai2050/fix-chunk

fix: read partial body

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "prom-remote-api"
-version = "0.2.1"
+version = "0.2.2"
 edition = "2021"
 license = "Apache-2.0"
 description = "Prometheus remote storage API for Rust"
@@ -12,6 +12,7 @@ readme = "README.md"
 
 [dependencies]
 async-trait = "0.1.61"
+bytes = "1.4.0"
 futures = "0.3.25"
 prost = "0.11.6"
 snap = "1.1.0"

examples/simple.rs

@@ -33,7 +33,6 @@ fn generate_samples(start_ms: i64, end_ms: i64, step_ms: i64) -> Vec<Sample> {
 
     // range query
     (start_ms..end_ms)
-        .into_iter()
         .step_by(step_ms as usize)
         .enumerate()
         .map(|(i, timestamp)| Sample {

src/web/warp.rs

@@ -2,6 +2,7 @@
 
 use std::convert::Infallible;
 
+use bytes::Bytes;
 use prost::Message;
 use warp::{
     body,
@@ -11,7 +12,7 @@ use warp::{
         StatusCode,
     },
     reject::{self, Reject},
-    reply, Buf, Filter, Rejection, Reply,
+    reply, Filter, Rejection, Reply,
 };
 
 use crate::{
@@ -63,16 +64,16 @@ impl Reject for Error {}
 // https://github.com/ParkMyCar/warp-protobuf/blob/master/src/lib.rs#L102
 pub fn protobuf_body<T: Message + Send + Default>(
 ) -> impl Filter<Extract = (T,), Error = Rejection> + Copy {
-    async fn from_reader<T: Message + Send + Default>(buf: impl Buf) -> Result<T, Rejection> {
-        util::decode_snappy(buf.chunk())
+    async fn from_reader<T: Message + Send + Default>(bytes: Bytes) -> Result<T, Rejection> {
+        util::decode_snappy(&bytes)
             .map_err(reject::custom)
             .and_then(|decoded_buf| {
                 T::decode(decoded_buf.as_slice())
                     .map_err(|err| reject::custom(Error::ProtoDecode(err)))
             })
     }
 
-    body::aggregate().and_then(from_reader)
+    body::bytes().and_then(from_reader)
 }
 
 impl warp::Reply for ReadResponse {