Deprecate Arbiter's spawn related helpers (#770)

Author: fafhrd91

Cargo.toml

@@ -51,16 +51,16 @@ ntex-macros = { path = "ntex-macros" }
 ntex-util = { path = "ntex-util" }
 
 [workspace.dependencies]
-ntex = "3.4.0"
+ntex = "3.5.0"
 ntex-bytes = "1.5.0"
 ntex-codec = "1.1.0"
 ntex-io = "3.9.0"
 ntex-dispatcher = "3.1.0"
 ntex-net = "3.7.0"
 ntex-http = "1.0.0"
 ntex-router = "1.0.0"
-ntex-rt = "3.7.0"
-ntex-server = "3.8.0"
+ntex-rt = "3.8.0"
+ntex-server = "3.9.0"
 ntex-service = "4.5.0"
 ntex-tls = "3.3.0"
 ntex-macros = "3.1.0"

ntex-rt/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [3.8.0] - 2026-02-16
+
+* Export `get_item()` and `set_item()` methods
+
+* Deprecate Arbiter's spawn related helpers
+
 ## [3.7.1] - 2026-02-13
 
 * Export JoinHandle for neon runtime

ntex-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-rt"
-version = "3.7.1"
+version = "3.8.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex runtime"
 keywords = ["network", "framework", "async", "futures"]

ntex-rt/src/arbiter.rs

