Allow to set default response payload limit and timeout (#276)

.github/workflows/linux.yml

@@ -67,5 +67,6 @@ jobs:
           cargo install cargo-cache --version 0.6.2 --no-default-features --features ci-autoclean
 
       - name: Clear the cargo caches
+        continue-on-error: true
         run: |
           cargo-cache

ntex/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.7.17] - 2024-01-05
+
+* Allow to set default response payload limit and timeout
+
 ## [0.7.16] - 2023-12-15
 
 * Stop timer before handling UPGRADE h1 requests

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "0.7.16"
+version = "0.7.17"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"

ntex/src/http/client/builder.rs

@@ -37,6 +37,8 @@ impl ClientBuilder {
             config: ClientConfig {
                 headers: HeaderMap::new(),
                 timeout: Millis(5_000),
+                response_pl_limit: 262_144,
+                response_pl_timeout: Millis(10_000),
                 connector: Box::new(ConnectorWrapper(Connector::default().finish().into())),
             },
         }
@@ -91,6 +93,22 @@ impl ClientBuilder {
         self
     }
 
+    /// Max size of response payload.
+    /// By default max size is 256Kb
+    pub fn response_payload_limit(mut self, limit: usize) -> Self {
+        self.config.response_pl_limit = limit;
+        self
+    }
+
+    /// Set response timeout.
+    ///
+    /// Response payload timeout is the total time before a payload must be received.
+    /// Default value is 10 seconds.
+    pub fn response_payload_timeout(mut self, timeout: Millis) -> Self {
+        self.config.response_pl_timeout = timeout;
+        self
+    }
+
     /// Add default header. Headers added by this method
     /// get added to every request.
     pub fn header<K, V>(mut self, key: K, value: V) -> Self

ntex/src/http/client/connect.rs

@@ -1,11 +1,11 @@
-use std::{fmt, net};
+use std::{fmt, net, rc::Rc};
 
 use crate::http::{body::Body, RequestHeadType};
 use crate::{service::Pipeline, service::Service, time::Millis, util::BoxFuture};
 
 use super::error::{ConnectError, SendRequestError};
 use super::response::ClientResponse;
-use super::{Connect as ClientConnect, Connection};
+use super::{ClientConfig, Connect as ClientConnect, Connection};
 
 pub(super) struct ConnectorWrapper<T>(pub(crate) Pipeline<T>);
 
@@ -27,6 +27,7 @@ pub(super) trait Connect: fmt::Debug {
         body: Body,
         addr: Option<net::SocketAddr>,
         timeout: Millis,
+        cfg: Rc<ClientConfig>,
     ) -> BoxFuture<'_, Result<ClientResponse, SendRequestError>>;
 }
 
@@ -40,6 +41,7 @@ where
         body: Body,
         addr: Option<net::SocketAddr>,
         timeout: Millis,
+        cfg: Rc<ClientConfig>,
     ) -> BoxFuture<'_, Result<ClientResponse, SendRequestError>> {
         Box::pin(async move {
             // connect to the host
@@ -54,7 +56,7 @@ where
             connection
                 .send_request(head, body, timeout)
                 .await
-                .map(|(head, payload)| ClientResponse::new(head, payload))
+                .map(|(head, payload)| ClientResponse::new(head, payload, cfg))
         })
     }
 }

ntex/src/http/client/frozen.rs

@@ -6,8 +6,7 @@ use crate::http::header::{self, HeaderMap, HeaderName, HeaderValue};
 use crate::http::{Method, RequestHead, RequestHeadType, Uri};
 use crate::{time::Millis, util::Bytes, util::Stream};
 
-use super::sender::SendClientRequest;
+use super::{sender::SendClientRequest, ClientConfig};
-use super::ClientConfig;
 
 /// `FrozenClientRequest` struct represents clonable client request.
 /// It could be used to send same request multiple times.

ntex/src/http/client/mod.rs

