Cherry pick stream upload (interfaces only) from #14 (#15)

Author: bdbai

Cargo.lock

@@ -35,12 +35,10 @@ nyquest-interface = { version = "0.1.0", path = "../../nyquest-interface", defau
 curl = { version = "0.4.47", default-features = false }
 curl-sys = { version = "0.4", default-features = false }
 iconv-native = { version = "0.1.0", optional = true, default-features = false }
-mio = { version = "1", optional = true, default-features = false }
 futures-channel = { version = "0.3", optional = true, default-features = false, features = [
     "alloc",
 ] }
 futures-util = { version = "0.3", optional = true, default-features = false, features = [
     "std",
 ] }
 slab = { version = "0.4", optional = true, default-features = false }
-memchr = "2"

backends/curl/Cargo.toml

@@ -2,13 +2,15 @@ use std::io;
 use std::task::{ready, Poll};
 use std::{pin::Pin, sync::Arc, task::Context};
 
-use curl::easy::Easy;
+use curl::easy::Easy2;
 use nyquest_interface::r#async::{futures_io, AsyncResponse};
 use nyquest_interface::Error as NyquestError;
 
-use crate::url::concat_url;
-
+mod handler;
 mod r#loop;
+mod pause;
+
+use crate::url::concat_url;
 
 pub struct CurlMultiClientInner {
     options: nyquest_interface::client::ClientOptions,
@@ -114,17 +116,20 @@ impl nyquest_interface::r#async::AsyncClient for CurlMultiClient {
         &self,
         req: nyquest_interface::r#async::Request,
     ) -> nyquest_interface::Result<Self::Response> {
-        let req = loop {
-            // TODO: CURLOPT_SHARE
-            let mut easy = Easy::new();
+        let req = {
+            let mut easy = Easy2::new(handler::AsyncHandler::default());
+            let raw_handle = easy.raw();
+            easy.get_mut().pause = Some(pause::EasyPause::new(raw_handle));
             // FIXME: properly concat base_url and url
             let url = concat_url(self.inner.options.base_url.as_deref(), &req.relative_uri);
-            crate::request::populate_request(&url, &req, &self.inner.options, &mut easy)?;
-            let req = self.inner.loop_manager.start_request(easy).await?;
-            match req {
-                r#loop::MaybeStartedRequest::Gone => {}
-                r#loop::MaybeStartedRequest::Started(req) => break req,
-            }
+            crate::request::populate_request(
+                &url,
+                req,
+                &self.inner.options,
+                &mut easy,
+                |_, _| unimplemented!(),
+            )?;
+            self.inner.loop_manager.start_request(easy).await?
         };
         let mut res = req.wait_for_response().await?;
         res.max_response_buffer_size = self.inner.options.max_response_buffer_size;
@@ -141,8 +146,8 @@ impl nyquest_interface::r#async::AsyncBackend for crate::CurlBackend {
     ) -> Result<Self::AsyncClient, NyquestError> {
         Ok(CurlMultiClient {
             inner: Arc::new(CurlMultiClientInner {
-                options,
                 loop_manager: r#loop::LoopManager::new(),
+                options,
             }),
         })
     }

backends/curl/src/async.rs

@@ -0,0 +1,64 @@
+use std::sync::Arc;
+
+use curl::easy::{Handler, WriteError};
+
+use super::pause::EasyPause;
+use super::r#loop::SharedRequestContext;
+
+#[derive(Default)]
+pub(super) struct AsyncHandler {
+    // To be filled in the loop
+    pub(super) ctx: Option<Arc<SharedRequestContext>>,
+    // To be filled after Easy2 is constructed
+    pub(super) pause: Option<EasyPause>,
+}
+
+struct AsyncHandlerRef<'a> {
+    ctx: &'a SharedRequestContext,
+    pause: &'a mut EasyPause,
+}
+
+impl AsyncHandler {
+    fn get_ref(&mut self) -> Option<AsyncHandlerRef<'_>> {
+        let ctx = self.ctx.as_ref()?;
+        let pause = self.pause.as_mut()?;
+        Some(AsyncHandlerRef { ctx, pause })
+    }
+}
+
+impl Handler for AsyncHandler {
+    fn write(&mut self, data: &[u8]) -> Result<usize, WriteError> {
+        let Some(inner) = self.get_ref() else {
+            // ... signals an error condition to the library and returns CURLE_WRITE_ERROR.
+            return Ok(0);
+        };
+        {
+            let mut state = inner.ctx.state.lock().unwrap();
+            let state = &mut state.0;
+            state.write_data(data);
+        }
+        unsafe {
+            inner.pause.pause_recv();
+        }
+        inner.ctx.waker.wake();
+        Ok(data.len())
+    }
+
+    fn header(&mut self, data: &[u8]) -> bool {
+        let Some(inner) = self.get_ref() else {
+            // ... signals an error condition to the library and returns CURLE_WRITE_ERROR.
+            return false;
+        };
+        {
+            let mut state = inner.ctx.state.lock().unwrap();
+            let state = &mut state.0;
+            if state.push_header_data(data) {
+                unsafe {
+                    inner.pause.pause_recv();
+                }
+            }
+        }
+        inner.ctx.waker.wake();
+        true
+    }
+}