backend/curl: fix stream read unlimited buffering (#22)

* curl fix stream read unlimited buffering

* nsurlsession fix std::io

Author: bdbai

backends/curl/src/async.rs

@@ -103,6 +103,9 @@ impl futures_io::AsyncRead for CurlAsyncResponse {
         Poll::Ready(match poll_res {
             Ok(None) => Ok(0),
             Ok(Some(read_len)) => Ok(read_len),
+            Err(NyquestError::RequestTimeout) => {
+                return Poll::Ready(Err(io::ErrorKind::TimedOut.into()))
+            }
             Err(NyquestError::Io(e)) => return Poll::Ready(Err(e)),
             Err(e) => unreachable!("Unexpected error: {}", e),
         })

backends/curl/src/async/loop.rs

@@ -128,7 +128,7 @@ impl RequestHandle {
     pub(super) fn poll_bytes<T>(
         &mut self,
         cx: &mut Context<'_>,
-        mut cb: impl FnMut(&mut Vec<u8>) -> nyquest_interface::Result<T>,
+        cb: impl FnOnce(&mut Vec<u8>) -> nyquest_interface::Result<T>,
     ) -> Poll<nyquest_interface::Result<Option<T>>> {
         let mut state = self.shared_context.state.lock().unwrap();
         if !state.0.response_buffer.is_empty() {

backends/curl/src/blocking.rs

@@ -93,18 +93,18 @@ impl CurlEasyClient {
 impl io::Read for CurlBlockingResponse {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         let handle = self.handle.handle_mut();
-        match handle.poll_until_partial_response() {
-            Ok(()) => {}
-            Err(NyquestError::Io(e)) => return Err(e),
-            Err(e) => unreachable!("Unexpected error: {e:?}"),
-        }
-        let written = handle.with_response_buffer_mut(|response_buf| {
+        let res = handle.poll_bytes(|response_buf| {
             let len = response_buf.len().min(buf.len());
             buf[..len].copy_from_slice(&response_buf[..len]);
             response_buf.drain(..len);
             len
         });
-        Ok(written)
+        match res {
+            Ok(len) => Ok(len),
+            Err(NyquestError::RequestTimeout) => Err(io::ErrorKind::TimedOut.into()),
+            Err(NyquestError::Io(e)) => Err(e),
+            Err(e) => unreachable!("Unexpected error: {}", e),
+        }
     }
 }
 

backends/curl/src/blocking/multi_easy.rs

@@ -196,15 +196,22 @@ impl MultiEasy {
         .map(|_| ())
     }
 
-    pub fn poll_until_partial_response(&mut self) -> NyquestResult<()> {
+    pub fn poll_bytes<T>(&mut self, cb: impl FnOnce(&mut Vec<u8>) -> T) -> NyquestResult<T> {
+        {
+            let mut state = self.state.lock().unwrap();
+            if !state.response_buffer.is_empty() {
+                return Ok(cb(&mut state.response_buffer));
+            }
+        }
         self.poll_until(|state| {
             let is_empty = state.lock().unwrap().response_buffer.is_empty();
             Ok(if is_empty {
                 ControlFlow::Continue(())
             } else {
                 ControlFlow::Break(())
             })
-        })
+        })?;
+        Ok(self.with_response_buffer_mut(cb))
     }
 
     pub fn take_response_buffer(&mut self) -> Vec<u8> {

backends/nsurlsession/src/blocking.rs

@@ -48,6 +48,9 @@ impl std::io::Read for NSUrlSessionBlockingResponse {
                 Ok(read_len @ 1..) => {
                     return Ok(read_len);
                 }
+                Err(NyquestError::RequestTimeout) => {
+                    return Err(std::io::ErrorKind::TimedOut.into())
+                }
                 Err(NyquestError::Io(e)) => return Err(e),
                 Err(e) => unreachable!("Unexpected error: {e}"),
                 Ok(0) if inner.shared.is_completed() => return Ok(0),