@@ -70,23 +70,27 @@ pub struct Connect {
 ///      println!("Response: {:?}", res);
 /// }
 /// ```
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 pub struct Client(Rc<ClientConfig>);
 
 #[derive(Debug)]
-struct ClientConfig {
+pub(crate) struct ClientConfig {
     pub(self) connector: Box<dyn HttpConnect>,
     pub(self) headers: HeaderMap,
     pub(self) timeout: Millis,
+    pub(self) response_pl_limit: usize,
+    pub(self) response_pl_timeout: Millis,
 }
 
-impl Default for Client {
+impl Default for ClientConfig {
     fn default() -> Self {
-        Client(Rc::new(ClientConfig {
+        ClientConfig {
-            connector: Box::new(ConnectorWrapper(Connector::default().finish().into())),
             headers: HeaderMap::new(),
             timeout: Millis(5_000),
-        }))
+            response_pl_limit: 262_144,
+            response_pl_timeout: Millis(10_000),
+            connector: Box::new(ConnectorWrapper(Connector::default().finish().into())),
+        }
     }
 }
 

ntex/src/http/client/response.rs

@@ -1,6 +1,6 @@
 use std::cell::{Ref, RefMut};
 use std::task::{Context, Poll};
-use std::{fmt, future::Future, marker::PhantomData, mem, pin::Pin};
+use std::{fmt, future::Future, marker::PhantomData, mem, pin::Pin, rc::Rc};
 
 use serde::de::DeserializeOwned;
 
@@ -13,12 +13,13 @@ use crate::http::{HeaderMap, HttpMessage, Payload, ResponseHead, StatusCode, Ver
 use crate::time::{Deadline, Millis};
 use crate::util::{Bytes, BytesMut, Extensions, Stream};
 
-use super::error::JsonPayloadError;
+use super::{error::JsonPayloadError, ClientConfig};
 
 /// Client Response
 pub struct ClientResponse {
     pub(crate) head: ResponseHead,
     pub(crate) payload: Payload,
+    config: Rc<ClientConfig>,
 }
 
 impl HttpMessage for ClientResponse {
@@ -58,12 +59,20 @@ impl HttpMessage for ClientResponse {
 
 impl ClientResponse {
     /// Create new client response instance
-    pub(crate) fn new(head: ResponseHead, payload: Payload) -> Self {
+    pub(crate) fn new(
-        ClientResponse { head, payload }
+        head: ResponseHead,
+        payload: Payload,
+        config: Rc<ClientConfig>,
+    ) -> Self {
+        ClientResponse {
+            head,
+            payload,
+            config,
+        }
     }
 
-    pub(crate) fn with_empty_payload(head: ResponseHead) -> Self {
+    pub(crate) fn with_empty_payload(head: ResponseHead, config: Rc<ClientConfig>) -> Self {
-        ClientResponse::new(head, Payload::None)
+        ClientResponse::new(head, Payload::None, config)
     }
 
     #[inline]
@@ -193,7 +202,11 @@ impl MessageBody {
         MessageBody {
             length: len,
             err: None,
-            fut: Some(ReadBody::new(res.take_payload(), 262_144)),
+            fut: Some(ReadBody::new(
+                res.take_payload(),
+                res.config.response_pl_limit,
+                res.config.response_pl_timeout,
+            )),
         }
     }
 
@@ -236,7 +249,8 @@ impl Future for MessageBody {
         }
 
         if let Some(len) = this.length.take() {
-            if len > this.fut.as_ref().unwrap().limit {
+            let limit = this.fut.as_ref().unwrap().limit;
+            if limit > 0 && len > limit {
                 return Poll::Ready(Err(PayloadError::Overflow));
             }
         }
@@ -264,9 +278,9 @@ where
     U: DeserializeOwned,
 {
     /// Create `JsonBody` for request.
-    pub fn new(req: &mut ClientResponse) -> Self {
+    pub fn new(res: &mut ClientResponse) -> Self {
         // check content-type
-        let json = if let Ok(Some(mime)) = req.mime_type() {
+        let json = if let Ok(Some(mime)) = res.mime_type() {
             mime.subtype() == mime::JSON || mime.suffix() == Some(mime::JSON)
         } else {
             false
@@ -281,7 +295,7 @@ where
         }
 
         let mut len = None;
-        if let Some(l) = req.headers().get(&CONTENT_LENGTH) {
+        if let Some(l) = res.headers().get(&CONTENT_LENGTH) {
             if let Ok(s) = l.to_str() {
                 if let Ok(l) = s.parse::<usize>() {
                     len = Some(l)
@@ -292,7 +306,11 @@ where
         JsonBody {
             length: len,
             err: None,
-            fut: Some(ReadBody::new(req.take_payload(), 65536)),
+            fut: Some(ReadBody::new(
+                res.take_payload(),
+                res.config.response_pl_limit,
+                res.config.response_pl_timeout,
+            )),
             _t: PhantomData,
         }
     }
@@ -331,7 +349,8 @@ where
         }
 
         if let Some(len) = self.length.take() {
-            if len > self.fut.as_ref().unwrap().limit {
+            let limit = self.fut.as_ref().unwrap().limit;
+            if limit > 0 && len > limit {
                 return Poll::Ready(Err(JsonPayloadError::Payload(PayloadError::Overflow)));
             }
         }
@@ -353,12 +372,12 @@ struct ReadBody {
 }
 
 impl ReadBody {
-    fn new(stream: Payload, limit: usize) -> Self {
+    fn new(stream: Payload, limit: usize, timeout: Millis) -> Self {
         Self {
             stream,
             limit,
             buf: BytesMut::with_capacity(std::cmp::min(limit, 32768)),
-            timeout: Deadline::new(Millis(10000)),
+            timeout: Deadline::new(timeout),
         }
     }
 }
@@ -372,7 +391,7 @@ impl Future for ReadBody {
         loop {
             return match Pin::new(&mut this.stream).poll_next(cx)? {
                 Poll::Ready(Some(chunk)) => {
-                    if (this.buf.len() + chunk.len()) > this.limit {
+                    if this.limit > 0 && (this.buf.len() + chunk.len()) > this.limit {
                         Poll::Ready(Err(PayloadError::Overflow))
                     } else {
                         this.buf.extend_from_slice(&chunk);

ntex/src/http/client/sender.rs

@@ -16,8 +16,7 @@ use crate::http::encoding::Decoder;
 use crate::http::Payload;
 
 use super::error::{FreezeRequestError, InvalidUrl, SendRequestError};
-use super::response::ClientResponse;
+use super::{ClientConfig, ClientResponse};
-use super::ClientConfig;
 
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum PrepForSendingError {
@@ -136,8 +135,9 @@ impl RequestHeadType {
 
         let fut = Box::pin(async move {
             config
+                .clone()
                 .connector
-                .send_request(self, body, addr, timeout)
+                .send_request(self, body, addr, timeout, config)
                 .await
         });
 

ntex/src/http/client/test.rs

@@ -106,9 +106,9 @@ impl TestResponse {
         }
 
         if let Some(pl) = self.payload {
-            ClientResponse::new(head, pl)
+            ClientResponse::new(head, pl, Default::default())
         } else {
-            ClientResponse::new(head, h1::Payload::empty().into())
+            ClientResponse::new(head, h1::Payload::empty().into(), Default::default())
         }
     }
 }
@@ -118,8 +118,8 @@ mod tests {
     use super::*;
     use crate::http::header;
 
-    #[test]
+    #[crate::rt_test]
-    fn test_basics() {
+    async fn test_basics() {
         let res = {
             #[cfg(feature = "cookie")]
             {

ntex/src/http/config.rs

@@ -64,8 +64,8 @@ impl Default for ReadRate {
     fn default() -> Self {
         ReadRate {
             rate: 256,
-            timeout: Seconds(1),
+            timeout: Seconds(5),
-            max_timeout: Seconds(4),
+            max_timeout: Seconds(15),
         }
     }
 }
@@ -112,7 +112,7 @@ impl ServiceConfig {
             headers_read_rate: Some(ReadRate {
                 rate: 256,
                 timeout: client_timeout,
-                max_timeout: client_timeout + Seconds(3),
+                max_timeout: client_timeout + Seconds(15),
             }),
             payload_read_rate: None,
         }

ntex/src/http/h1/dispatcher.rs

@@ -895,7 +895,7 @@ where
             }
         }
 
-        log::trace!("Timeout during reading");
+        log::trace!("{}: Timeout during reading", self.io.tag());
         if self.flags.contains(Flags::READ_PL_TIMEOUT) {
             self.set_payload_error(PayloadError::Io(io::Error::new(
                 io::ErrorKind::TimedOut,

ntex/src/ws/client.rs

@@ -13,7 +13,7 @@ use nanorand::{Rng, WyRand};
 
 use crate::connect::{Connect, ConnectError, Connector};
 use crate::http::header::{self, HeaderMap, HeaderName, HeaderValue, AUTHORIZATION};
-use crate::http::{body::BodySize, client::ClientResponse, error::HttpError, h1};
+use crate::http::{body::BodySize, client, client::ClientResponse, error::HttpError, h1};
 use crate::http::{ConnectionType, RequestHead, RequestHeadType, StatusCode, Uri};
 use crate::io::{
     Base, DispatchItem, Dispatcher, DispatcherConfig, Filter, Io, Layer, Sealed,
@@ -35,6 +35,7 @@ pub struct WsClient<F, T> {
     timeout: Millis,
     extra_headers: RefCell<Option<HeaderMap>>,
     config: DispatcherConfig,
+    client_cfg: Rc<client::ClientConfig>,
     _t: marker::PhantomData<F>,
 }
 
@@ -243,7 +244,7 @@ where
         // response and ws io
         Ok(WsConnection::new(
             io,
-            ClientResponse::with_empty_payload(response),
+            ClientResponse::with_empty_payload(response, self.client_cfg.clone()),
             if server_mode {
                 ws::Codec::new().max_size(max_size)
             } else {
@@ -639,6 +640,7 @@ where
             timeout: inner.timeout,
             config: inner.config,
             extra_headers: RefCell::new(None),
+            client_cfg: Default::default(),
             _t: marker::PhantomData,
         })
     }