ntex_net/
lib.rs

1//! Utility for async runtime abstraction
2#![deny(clippy::pedantic)]
3#![allow(
4    clippy::clone_on_copy,
5    clippy::cast_possible_truncation,
6    clippy::missing_fields_in_debug,
7    clippy::must_use_candidate,
8    clippy::missing_errors_doc,
9    clippy::missing_panics_doc
10)]
11use std::{io, net, net::SocketAddr, panic};
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(windows)]
27pub mod iocp;
28
29#[cfg(any(unix, windows))]
30mod helpers;
31
32#[cfg(feature = "tokio")]
33pub mod tokio;
34
35#[cfg(feature = "compio")]
36mod compio;
37
38#[allow(clippy::wrong_self_convention)]
39pub trait Reactor: Driver {
40    fn tcp_connect(&self, addr: net::SocketAddr, cfg: SharedCfg) -> channel::Receiver<Io>;
41
42    fn unix_connect(
43        &self,
44        addr: std::path::PathBuf,
45        cfg: SharedCfg,
46    ) -> channel::Receiver<Io>;
47
48    /// Convert std `TcpStream` to `Io`
49    fn from_tcp_stream(&self, stream: net::TcpStream, cfg: SharedCfg) -> io::Result<Io>;
50
51    #[cfg(unix)]
52    /// Convert std `UnixStream` to `Io`
53    fn from_unix_stream(
54        &self,
55        _: std::os::unix::net::UnixStream,
56        _: SharedCfg,
57    ) -> io::Result<Io>;
58}
59
60#[inline]
61/// Opens a TCP connection to a remote host.
62pub async fn tcp_connect(addr: SocketAddr, cfg: SharedCfg) -> io::Result<Io> {
63    with_current(|driver| driver.tcp_connect(addr, cfg)).await
64}
65
66#[inline]
67/// Opens a unix stream connection.
68pub async fn unix_connect<'a, P>(addr: P, cfg: SharedCfg) -> io::Result<Io>
69where
70    P: AsRef<std::path::Path> + 'a,
71{
72    with_current(|driver| driver.unix_connect(addr.as_ref().into(), cfg)).await
73}
74
75#[inline]
76/// Convert std `TcpStream` to `TcpStream`
77pub fn from_tcp_stream(stream: net::TcpStream, cfg: SharedCfg) -> io::Result<Io> {
78    with_current(|driver| driver.from_tcp_stream(stream, cfg))
79}
80
81#[cfg(unix)]
82#[inline]
83/// Convert std `UnixStream` to `UnixStream`
84pub fn from_unix_stream(
85    stream: std::os::unix::net::UnixStream,
86    cfg: SharedCfg,
87) -> io::Result<Io> {
88    with_current(|driver| driver.from_unix_stream(stream, cfg))
89}
90
91fn with_current<T, F: FnOnce(&dyn Reactor) -> T>(f: F) -> T {
92    #[cold]
93    fn not_in_ntex_driver() -> ! {
94        panic!("not in a ntex driver")
95    }
96
97    if CURRENT_DRIVER.is_set() {
98        CURRENT_DRIVER.with(|d| f(&**d))
99    } else {
100        not_in_ntex_driver()
101    }
102}
103
104scoped_tls::scoped_thread_local!(static CURRENT_DRIVER: Box<dyn Reactor>);
105
106#[derive(Debug)]
107pub struct DefaultRuntime;
108
109impl Runner for DefaultRuntime {
110    #[allow(unused_variables, clippy::too_many_lines)]
111    fn block_on(&self, fut: BlockFuture) {
112        #[cfg(feature = "tokio")]
113        {
114            let driver: Box<dyn Reactor> = Box::new(self::tokio::TokioDriver);
115
116            CURRENT_DRIVER.set(&driver, || {
117                crate::tokio::block_on(fut);
118                unsafe {
119                    ntex_rt::remove_all_items();
120                }
121            });
122        }
123
124        #[cfg(all(feature = "compio", not(feature = "tokio")))]
125        {
126            let driver: Box<dyn Reactor> = Box::new(self::compio::CompioDriver);
127
128            CURRENT_DRIVER.set(&driver, || {
129                crate::compio::block_on(fut);
130                unsafe {
131                    ntex_rt::remove_all_items();
132                }
133            });
134        }
135
136        #[cfg(all(windows, not(feature = "tokio"), not(feature = "compio")))]
137        {
138            let driver = crate::iocp::Driver::new().expect("Cannot construct driver");
139            let driver: Box<dyn Reactor> = Box::new(driver);
140
141            CURRENT_DRIVER.set(&driver, || {
142                let res = panic::catch_unwind(panic::AssertUnwindSafe(|| {
143                    let rt = ntex_rt::Runtime::new(driver.handle());
144                    rt.block_on(fut, &*driver);
145                }));
146                unsafe {
147                    if let Err(err) = res {
148                        ntex_rt::remove_all_items();
149                        panic::resume_unwind(err);
150                    } else {
151                        ntex_rt::remove_all_items();
152                    }
153                }
154            });
155        }
156
157        #[cfg(all(unix, not(feature = "tokio"), not(feature = "compio")))]
158        {
159            #[cfg(feature = "neon-polling")]
160            {
161                let driver =
162                    crate::polling::Driver::new().expect("Cannot construct driver");
163                let driver: Box<dyn Reactor> = Box::new(driver);
164
165                CURRENT_DRIVER.set(&driver, || {
166                    let res = panic::catch_unwind(panic::AssertUnwindSafe(|| {
167                        let rt = ntex_rt::Runtime::new(driver.handle());
168                        rt.block_on(fut, &*driver);
169                    }));
170                    unsafe {
171                        if let Err(err) = res {
172                            ntex_rt::remove_all_items();
173                            panic::resume_unwind(err);
174                        } else {
175                            ntex_rt::remove_all_items();
176                        }
177                    }
178                });
179            }
180
181            #[cfg(all(target_os = "linux", feature = "neon-uring"))]
182            {
183                let driver =
184                    crate::uring::Driver::new(2048).expect("Cannot construct driver");
185                let driver: Box<dyn Reactor> = Box::new(driver);
186
187                CURRENT_DRIVER.set(&driver, || {
188                    let res = panic::catch_unwind(panic::AssertUnwindSafe(|| {
189                        let rt = ntex_rt::Runtime::new(driver.handle());
190                        rt.block_on(fut, &*driver);
191                    }));
192                    unsafe {
193                        if let Err(err) = res {
194                            ntex_rt::remove_all_items();
195                            panic::resume_unwind(err);
196                        } else {
197                            ntex_rt::remove_all_items();
198                        }
199                    }
200                });
201            }
202
203            #[cfg(all(not(feature = "neon-uring"), not(feature = "neon-polling")))]
204            {
205                #[cfg(target_os = "linux")]
206                let driver: Box<dyn Reactor> =
207                    if let Ok(driver) = crate::uring::Driver::new(2048) {
208                        Box::new(driver)
209                    } else {
210                        Box::new(
211                            crate::polling::Driver::new().expect("Cannot construct driver"),
212                        )
213                    };
214
215                #[cfg(not(target_os = "linux"))]
216                let driver: Box<dyn Reactor> = Box::new(
217                    crate::polling::Driver::new().expect("Cannot construct driver"),
218                );
219
220                CURRENT_DRIVER.set(&driver, || {
221                    let res = panic::catch_unwind(panic::AssertUnwindSafe(|| {
222                        let rt = ntex_rt::Runtime::new(driver.handle());
223                        rt.block_on(fut, &*driver);
224                    }));
225                    unsafe {
226                        if let Err(err) = res {
227                            ntex_rt::remove_all_items();
228                            panic::resume_unwind(err);
229                        } else {
230                            ntex_rt::remove_all_items();
231                        }
232                    }
233                });
234            }
235        }
236    }
237}
ntex_net/lib.rs

ntex_net/
lib.rs
