feat(fast-down/curl): fix flow control

Author: CyanChanges

Cargo.lock

@@ -26,6 +26,7 @@ kanal.workspace = true
 slab = "0.4.11"
 oneshot = "0.1.11"
 anyhow = "1.0.99"
+atomic-wait = { version = "1.1.0", optional = true }
 
 [[bin]]
 name = "libcurl-main"
@@ -34,7 +35,7 @@ path = "curl.rs"
 
 [features]
 reqwest = ["dep:reqwest"]
-curl = ["dep:curl"]
+curl = ["dep:curl", "dep:atomic-wait"]
 
 [dev-dependencies]
 tokio = { version = "1.47.1", features = [

crates/fast-down/Cargo.toml

@@ -1,5 +1,8 @@
+use fast_down::curl::worker::{DataSignal, Op, multi, options, SIG_EVENT, STATE_SEND_FAILED};
+use kanal::ReceiveError;
+use std::sync::Arc;
 use std::thread;
-use fast_down::curl::worker::{multi, Op};
+use std::time::Duration;
 
 pub fn main() {
     let (tx_ops, rx_ops) = kanal::unbounded();
@@ -8,9 +11,39 @@ pub fn main() {
     });
     let (tx_ret, ret) = oneshot::channel();
     let (tx_data, rx_data) = kanal::bounded(1);
-    tx_ops.send(Op::New(tx_data, curl::easy::List::new(), "https://1.1.1.1/cdn-cgi/trace".to_string(), tx_ret)).unwrap();
-    for data in rx_data.into_iter() {
-        dbg!(String::from_utf8_lossy(&*data));
+    let signal: Arc<DataSignal> = Default::default();
+    tx_ops
+        .send(Op::New(
+            tx_data,
+            options::New {
+                signal: signal.clone(),
+                headers: curl::easy::List::new(),
+                url: "https://github.com/".to_string(),
+                extra: None,
+            },
+            tx_ret,
+        ))
+        .unwrap();
+    let th = ret.recv().unwrap();
+    loop {
+        if signal.is_send_failed() {
+            tx_ops.send(Op::UnpauseData(th)).unwrap();
+        }
+
+        match rx_data.try_recv() {
+            Ok(Some(data)) => {
+                eprint!("{}", std::str::from_utf8(&data).unwrap());
+                thread::sleep(Duration::from_millis(100));
+            }
+            Ok(None) => {
+                if signal.is_send_failed() {
+                    tx_ops.send(Op::UnpauseData(th)).unwrap();
+                }
+                atomic_wait::wait(signal.signal(), SIG_EVENT);
+            }
+            Err(ReceiveError::SendClosed) => break,
+            Err(ReceiveError::Closed) => unreachable!(),
+        }
     }
     handle.join().unwrap();
-}
+}