backends/feat: stream upload (#14)

* winrt blocking stream upload

* winrt async stream upload

* curl Easy2 (WIP)

* curl Easy2

* curl request context shared

* winrt fix multipart upload

* curl refactor (WIP)

* refactor unsized stream

* nsurlsession add dummy input stream

* nsurlsession pub crate input stream ivars

* nsurlsession do not override cfstream stuff

* nsurlsession retain runloop

* nsurlsession async stream

* nsurlsession fix body stream

* nsurlsession do not poll stream on eof

* curl stream upload (WIP)

async unsized mime stream is not working?

* [skip ci] curl revert debug eprint

* reqwest stream upload (non-wasm)

* curl fix multipart unsized upload

* interface avoid changess for utils

* nsurlsession multipart stream

* curl fix missing futures-io

* curl fix missing futures-util io

* nsurlsession fix missing futures-util io

* update README

Author: bdbai

Cargo.lock

@@ -59,22 +59,19 @@ The following items are planned in MVP:
 - [x] Nyquest blocking API
 - [x] Nyquest async API
 - [x] Backend: WinRT HttpClient
-  - [x] Blocking
-  - [x] Async
 - [x] Backend: libcurl
-  - [x] Blocking
-  - [x] Async
 - [x] Backend: NSURLSession
 - [x] Backend: reqwest (with WASM support)
 - [x] Client Options
 - [x] Streaming download
+- [x] Streaming upload
 - [x] Test framework for backends
 - [x] Presets
 - [x] Documentation
 
 Future work may include:
 
-- [ ] Streaming upload
+- [ ] WebSocket
 - [ ] Backend: WASM fetch
 - [ ] Cookie management
 - [ ] Progress tracking

README.md

@@ -41,6 +41,7 @@ futures-channel = { version = "0.3", optional = true, default-features = false,
 ] }
 futures-util = { version = "0.3", optional = true, default-features = false, features = [
     "std",
+    "io",
 ] }
 slab = { version = "0.4", optional = true, default-features = false }
 pin-project-lite = "0.2"

backends/curl/Cargo.toml

@@ -1,20 +1,27 @@
+use std::cell::RefCell;
 use std::io;
+use std::pin::pin;
 use std::task::{ready, Poll};
 use std::{pin::Pin, sync::Arc, task::Context};
 
+use futures_util::future::{select, Either};
 use nyquest_interface::r#async::{futures_io, AsyncResponse};
 use nyquest_interface::Error as NyquestError;
 
 mod handler;
 mod r#loop;
 mod pause;
+mod read_task;
 mod set;
+mod shared;
 
 use crate::curl_ng::easy::{AsRawEasyMut as _, Share};
 use crate::r#async::handler::AsyncHandler;
 use crate::request::{create_easy, AsCallbackMut as _};
 use crate::url::concat_url;
 
+type Easy = crate::request::BoxEasyHandle<handler::AsyncHandler>;
+
 pub struct CurlMultiClientInner {
     options: nyquest_interface::client::ClientOptions,
     loop_manager: r#loop::LoopManager,
@@ -123,16 +130,40 @@ impl nyquest_interface::r#async::AsyncClient for CurlMultiClient {
         &self,
         req: nyquest_interface::r#async::Request,
     ) -> nyquest_interface::Result<Self::Response> {
-        let req = {
+        let (req, read_task_collection) = {
             let mut easy = create_easy(AsyncHandler::default(), &self.inner.share)?;
             let raw = easy.as_mut().as_raw_easy_mut().raw();
             easy.as_callback_mut().pause = Some(pause::EasyPause::new(raw));
             // 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, easy.as_mut())?;
-            self.inner.loop_manager.start_request(easy).await?
+            let req_ctx = easy.as_callback_mut().ctx.clone();
+            let read_task_collection = RefCell::new(read_task::ReadTaskCollection::new(req_ctx));
+            crate::request::populate_request(
+                &url,
+                req,
+                &self.inner.options,
+                easy.as_mut(),
+                |easy, stream| {
+                    read_task_collection
+                        .borrow_mut()
+                        .add_in_handler(easy, stream)
+                },
+                |stream| {
+                    read_task_collection
+                        .borrow_mut()
+                        .add_mime_part_reader(stream)
+                },
+            )?;
+            let req = self.inner.loop_manager.start_request(easy).await?;
+            (req, read_task_collection.into_inner())
+        };
+        let res_task = pin!(req.wait_for_response());
+        let read_task_collection = pin!(read_task_collection.execute(&self.inner.loop_manager));
+        let mut res = match select(res_task, read_task_collection).await {
+            Either::Left((res, _)) => res?,
+            Either::Right((Err(e), _)) => return Err(e),
+            Either::Right((Ok(_), _)) => unreachable!(),
         };
-        let mut res = req.wait_for_response().await?;
         res.max_response_buffer_size = self.inner.options.max_response_buffer_size;
         Ok(res)
     }

backends/curl/src/async.rs

@@ -3,19 +3,18 @@ use std::sync::Arc;
 use curl::easy::WriteError;
 
 use super::pause::EasyPause;
-use super::r#loop::SharedRequestContext;
+use super::shared::SharedRequestStates;
 use crate::curl_ng::easy::EasyCallback;
 
 #[derive(Default)]
 pub(super) struct AsyncHandler {
-    // To be filled in the loop
-    pub(super) ctx: Arc<SharedRequestContext>,
+    pub(super) ctx: Arc<SharedRequestStates>,
     // To be filled after Easy2 is constructed
     pub(super) pause: Option<EasyPause>,
 }
 
 struct AsyncHandlerRef<'a> {
-    ctx: &'a SharedRequestContext,
+    ctx: &'a SharedRequestStates,
     pause: &'a mut EasyPause,
 }
 
@@ -63,11 +62,20 @@ impl EasyCallback for AsyncHandler {
         true
     }
 
-    fn read(&mut self, _buf: &mut [u8]) -> Result<usize, curl::easy::ReadError> {
-        unimplemented!()
+    fn read(&mut self, buf: &mut [u8]) -> Result<usize, curl::easy::ReadError> {
+        let mut state = self.ctx.state.lock().unwrap();
+        let stream = state
+            .req_streams
+            .get_mut(0)
+            .ok_or(curl::easy::ReadError::Abort)?;
+        stream.read(buf, &self.ctx)
     }
 
-    fn seek(&mut self, _whence: std::io::SeekFrom) -> curl::easy::SeekResult {
-        unimplemented!()
+    fn seek(&mut self, whence: std::io::SeekFrom) -> curl::easy::SeekResult {
+        let mut state = self.ctx.state.lock().unwrap();
+        let Some(stream) = state.req_streams.get_mut(0) else {
+            return curl::easy::SeekResult::Fail;
+        };
+        stream.seek(whence, &self.ctx)
     }
 }