fix: ntex update new task system

Author: 0xle0ne

Makefile.toml

@@ -48,6 +48,7 @@ args = [
   "--",
   "--test-threads",
   "1",
+  "--nocapture",
 ]
 
 # Test nanocld_client

bin/nanocld/src/services/event/mod.rs

@@ -81,7 +81,10 @@ mod tests {
         "watch events condition"
       );
       let mut stream = res.into_stream();
-      while (stream.next().await).is_some() {}
+      while let Some(Ok(event)) = stream.next().await {
+        let string = String::from_utf8_lossy(&event);
+        log::info!("Received event: {string}");
+      }
     });
     let cargo = CargoSpecPartial {
       name: CARGO_NAME.to_owned(),

bin/nanocld/src/tasks/task_manager.rs

@@ -70,11 +70,20 @@ impl TaskManager {
 
   pub async fn remove_task(&self, key: &str) {
     let mut tasks = self.tasks.lock().await;
-    let task = tasks.get(key);
-    if let Some(task) = task {
-      task.fut.abort();
-      log::debug!("Removing task: {key} {}", task.kind);
-      tasks.remove(key);
+    if let Some(task) = tasks.remove(key) {
+      let kind = task.kind.clone();
+      if !task.fut.is_finished() {
+        log::debug!("Cancelling task: {key} {kind}");
+        match Arc::try_unwrap(task.fut) {
+          Ok(handle) => handle.cancel(),
+          Err(_handle) => {
+            log::debug!(
+              "Task cancellation skipped for {key} {kind} because handle is shared",
+            );
+          }
+        }
+      }
+      log::debug!("Removing task: {key} {kind}");
     }
   }
 

bin/ncproxy/src/models/system.rs

@@ -43,10 +43,13 @@ impl SystemEvent {
   }
 
   pub fn handle(&mut self, _e: SystemEventKind) {
-    let abort_handle = self.0.task.abort_handle();
-    if !abort_handle.is_finished() {
-      log::info!("system: aborting reload task");
-      abort_handle.abort();
+    if !self.0.task.is_finished() {
+      log::info!("system: canceling reload task");
+      let task = std::mem::replace(
+        &mut self.0.task,
+        rt::spawn(async move { Ok::<_, IoError>(()) }),
+      );
+      task.cancel();
     }
     let client = self.0.client.clone();
     self.0.task = rt::spawn(async move {