Deprecate Arbiter's spawn related helpers

Author: fafhrd91

Cargo.toml

@@ -59,7 +59,7 @@ 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-rt = "3.8.0"
 ntex-server = "3.8.0"
 ntex-service = "4.5.0"
 ntex-tls = "3.3.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,6 +30,7 @@ pub struct Arbiter {
     id: usize,
     pub(crate) sys_id: usize,
     name: Arc<String>,
+    hnd: Handle,
     sender: Sender<ArbiterCommand>,
     thread_handle: Option<thread::JoinHandle<()>>,
 }
@@ -108,19 +110,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);
 
-                config.block_on(async move {
+                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");
+
                     // start arbiter controller
                     crate::spawn(
                         ArbiterController {
@@ -149,8 +155,11 @@ impl Arbiter {
                 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,
@@ -169,6 +178,7 @@ impl Arbiter {
             sys_id,
             name,
             sender,
+            hnd: Handle::current(),
             thread_handle: None,
         }
     }
@@ -183,6 +193,13 @@ impl Arbiter {
         self.name.as_ref()
     }
 
+    /// Handle to a runtime
+    pub fn handle(&self) -> &Handle {
+        &self.hnd
+    }
+
+    #[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 +210,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 +234,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 +256,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 +271,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 +382,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)
+    })
+}

ntex-rt/src/builder.rs

@@ -108,43 +108,26 @@ impl Builder {
     ///
     /// This method panics if it can not create tokio runtime
     pub fn build<R: Runner>(self, runner: R) -> SystemRunner {
-        let name = self.name.clone();
-        let testing = self.testing;
-        let stack_size = self.stack_size;
-        let stop_on_panic = self.stop_on_panic;
-
-        self.build_with(SystemConfig {
-            name,
-            testing,
-            stack_size,
-            stop_on_panic,
+        let config = SystemConfig {
+            name: self.name.clone(),
+            testing: self.testing,
+            stack_size: self.stack_size,
+            stop_on_panic: self.stop_on_panic,
+            ping_interval: self.ping_interval,
+            pool_limit: self.pool_limit,
+            pool_recv_timeout: self.pool_recv_timeout,
             runner: Arc::new(runner),
-        })
+        };
+        self.build_with(config)
     }
 
     /// Create new System.
     ///
     /// This method panics if it can not create tokio runtime
     pub fn build_with(self, config: SystemConfig) -> SystemRunner {
-        let (stop_tx, stop) = oneshot::channel();
-        let (sys_sender, sys_receiver) = unbounded();
-
-        // thread pool
-        let pool = ThreadPool::new(&self.name, self.pool_limit, self.pool_recv_timeout);
-
-        let (arb, controller) = Arbiter::new_system(self.name.clone());
-        let _ = sys_sender.try_send(SystemCommand::RegisterArbiter(arb.id(), arb.clone()));
-        let system = System::construct(sys_sender, arb, config.clone(), pool);
-
-        // system arbiter
-        let support = SystemSupport::new(stop_tx, sys_receiver, self.ping_interval);
-
         // init system arbiter and run configuration method
         SystemRunner {
-            stop,
-            support,
-            controller,
-            system,
+            runner: config.runner.clone(),
             config,
             _t: PhantomData,
         }
@@ -154,20 +137,12 @@ impl Builder {
 /// Helper object that runs System's event loop
 #[must_use = "SystemRunner must be run"]
 pub struct SystemRunner {
-    stop: oneshot::Receiver<i32>,
-    support: SystemSupport,
-    controller: ArbiterController,
-    system: System,
     config: SystemConfig,
+    runner: Arc<dyn Runner>,
     _t: PhantomData<Rc<()>>,
 }
 
 impl SystemRunner {
-    /// Get current system.
-    pub fn system(&self) -> System {
-        self.system.clone()
-    }
-
     /// This function will start event loop and will finish once the
     /// `System::stop()` function is called.
     pub fn run_until_stop(self) -> io::Result<()> {
@@ -182,16 +157,12 @@ impl SystemRunner {
     {
         log::info!("Starting {:?} system", self.config.name);
 
-        let SystemRunner {
-            controller,
-            stop,
-            support,
-            config,
-            ..
-        } = self;
+        let SystemRunner { config, runner, .. } = self;
 
         // run loop
-        config.block_on(async move {
+        crate::driver::block_on(runner.as_ref(), async move {
+            let (_, support, controller, stop) = create(&config);
+
             f()?;
 
             crate::spawn(support.run());
@@ -215,14 +186,11 @@ impl SystemRunner {
         F: Future<Output = R> + 'static,
         R: 'static,
     {
-        let SystemRunner {
-            controller,
-            support,
-            config,
-            ..
-        } = self;
+        let SystemRunner { config, runner, .. } = self;
+
+        crate::driver::block_on(runner.as_ref(), async move {
+            let (_, support, controller, _) = create(&config);
 
-        config.block_on(async move {
             crate::spawn(support.run());
             crate::spawn(controller.run());
             fut.await
@@ -236,15 +204,13 @@ impl SystemRunner {
         F: Future<Output = R> + 'static,
         R: 'static,
     {
-        let SystemRunner {
-            controller,
-            support,
-            ..
-        } = self;
+        let SystemRunner { config, .. } = self;
 
         // run loop
         tok_io::task::LocalSet::new()
             .run_until(async move {
+                let (sys, support, controller, _) = create(&config);
+
                 crate::spawn(support.run());
                 crate::spawn(controller.run());
                 fut.await
@@ -256,9 +222,34 @@ impl SystemRunner {
 impl fmt::Debug for SystemRunner {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("SystemRunner")
-            .field("system", &self.system)
-            .field("support", &self.support)
             .field("config", &self.config)
             .finish()
     }
 }
+
+/// Create new System.
+///
+/// This method panics if it can not create tokio runtime
+fn create(
+    config: &SystemConfig,
+) -> (
+    System,
+    SystemSupport,
+    ArbiterController,
+    oneshot::Receiver<i32>,
+) {
+    let (stop_tx, stop) = oneshot::channel();
+    let (sys_sender, sys_receiver) = unbounded();
+
+    // thread pool
+    let pool = ThreadPool::new(&config.name, config.pool_limit, config.pool_recv_timeout);
+
+    let (arb, controller) = Arbiter::new_system(config.name.clone());
+    let _ = sys_sender.try_send(SystemCommand::RegisterArbiter(arb.id(), arb.clone()));
+    let system = System::construct(sys_sender, arb, config.clone(), pool);
+
+    // system arbiter
+    let support = SystemSupport::new(stop_tx, sys_receiver, config.ping_interval);
+
+    (system, support, controller, stop)
+}

ntex-rt/src/driver.rs

@@ -1,5 +1,5 @@
 //! The platform-specified driver.
-use std::{fmt, io, pin::Pin};
+use std::{cell::RefCell, fmt, io, pin::Pin, rc::Rc};
 
 use crate::rt::Runtime;
 
@@ -102,3 +102,20 @@ macro_rules! syscall {
         }
     }};
 }
+
+/// Execute a future with custom `block_on` method and wait for result
+pub(super) fn block_on<F, R>(run: &dyn Runner, fut: F) -> R
+where
+    F: Future<Output = R> + 'static,
+    R: 'static,
+{
+    // run loop
+    let result = Rc::new(RefCell::new(None));
+    let result_inner = result.clone();
+
+    run.block_on(Box::pin(async move {
+        let r = fut.await;
+        *result_inner.borrow_mut() = Some(r);
+    }));
+    result.borrow_mut().take().unwrap()
+}