Update neon apis (#520)

Cargo.toml

@@ -45,6 +45,8 @@ ntex-util = { path = "ntex-util" }
 ntex-compio = { path = "ntex-compio" }
 ntex-tokio = { path = "ntex-tokio" }
 
+ntex-neon = { git = "https://github.com/ntex-rs/neon.git" }
+
 [workspace.dependencies]
 async-task = "4.5.0"
 bitflags = "2"

ntex-net/Cargo.toml

@@ -40,7 +40,7 @@ ntex-util = "2.5"
 
 ntex-tokio = { version = "0.5.3", optional = true }
 ntex-compio = { version = "0.2.4", optional = true }
-ntex-neon = { version = "0.1.0", optional = true }
+ntex-neon = { version = "0.1.1", optional = true }
 
 bitflags = { workspace = true }
 cfg-if = { workspace = true }

ntex-net/src/rt_polling/connect.rs

@@ -95,7 +95,7 @@ impl ConnectOps {
                 s
             } else {
                 let mut inner = None;
-                rt.driver().register_handler(|api| {
+                rt.driver().register(|api| {
                     let ops = Rc::new(ConnectOpsInner {
                         api,
                         connects: RefCell::new(Slab::new()),

ntex-net/src/rt_polling/driver.rs

@@ -1,6 +1,6 @@
 use std::{cell::Cell, collections::VecDeque, io, rc::Rc, task, task::Poll};
 
-use ntex_neon::driver::op::{CloseSocket, Handler, Interest};
+use ntex_neon::driver::op::{close_socket, Handler, Interest};
 use ntex_neon::driver::{AsRawFd, DriverApi, RawFd};
 use ntex_neon::{syscall, Runtime};
 use slab::Slab;
@@ -57,7 +57,7 @@ impl<T: AsRawFd + 'static> StreamOps<T> {
                 s
             } else {
                 let mut inner = None;
-                rt.driver().register_handler(|api| {
+                rt.driver().register(|api| {
                     let ops = Rc::new(StreamOpsInner {
                         api,
                         feed: Cell::new(Some(VecDeque::new())),
@@ -216,10 +216,8 @@ impl<T> StreamCtl<T> {
         let (io, fd) =
             self.with(|streams| (streams[self.id].io.take(), streams[self.id].fd));
         if let Some(io) = io {
-            let op = CloseSocket::from_raw_fd(fd);
-            let fut = ntex_neon::submit(op);
             std::mem::forget(io);
-            fut.await.0?;
+            close_socket(fd).await?;
         }
         Ok(())
     }

ntex-net/src/rt_uring/connect.rs

@@ -90,7 +90,7 @@ impl ConnectOps {
                 s
             } else {
                 let mut inner = None;
-                rt.driver().register_handler(|api| {
+                rt.driver().register(|api| {
                     let ops = Rc::new(ConnectOpsInner {
                         api,
                         ops: RefCell::new(Slab::new()),

ntex-net/src/rt_uring/driver.rs

@@ -64,7 +64,7 @@ impl<T: AsRawFd + 'static> StreamOps<T> {
                 s
             } else {
                 let mut inner = None;
-                rt.driver().register_handler(|api| {
+                rt.driver().register(|api| {
                     let mut ops = Slab::new();
                     ops.insert(Operation::Nop);
 

ntex-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-rt"
-version = "0.4.27"
+version = "0.4.28"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -42,7 +42,7 @@ tok-io = { version = "1", package = "tokio", default-features = false, features
     "net",
 ], optional = true }
 
-ntex-neon = { version = "0.1", optional = true }
+ntex-neon = { version = "0.1.1", optional = true }
 
 [dev-dependencies]
 env_logger = "0.11"

ntex-rt/src/lib.rs

@@ -258,11 +258,12 @@ mod neon {
     /// Runs the provided future, blocking the current thread until the future
     /// completes.
     pub fn block_on<F: Future<Output = ()>>(fut: F) {
+        let rt = Runtime::new().unwrap();
         log::info!(
             "Starting neon runtime, driver {:?}",
-            ntex_neon::driver::DriverType::current()
+            rt.driver().tp().name()
         );
-        let rt = Runtime::new().unwrap();
+
         rt.block_on(fut);
     }
 
@@ -376,17 +377,7 @@ mod neon {
 
     impl<T> JoinHandle<T> {
         pub fn is_finished(&self) -> bool {
-            if let Some(hnd) = &self.fut {
+            false
-                hnd.is_finished()
-            } else {
-                true
-            }
-        }
-    }
-
-    impl<T> Drop for JoinHandle<T> {
-        fn drop(&mut self) {
-            self.fut.take().unwrap().detach();
         }
     }
 
@@ -396,7 +387,8 @@ mod neon {
         fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
             Poll::Ready(
                 ready!(Pin::new(self.fut.as_mut().unwrap()).poll(cx))
-                    .map_err(|_| JoinError),
+                    .map_err(|_| JoinError)
+                    .and_then(|result| result.map_err(|_| JoinError)),
             )
         }
     }