ntex_net/
lib.rs

1//! Utility for async runtime abstraction
2#![deny(clippy::pedantic)]
3#![allow(
4    clippy::missing_fields_in_debug,
5    clippy::must_use_candidate,
6    clippy::return_self_not_must_use,
7    clippy::missing_errors_doc,
8    clippy::missing_panics_doc,
9    clippy::cast_possible_truncation
10)]
11use std::{io, net, net::SocketAddr};
12
13use ntex_io::Io;
14use ntex_rt::{BlockFuture, Driver, Runner};
15use ntex_service::cfg::SharedCfg;
16
17pub mod channel;
18pub mod connect;
19
20#[cfg(unix)]
21pub mod polling;
22
23#[cfg(target_os = "linux")]
24pub mod uring;
25
26#[cfg(unix)]
27mod helpers;
28
29#[cfg(feature = "tokio")]
30pub mod tokio;
31
32#[cfg(feature = "compio")]
33mod compio;
34
35#[allow(clippy::wrong_self_convention)]
36pub trait Reactor: Driver {
37    fn tcp_connect(&self, addr: net::SocketAddr, cfg: SharedCfg) -> channel::Receiver<Io>;
38
39    fn unix_connect(
40        &self,
41        addr: std::path::PathBuf,
42        cfg: SharedCfg,
43    ) -> channel::Receiver<Io>;
44
45    /// Convert std `TcpStream` to `Io`
46    fn from_tcp_stream(&self, stream: net::TcpStream, cfg: SharedCfg) -> io::Result<Io>;
47
48    #[cfg(unix)]
49    /// Convert std `UnixStream` to `Io`
50    fn from_unix_stream(
51        &self,
52        _: std::os::unix::net::UnixStream,
53        _: SharedCfg,
54    ) -> io::Result<Io>;
55}
56
57#[inline]
58/// Opens a TCP connection to a remote host.
59pub async fn tcp_connect(addr: SocketAddr, cfg: SharedCfg) -> io::Result<Io> {
60    with_current(|driver| driver.tcp_connect(addr, cfg)).await
61}
62
63#[inline]
64/// Opens a unix stream connection.
65pub async fn unix_connect<'a, P>(addr: P, cfg: SharedCfg) -> io::Result<Io>
66where
67    P: AsRef<std::path::Path> + 'a,
68{
69    with_current(|driver| driver.unix_connect(addr.as_ref().into(), cfg)).await
70}
71
72#[inline]
73/// Convert std `TcpStream` to `TcpStream`
74pub fn from_tcp_stream(stream: net::TcpStream, cfg: SharedCfg) -> io::Result<Io> {
75    with_current(|driver| driver.from_tcp_stream(stream, cfg))
76}
77
78#[cfg(unix)]
79#[inline]
80/// Convert std `UnixStream` to `UnixStream`
81pub fn from_unix_stream(
82    stream: std::os::unix::net::UnixStream,
83    cfg: SharedCfg,
84) -> io::Result<Io> {
85    with_current(|driver| driver.from_unix_stream(stream, cfg))
86}
87
88fn with_current<T, F: FnOnce(&dyn Reactor) -> T>(f: F) -> T {
89    #[cold]
90    fn not_in_ntex_driver() -> ! {
91        panic!("not in a ntex driver")
92    }
93
94    if CURRENT_DRIVER.is_set() {
95        CURRENT_DRIVER.with(|d| f(&**d))
96    } else {
97        not_in_ntex_driver()
98    }
99}
100
101scoped_tls::scoped_thread_local!(static CURRENT_DRIVER: Box<dyn Reactor>);
102
103#[derive(Debug)]
104pub struct DefaultRuntime;
105
106impl Runner for DefaultRuntime {
107    fn block_on(&self, fut: BlockFuture) {
108        #[cfg(feature = "tokio")]
109        {
110            let driver: Box<dyn Reactor> = Box::new(self::tokio::TokioDriver);
111
112            CURRENT_DRIVER.set(&driver, || {
113                crate::tokio::block_on(fut);
114            });
115        }
116
117        #[cfg(all(feature = "compio", not(feature = "tokio")))]
118        {
119            let driver: Box<dyn Reactor> = Box::new(self::compio::CompioDriver);
120
121            CURRENT_DRIVER.set(&driver, || {
122                crate::compio::block_on(fut);
123            });
124        }
125
126        #[cfg(all(not(feature = "tokio"), not(feature = "compio")))]
127        {
128            #[cfg(feature = "neon-polling")]
129            {
130                let driver =
131                    crate::polling::Driver::new().expect("Cannot construct driver");
132                let driver: Box<dyn Reactor> = Box::new(driver);
133
134                CURRENT_DRIVER.set(&driver, || {
135                    let rt = ntex_rt::Runtime::new(driver.handle());
136                    rt.block_on(fut, &*driver);
137                    driver.clear();
138                });
139            }
140
141            #[cfg(all(target_os = "linux", feature = "neon-uring"))]
142            {
143                let driver =
144                    crate::uring::Driver::new(2048).expect("Cannot construct driver");
145                let driver: Box<dyn Reactor> = Box::new(driver);
146
147                CURRENT_DRIVER.set(&driver, || {
148                    let rt = ntex_rt::Runtime::new(driver.handle());
149                    rt.block_on(fut, &*driver);
150                    driver.clear();
151                });
152            }
153
154            #[cfg(all(not(feature = "neon-uring"), not(feature = "neon-polling")))]
155            {
156                #[cfg(target_os = "linux")]
157                let driver: Box<dyn Reactor> =
158                    if let Ok(driver) = crate::uring::Driver::new(2048) {
159                        Box::new(driver)
160                    } else {
161                        Box::new(
162                            crate::polling::Driver::new().expect("Cannot construct driver"),
163                        )
164                    };
165
166                #[cfg(not(target_os = "linux"))]
167                let driver: Box<dyn Reactor> = Box::new(
168                    crate::polling::Driver::new().expect("Cannot construct driver"),
169                );
170
171                CURRENT_DRIVER.set(&driver, || {
172                    let rt = ntex_rt::Runtime::new(driver.handle());
173                    rt.block_on(fut, &*driver);
174                    driver.clear();
175                });
176            }
177        }
178    }
179}
ntex_net/lib.rs

ntex_net/
lib.rs