@@ -5,6 +5,7 @@ use std::{cell::RefCell, collections::HashMap, fmt, future::Future, pin::Pin, th
 
 use async_channel::{Receiver, Sender, unbounded};
 
+use crate::Handle;
 use crate::system::{FnExec, Id, System, SystemCommand};
 
 thread_local!(
@@ -29,7 +30,8 @@ pub struct Arbiter {
     id: usize,
     pub(crate) sys_id: usize,
     name: Arc<String>,
-    sender: Sender<ArbiterCommand>,
+    pub(crate) hnd: Option<Handle>,
+    pub(crate) sender: Sender<ArbiterCommand>,
     thread_handle: Option<thread::JoinHandle<()>>,
 }
 
@@ -47,22 +49,40 @@ impl Default for Arbiter {
 
 impl Clone for Arbiter {
     fn clone(&self) -> Self {
-        Self::with_sender(self.sys_id, self.id, self.name.clone(), self.sender.clone())
+        Self {
+            id: self.id,
+            sys_id: self.sys_id,
+            name: self.name.clone(),
+            sender: self.sender.clone(),
+            hnd: self.hnd.clone(),
+            thread_handle: None,
+        }
     }
 }
 
 impl Arbiter {
     #[allow(clippy::borrowed_box)]
-    pub(super) fn new_system(name: String) -> (Self, ArbiterController) {
+    pub(super) fn new_system(sys_id: usize, name: String) -> (Self, ArbiterController) {
         let (tx, rx) = unbounded();
 
-        let arb = Arbiter::with_sender(0, 0, Arc::new(name), tx);
+        let arb = Arbiter::with_sender(sys_id, 0, Arc::new(name), tx);
         ADDR.with(|cell| *cell.borrow_mut() = Some(arb.clone()));
         STORAGE.with(|cell| cell.borrow_mut().clear());
 
         (arb, ArbiterController { rx, stop: None })
     }
 
+    pub(super) fn dummy() -> Self {
+        Arbiter {
+            id: 0,
+            hnd: None,
+            name: String::new().into(),
+            sys_id: 0,
+            sender: unbounded().0,
+            thread_handle: None,
+        }
+    }
+
     /// Returns the current thread's arbiter's address
     ///
     /// # Panics
@@ -75,6 +95,12 @@ impl Arbiter {
         })
     }
 
+    pub(crate) fn set_current(&self) {
+        ADDR.with(|cell| {
+            *cell.borrow_mut() = Some(self.clone());
+        });
+    }
+
     /// Stop arbiter from continuing it's event loop.
     pub fn stop(&self) {
         let _ = self.sender.try_send(ArbiterCommand::Stop);
@@ -108,19 +134,23 @@ impl Arbiter {
 
         let name = name2.clone();
         let sys_id = sys.id();
+        let (arb_hnd_tx, arb_hnd_rx) = oneshot::channel();
 
         let handle = builder
             .spawn(move || {
                 log::info!("Starting {name2:?} arbiter");
 
-                let arb = Arbiter::with_sender(sys_id.0, id, name2, arb_tx);
-
                 let (stop, stop_rx) = oneshot::channel();
                 STORAGE.with(|cell| cell.borrow_mut().clear());
 
-                System::set_current(sys);
+                System::set_current(sys.clone());
+
+                crate::driver::block_on(config.runner.as_ref(), async move {
+                    let arb = Arbiter::with_sender(sys_id.0, id, name2, arb_tx);
+                    arb_hnd_tx
+                        .send(arb.hnd.clone())
+                        .expect("Controller thread has gone");
 
-                config.block_on(async move {
                     // start arbiter controller
                     crate::spawn(
                         ArbiterController {
@@ -132,7 +162,7 @@ impl Arbiter {
                     ADDR.with(|cell| *cell.borrow_mut() = Some(arb.clone()));
 
                     // register arbiter
-                    let _ = System::current()
+                    let _ = sys
                         .sys()
                         .try_send(SystemCommand::RegisterArbiter(Id(id), arb));
 
@@ -144,13 +174,18 @@ impl Arbiter {
                 let _ = System::current()
                     .sys()
                     .try_send(SystemCommand::UnregisterArbiter(Id(id)));
+
+                STORAGE.with(|cell| cell.borrow_mut().clear());
             })
             .unwrap_or_else(|err| {
                 panic!("Cannot spawn an arbiter's thread {:?}: {:?}", &name, err)
             });
 
+        let hnd = arb_hnd_rx.recv().expect("Could not start new arbiter");
+
         Arbiter {
             id,
+            hnd,
             name,
             sys_id: sys_id.0,
             sender: arb_tx2,
@@ -164,11 +199,21 @@ impl Arbiter {
         name: Arc<String>,
         sender: Sender<ArbiterCommand>,
     ) -> Self {
+        #[cfg(feature = "tokio")]
+        let hnd = { Handle::new(sender.clone()) };
+
+        #[cfg(feature = "compio")]
+        let hnd = { Handle::new(sender.clone()) };
+
+        #[cfg(all(not(feature = "compio"), not(feature = "tokio")))]
+        let hnd = { Handle::current() };
+
         Self {
             id,
             sys_id,
             name,
             sender,
+            hnd: Some(hnd),
             thread_handle: None,
         }
     }
@@ -183,6 +228,14 @@ impl Arbiter {
         self.name.as_ref()
     }
 
+    #[inline]
+    /// Handle to a runtime
+    pub fn handle(&self) -> &Handle {
+        self.hnd.as_ref().unwrap()
+    }
+
+    #[doc(hidden)]
+    #[deprecated(since = "3.8.0", note = "use `ntex_rt::spawn()`")]
     /// Send a future to the Arbiter's thread, and spawn it.
     pub fn spawn<F>(&self, future: F)
     where
@@ -193,12 +246,13 @@ impl Arbiter {
             .try_send(ArbiterCommand::Execute(Box::pin(future)));
     }
 
-    #[rustfmt::skip]
+    #[doc(hidden)]
+    #[deprecated(since = "3.8.0", note = "use `ntex_rt::Handle::spawn()`")]
     /// Send a function to the Arbiter's thread and spawns it's resulting future.
     /// This can be used to spawn non-send futures on the arbiter thread.
     pub fn spawn_with<F, R, O>(
         &self,
-        f: F
+        f: F,
     ) -> impl Future<Output = Result<O, oneshot::RecvError>> + Send + 'static
     where
         F: FnOnce() -> R + Send + 'static,
@@ -216,11 +270,15 @@ impl Arbiter {
         rx
     }
 
-    #[rustfmt::skip]
+    #[doc(hidden)]
+    #[deprecated(since = "3.8.0", note = "use `ntex_rt::Handle::spawn()`")]
     /// Send a function to the Arbiter's thread. This function will be executed asynchronously.
     /// A future is created, and when resolved will contain the result of the function sent
     /// to the Arbiters thread.
-    pub fn exec<F, R>(&self, f: F) -> impl Future<Output = Result<R, oneshot::RecvError>> + Send + 'static
+    pub fn exec<F, R>(
+        &self,
+        f: F,
+    ) -> impl Future<Output = Result<R, oneshot::RecvError>> + Send + 'static
     where
         F: FnOnce() -> R + Send + 'static,
         R: Send + 'static,
@@ -234,6 +292,8 @@ impl Arbiter {
         rx
     }
 
+    #[doc(hidden)]
+    #[deprecated(since = "3.8.0", note = "use `ntex_rt::Handle::spawn()`")]
     /// Send a function to the Arbiter's thread, and execute it. Any result from the function
     /// is discarded.
     pub fn exec_fn<F>(&self, f: F)
@@ -247,17 +307,22 @@ impl Arbiter {
             })));
     }
 
+    #[doc(hidden)]
+    #[deprecated(since = "3.8.0", note = "use `ntex_rt::set_item()`")]
     /// Set item to current arbiter's storage
     pub fn set_item<T: 'static>(item: T) {
-        STORAGE
-            .with(move |cell| cell.borrow_mut().insert(TypeId::of::<T>(), Box::new(item)));
+        set_item(item);
     }
 
+    #[doc(hidden)]
+    #[deprecated(since = "3.8.0", note = "use `ntex_rt::get_item()`")]
     /// Check if arbiter storage contains item
     pub fn contains_item<T: 'static>() -> bool {
         STORAGE.with(move |cell| cell.borrow().get(&TypeId::of::<T>()).is_some())
     }
 
+    #[doc(hidden)]
+    #[deprecated(since = "3.8.0", note = "use `ntex_rt::get_item()`")]
     /// Get a reference to a type previously inserted on this arbiter's storage
     ///
     /// # Panics
@@ -353,3 +418,22 @@ impl ArbiterController {
         }
     }
 }
+
+/// Set item to current runtime's storage
+pub fn set_item<T: 'static>(item: T) {
+    STORAGE.with(move |cell| cell.borrow_mut().insert(TypeId::of::<T>(), Box::new(item)));
+}
+
+/// Get a reference to a type previously inserted on this runtime's storage
+pub fn get_item<T: 'static, F, R>(f: F) -> R
+where
+    F: FnOnce(Option<&mut T>) -> R,
+{
+    STORAGE.with(move |cell| {
+        let mut st = cell.borrow_mut();
+        let item = st
+            .get_mut(&TypeId::of::<T>())
+            .and_then(|boxed| boxed.downcast_mut());
+        f(item)
+    })
+}