chore: make function upstream_response_body_filter async

pingora-proxy/src/lib.rs

@@ -224,22 +224,24 @@ impl<SV> HttpProxy<SV> {
         }
     }
 
-    fn upstream_filter(
+    async fn upstream_filter(
         &self,
         session: &mut Session,
         task: &mut HttpTask,
         ctx: &mut SV::CTX,
     ) -> Result<()>
     where
-        SV: ProxyHttp,
+        SV: ProxyHttp + Send + Sync,
     {
         match task {
             HttpTask::Header(header, _eos) => {
                 self.inner.upstream_response_filter(session, header, ctx)
             }
-            HttpTask::Body(data, eos) => self
-                .inner
-                .upstream_response_body_filter(session, data, *eos, ctx),
+            HttpTask::Body(data, eos) => {
+                self.inner
+                    .upstream_response_body_filter(session, data, *eos, ctx)
+                    .await
+            }
             HttpTask::Trailer(Some(trailers)) => self
                 .inner
                 .upstream_response_trailer_filter(session, trailers, ctx)?,

pingora-proxy/src/proxy_h1.rs

@@ -693,7 +693,7 @@ impl<SV> HttpProxy<SV> {
     {
         // skip caching if already served from cache
         if !from_cache {
-            self.upstream_filter(session, &mut task, ctx)?;
+            self.upstream_filter(session, &mut task, ctx).await?;
 
             // cache the original response before any downstream transformation
             // requests that bypassed cache still need to run filters to see if the response has become cacheable

pingora-proxy/src/proxy_h2.rs

@@ -381,7 +381,7 @@ impl<SV> HttpProxy<SV> {
         SV::CTX: Send + Sync,
     {
         if !from_cache {
-            self.upstream_filter(session, &mut task, ctx)?;
+            self.upstream_filter(session, &mut task, ctx).await?;
 
             // cache the original response before any downstream transformation
             // requests that bypassed cache still need to run filters to see if the response has become cacheable

pingora-proxy/src/proxy_trait.rs

@@ -223,7 +223,7 @@ pub trait ProxyHttp {
     ///
     /// This function will be called every time a piece of response body is received. The `body` is
     /// **not the entire response body**.
-    fn upstream_response_body_filter(
+    async fn upstream_response_body_filter(
         &self,
         _session: &mut Session,
         _body: &mut Option<Bytes>,