From 689ea96c879ba331b44eea068f5956a67ea7a1d7 Mon Sep 17 00:00:00 2001 From: plusls Date: Thu, 7 May 2026 10:51:23 +0000 Subject: [PATCH 1/2] feat: add stream module api --- src/lib.rs | 7 ++ src/stream/conf.rs | 267 ++++++++++++++++++++++++++++++++++++++++++ src/stream/mod.rs | 7 ++ src/stream/module.rs | 104 ++++++++++++++++ src/stream/session.rs | 208 ++++++++++++++++++++++++++++++++ 5 files changed, 593 insertions(+) create mode 100644 src/stream/conf.rs create mode 100644 src/stream/mod.rs create mode 100644 src/stream/module.rs create mode 100644 src/stream/session.rs diff --git a/src/lib.rs b/src/lib.rs index 171c3676..e7ec451d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -145,6 +145,13 @@ pub mod ffi; #[cfg(ngx_feature = "http")] pub mod http; +/// The stream module. +/// +/// This modules provides wrappers and utilities to NGINX stream APIs, such as session, +/// configuration access. +#[cfg(ngx_feature = "stream")] +pub mod stream; + /// The log module. /// /// This module provides an interface into the NGINX logger framework. diff --git a/src/stream/conf.rs b/src/stream/conf.rs new file mode 100644 index 00000000..a4cad71c --- /dev/null +++ b/src/stream/conf.rs @@ -0,0 +1,267 @@ +use ::core::ptr::NonNull; + +use crate::{ + ffi::{ + ngx_conf_t, ngx_cycle_t, ngx_module_t, ngx_stream_conf_ctx_t, ngx_stream_core_srv_conf_t, + ngx_stream_session_t, ngx_stream_upstream_srv_conf_t, + }, + stream::StreamModule, +}; + +/// Utility trait for types containing Stream module main configuration +pub trait StreamModuleMainConfExt { + /// Get a non-null reference to the main configuration structure for Stream module + /// + /// # Safety + /// Caller must ensure that type `T` matches the configuration type for the specified module. + #[inline] + unsafe fn stream_main_conf_unchecked(&self, _module: &ngx_module_t) -> Option> { + None + } +} + +/// Utility trait for types containing Stream module server configuration +pub trait StreamModuleServerConfExt { + /// Get a non-null reference to the server configuration structure for Stream module + /// + /// # Safety + /// Caller must ensure that type `T` matches the configuration type for the specified module. + #[inline] + unsafe fn stream_server_conf_unchecked(&self, _module: &ngx_module_t) -> Option> { + None + } +} + +impl StreamModuleMainConfExt for ngx_stream_conf_ctx_t { + #[inline] + unsafe fn stream_main_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + NonNull::new(unsafe { *self.main_conf.add(module.ctx_index) }.cast()) + } +} +impl StreamModuleServerConfExt for ngx_stream_conf_ctx_t { + #[inline] + unsafe fn stream_server_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + NonNull::new(unsafe { *self.srv_conf.add(module.ctx_index) }.cast()) + } +} + +impl StreamModuleMainConfExt for ngx_cycle_t { + #[inline] + unsafe fn stream_main_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + let stream_conf = + unsafe { self.conf_ctx.add(nginx_sys::ngx_stream_module.index).as_ref()? }; + let conf_ctx = (*stream_conf).cast::(); + unsafe { conf_ctx.as_ref()?.stream_main_conf_unchecked(module) } + } +} + +impl StreamModuleMainConfExt for ngx_conf_t { + #[inline] + unsafe fn stream_main_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + let conf_ctx = self.ctx.cast::(); + unsafe { conf_ctx.as_ref()?.stream_main_conf_unchecked(module) } + } +} +impl StreamModuleServerConfExt for ngx_conf_t { + #[inline] + unsafe fn stream_server_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + let conf_ctx = self.ctx.cast::(); + unsafe { + let conf_ctx = conf_ctx.as_ref()?; + NonNull::new((*conf_ctx.srv_conf.add(module.ctx_index)).cast()) + } + } +} + +impl StreamModuleMainConfExt for ngx_stream_core_srv_conf_t { + #[inline] + unsafe fn stream_main_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + unsafe { self.ctx.as_ref()?.stream_main_conf_unchecked(module) } + } +} +impl StreamModuleServerConfExt for ngx_stream_core_srv_conf_t { + #[inline] + unsafe fn stream_server_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + unsafe { self.ctx.as_ref()?.stream_server_conf_unchecked(module) } + } +} + +impl StreamModuleMainConfExt for ngx_stream_session_t { + #[inline] + unsafe fn stream_main_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + NonNull::new(unsafe { *self.main_conf.add(module.ctx_index) }.cast()) + } +} +impl StreamModuleServerConfExt for ngx_stream_session_t { + #[inline] + unsafe fn stream_server_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + NonNull::new(unsafe { *self.srv_conf.add(module.ctx_index) }.cast()) + } +} + +impl StreamModuleServerConfExt for ngx_stream_upstream_srv_conf_t { + #[inline] + unsafe fn stream_server_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + let conf = self.srv_conf; + if conf.is_null() { + return None; + } + NonNull::new(unsafe { *conf.add(module.ctx_index) }.cast()) + } +} + +/// Trait to define and access main module configuration +/// +/// # Safety +/// Caller must ensure that type `StreamModuleMainConf::MainConf` matches the configuration type +/// for the specified module. +pub unsafe trait StreamModuleMainConf: StreamModule { + /// Type for main module configuration + type MainConf; + /// Get reference to main module configuration + fn main_conf(o: &impl StreamModuleMainConfExt) -> Option<&'static Self::MainConf> { + unsafe { Some(o.stream_main_conf_unchecked(Self::module())?.as_ref()) } + } + /// Get mutable reference to main module configuration + fn main_conf_mut(o: &impl StreamModuleMainConfExt) -> Option<&'static mut Self::MainConf> { + unsafe { Some(o.stream_main_conf_unchecked(Self::module())?.as_mut()) } + } +} + +/// Trait to define and access server-specific module configuration +/// +/// # Safety +/// Caller must ensure that type `StreamModuleServerConf::ServerConf` matches the configuration type +/// for the specified module. +pub unsafe trait StreamModuleServerConf: StreamModule { + /// Type for server-specific module configuration + type ServerConf; + /// Get reference to server-specific module configuration + fn server_conf(o: &impl StreamModuleServerConfExt) -> Option<&'static Self::ServerConf> { + unsafe { Some(o.stream_server_conf_unchecked(Self::module())?.as_ref()) } + } + /// Get mutable reference to server-specific module configuration + fn server_conf_mut( + o: &impl StreamModuleServerConfExt, + ) -> Option<&'static mut Self::ServerConf> { + unsafe { Some(o.stream_server_conf_unchecked(Self::module())?.as_mut()) } + } +} + +mod core { + use crate::stream::{ + StreamModule, StreamModuleMainConf, StreamModuleServerConf, StreamSessionHandler, + }; + use crate::{ + allocator::AllocError, + ffi::{ngx_stream_core_main_conf_t, ngx_stream_core_module, ngx_stream_core_srv_conf_t}, + ngx_conf_log_error, + }; + + /// Auxiliary structure to access `ngx_stream_core_module` configuration. + pub struct NgxStreamCoreModule; + + impl StreamModule for NgxStreamCoreModule { + fn module() -> &'static crate::ffi::ngx_module_t { + unsafe { &*::core::ptr::addr_of!(ngx_stream_core_module) } + } + } + unsafe impl StreamModuleMainConf for NgxStreamCoreModule { + type MainConf = ngx_stream_core_main_conf_t; + } + unsafe impl StreamModuleServerConf for NgxStreamCoreModule { + type ServerConf = ngx_stream_core_srv_conf_t; + } + + /// Stream phases in which a module can register handlers. + #[repr(usize)] + pub enum StreamPhase { + /// Post-accept phase + PostAccept = crate::ffi::ngx_stream_phases_NGX_STREAM_POST_ACCEPT_PHASE as _, + /// Pre-access phase + Preaccess = crate::ffi::ngx_stream_phases_NGX_STREAM_PREACCESS_PHASE as _, + /// Access phase + Access = crate::ffi::ngx_stream_phases_NGX_STREAM_ACCESS_PHASE as _, + /// Ssl phase + Ssl = crate::ffi::ngx_stream_phases_NGX_STREAM_SSL_PHASE as _, + /// Pre-read phase + Preread = crate::ffi::ngx_stream_phases_NGX_STREAM_PREREAD_PHASE as _, + /// Content phase + Content = crate::ffi::ngx_stream_phases_NGX_STREAM_CONTENT_PHASE as _, + /// Log phase + Log = crate::ffi::ngx_stream_phases_NGX_STREAM_LOG_PHASE as _, + } + + /// Register a request handler for a specified phase. + /// This function must be called from the module's `postconfiguration()` function. + pub fn add_phase_handler(cf: &mut nginx_sys::ngx_conf_t) -> Result<(), AllocError> + where + S: StreamSessionHandler, + { + let cmcf = NgxStreamCoreModule::main_conf_mut(cf).expect("stream core main conf"); + let s: *mut nginx_sys::ngx_stream_handler_pt = unsafe { + nginx_sys::ngx_array_push(&raw mut cmcf.phases[S::PHASE as usize].handlers).cast() + }; + if s.is_null() { + ngx_conf_log_error!( + nginx_sys::NGX_LOG_EMERG, + cf, + "failed to register {} handler", + S::name(), + ); + return Err(AllocError); + } + // set an H::PHASE phase handler + unsafe { + *s = Some(crate::stream::raw_handler::); + } + Ok(()) + } +} + +pub use core::{NgxStreamCoreModule, StreamPhase, add_phase_handler}; + +#[cfg(ngx_feature = "stream_ssl")] +mod ssl { + use crate::ffi::{ngx_stream_ssl_module, ngx_stream_ssl_srv_conf_t}; + + use crate::stream::{StreamModule, StreamModuleServerConf}; + + /// Auxiliary structure to access `ngx_stream_ssl_module` configuration. + pub struct NgxStreamSslModule; + + impl StreamModule for NgxStreamSslModule { + fn module() -> &'static crate::ffi::ngx_module_t { + unsafe { &*::core::ptr::addr_of!(ngx_stream_ssl_module) } + } + } + unsafe impl StreamModuleServerConf for NgxStreamSslModule { + type ServerConf = ngx_stream_ssl_srv_conf_t; + } +} +#[cfg(ngx_feature = "stream_ssl")] +pub use ssl::NgxStreamSslModule; + +mod upstream { + use super::{StreamModule, StreamModuleMainConf, StreamModuleServerConf}; + use crate::ffi::{ + ngx_stream_upstream_main_conf_t, ngx_stream_upstream_module, ngx_stream_upstream_srv_conf_t, + }; + + /// Auxiliary structure to access `ngx_stream_upstream_module` configuration. + pub struct NgxStreamUpstreamModule; + + impl StreamModule for NgxStreamUpstreamModule { + fn module() -> &'static crate::ffi::ngx_module_t { + unsafe { &*::core::ptr::addr_of!(ngx_stream_upstream_module) } + } + } + unsafe impl StreamModuleMainConf for NgxStreamUpstreamModule { + type MainConf = ngx_stream_upstream_main_conf_t; + } + unsafe impl StreamModuleServerConf for NgxStreamUpstreamModule { + type ServerConf = ngx_stream_upstream_srv_conf_t; + } +} + +pub use upstream::NgxStreamUpstreamModule; diff --git a/src/stream/mod.rs b/src/stream/mod.rs new file mode 100644 index 00000000..22ce7c04 --- /dev/null +++ b/src/stream/mod.rs @@ -0,0 +1,7 @@ +mod conf; +mod module; +mod session; + +pub use conf::*; +pub use module::*; +pub use session::*; diff --git a/src/stream/module.rs b/src/stream/module.rs new file mode 100644 index 00000000..5128b4fe --- /dev/null +++ b/src/stream/module.rs @@ -0,0 +1,104 @@ +use ::core::{ + ffi::{c_char, c_void}, + ptr, +}; + +use nginx_sys::{ngx_conf_t, ngx_int_t, ngx_module_t}; + +use crate::{ + core::{NGX_CONF_ERROR, Pool, Status}, + http::Merge, + stream::{StreamModuleMainConf, StreamModuleServerConf}, +}; + +/// The `StreamModule` trait provides the NGINX configuration stage interface. +/// +/// These functions allocate structures, initialize them, and merge through the configuration +/// layers. +/// +/// See for details. +pub trait StreamModule { + /// Returns reference to a global variable of type [ngx_module_t] created for this module. + fn module() -> &'static ngx_module_t; + + /// # Safety + /// + /// Callers should provide valid non-null `ngx_conf_t` arguments. Implementers must + /// guard against null inputs or risk runtime errors. + unsafe extern "C" fn preconfiguration(_cf: *mut ngx_conf_t) -> ngx_int_t { + Status::NGX_OK.into() + } + + /// # Safety + /// + /// Callers should provide valid non-null `ngx_conf_t` arguments. Implementers must + /// guard against null inputs or risk runtime errors. + unsafe extern "C" fn postconfiguration(_cf: *mut ngx_conf_t) -> ngx_int_t { + Status::NGX_OK.into() + } + + /// # Safety + /// + /// Callers should provide valid non-null `ngx_conf_t` arguments. Implementers must + /// guard against null inputs or risk runtime errors. + unsafe extern "C" fn create_main_conf(cf: *mut ngx_conf_t) -> *mut c_void + where + Self: StreamModuleMainConf, + Self::MainConf: Default, + { + unsafe { + let pool = Pool::from_ngx_pool((*cf).pool); + pool.allocate::(Default::default()) as *mut c_void + } + } + + /// # Safety + /// + /// Callers should provide valid non-null `ngx_conf_t` arguments. Implementers must + /// guard against null inputs or risk runtime errors. + unsafe extern "C" fn init_main_conf(_cf: *mut ngx_conf_t, _conf: *mut c_void) -> *mut c_char + where + Self: StreamModuleMainConf, + Self::MainConf: Default, + { + ptr::null_mut() + } + + /// # Safety + /// + /// Callers should provide valid non-null `ngx_conf_t` arguments. Implementers must + /// guard against null inputs or risk runtime errors. + unsafe extern "C" fn create_srv_conf(cf: *mut ngx_conf_t) -> *mut c_void + where + Self: StreamModuleServerConf, + Self::ServerConf: Default, + { + unsafe { + let pool = Pool::from_ngx_pool((*cf).pool); + pool.allocate::(Default::default()) as *mut c_void + } + } + + /// # Safety + /// + /// Callers should provide valid non-null `ngx_conf_t` arguments. Implementers must + /// guard against null inputs or risk runtime errors. + unsafe extern "C" fn merge_srv_conf( + _cf: *mut ngx_conf_t, + prev: *mut c_void, + conf: *mut c_void, + ) -> *mut c_char + where + Self: StreamModuleServerConf, + Self::ServerConf: Merge, + { + unsafe { + let prev = &mut *(prev as *mut Self::ServerConf); + let conf = &mut *(conf as *mut Self::ServerConf); + match conf.merge(prev) { + Ok(_) => ptr::null_mut(), + Err(_) => NGX_CONF_ERROR as _, + } + } + } +} diff --git a/src/stream/session.rs b/src/stream/session.rs new file mode 100644 index 00000000..76efacaa --- /dev/null +++ b/src/stream/session.rs @@ -0,0 +1,208 @@ +use core::ffi::c_void; +use core::fmt; +use core::ptr::NonNull; + +use nginx_sys::{ + NGX_ERROR, NGX_OK, ngx_connection_t, ngx_int_t, ngx_log_t, ngx_module_t, ngx_str_t, + ngx_stream_complex_value, ngx_stream_complex_value_t, ngx_stream_session_t, + ngx_stream_upstream_t, +}; + +use crate::{ + core::{NgxStr, Status}, + stream::{StreamModuleMainConfExt, StreamModuleServerConfExt, StreamPhase}, +}; + +/// Trait for converting handler return types into `ngx_int_t`. +/// Any desired error handling / logging logic can be implemented +/// in the `into_handler_status` method. +/// +/// There are predefined implementations for `ngx_int_t`, [`Status`], +/// [`Option`] with value type implementing [`IntoHandlerStatus`]. +pub trait IntoHandlerStatus +where + Self: Sized, +{ + /// Convert the handler return type into an `ngx_int_t`. + fn into_handler_status(self, _r: &Session) -> ngx_int_t; +} + +impl IntoHandlerStatus for Option +where + T: IntoHandlerStatus, +{ + #[inline] + fn into_handler_status(self, r: &Session) -> ngx_int_t { + self.map(|val| val.into_handler_status(r)).unwrap_or(NGX_ERROR as _) + } +} + +impl IntoHandlerStatus for ngx_int_t { + #[inline] + fn into_handler_status(self, _r: &Session) -> ngx_int_t { + self + } +} + +impl IntoHandlerStatus for Status { + #[inline] + fn into_handler_status(self, _r: &Session) -> ngx_int_t { + self.0 + } +} + +/// Trait for static request handler. +pub trait StreamSessionHandler { + /// The phase in which the handler is invoked. + const PHASE: StreamPhase; + /// The return type of the handler. + type Output: IntoHandlerStatus; + /// The handler function. + fn handler(session: &mut Session) -> Self::Output; + /// Handler name for logging purposes. + /// [`core::any::type_name`] is used by default. + fn name() -> &'static str { + core::any::type_name::() + } +} + +/// The C-compatible handler wrapper function. +/// +/// # Safety +/// +/// The caller has provided a valid non-null pointer to an [`ngx_stream_session_t`]. +pub(crate) unsafe extern "C" fn raw_handler(s: *mut ngx_stream_session_t) -> ngx_int_t +where + S: StreamSessionHandler, +{ + let s = unsafe { Session::from_ngx_stream_session(s) }; + S::handler(s).into_handler_status(s) +} + +/// Wrapper struct for an [`ngx_stream_session_t`] pointer, providing methods for working with Stream +/// session. +#[repr(transparent)] +pub struct Session(ngx_stream_session_t); + +impl<'a> From<&'a Session> for *const ngx_stream_session_t { + fn from(session: &'a Session) -> Self { + &raw const session.0 + } +} + +impl<'a> From<&'a mut Session> for *mut ngx_stream_session_t { + fn from(session: &'a mut Session) -> Self { + &raw mut session.0 + } +} + +impl AsRef for Session { + fn as_ref(&self) -> &ngx_stream_session_t { + &self.0 + } +} + +impl AsMut for Session { + fn as_mut(&mut self) -> &mut ngx_stream_session_t { + &mut self.0 + } +} + +impl Session { + /// Create a [`Session`] from an [`ngx_stream_session_t`]. + /// + /// # Safety + /// + /// The caller has provided a valid non-null pointer to a valid `ngx_stream_session_t` + /// which shares the same representation as `Request`. + pub unsafe fn from_ngx_stream_session<'a>(r: *mut ngx_stream_session_t) -> &'a mut Session { + unsafe { &mut *r.cast::() } + } + + /// Returns the result as an `Option` if it exists, otherwise `None`. + /// + /// The option wraps an ngx_stream_upstream_t instance, it will be none when the underlying NGINX + /// request does not have a pointer to a [`ngx_stream_upstream_t`] upstream structure. + pub fn upstream(&self) -> Option<*mut ngx_stream_upstream_t> { + if self.0.upstream.is_null() { + return None; + } + Some(self.0.upstream) + } + + /// Pointer to a [`ngx_connection_t`] client connection object. + /// + /// [`ngx_connection_t`]: https://nginx.org/en/docs/dev/development_guide.html#connection + pub fn connection(&self) -> *mut ngx_connection_t { + self.0.connection + } + + /// Pointer to a [`ngx_log_t`]. + /// + /// [`ngx_log_t`]: https://nginx.org/en/docs/dev/development_guide.html#logging + pub fn log(&self) -> *mut ngx_log_t { + unsafe { (*self.connection()).log } + } + + /// Get Module context pointer + fn get_module_ctx_ptr(&self, module: &ngx_module_t) -> *mut c_void { + unsafe { *self.0.ctx.add(module.ctx_index) } + } + + /// Get Module context + pub fn get_module_ctx(&self, module: &ngx_module_t) -> Option<&T> { + let ctx = self.get_module_ctx_ptr(module).cast::(); + // SAFETY: ctx is either NULL or allocated with ngx_p(c)alloc and + // explicitly initialized by the module + unsafe { ctx.as_ref() } + } + + /// Sets the value as the module's context. + pub fn set_module_ctx(&self, value: *mut c_void, module: &ngx_module_t) { + unsafe { + *self.0.ctx.add(module.ctx_index) = value; + }; + } + + /// Get the value of a [complex value]. + pub fn get_complex_value(&mut self, cv: &mut ngx_stream_complex_value_t) -> Option<&NgxStr> { + let r = (self as *mut Session).cast(); + let val = cv as *mut ngx_stream_complex_value_t; + // SAFETY: `ngx_stream_complex_value` does not mutate `r` or `val` and guarentees that + // a valid Nginx string is stored in `value` if it successfully returns. + unsafe { + let mut value = ngx_str_t::default(); + if ngx_stream_complex_value(r, val, &raw mut value) != NGX_OK as ngx_int_t { + return None; + } + Some(NgxStr::from_ngx_str(value)) + } + } +} + +impl StreamModuleMainConfExt for Session { + #[inline] + unsafe fn stream_main_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + unsafe { + // SAFETY: main_conf[module.ctx_index] is either NULL or allocated with ngx_p(c)alloc + // and explicitly initialized by the module + NonNull::new((*self.0.main_conf.add(module.ctx_index)).cast()) + } + } +} +impl StreamModuleServerConfExt for Session { + #[inline] + unsafe fn stream_server_conf_unchecked(&self, module: &ngx_module_t) -> Option> { + unsafe { + // SAFETY: srv_conf[module.ctx_index] is either NULL or allocated with ngx_p(c)alloc and + // explicitly initialized by the module + NonNull::new((*self.0.srv_conf.add(module.ctx_index)).cast()) + } + } +} + +impl fmt::Debug for Session { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Session").field("session_", &self.0).finish() + } +} From a5eb491250a1a16ab8c8bde7be754ad992a465bf Mon Sep 17 00:00:00 2001 From: plusls Date: Mon, 11 May 2026 11:15:37 +0800 Subject: [PATCH 2/2] fix: gate stream module behind stream cargo feature Stream API was compiled whenever NGINX reported stream support, but nginx-sys only generated stream FFI when its own stream feature was enabled. Add a stream feature to the root crate that propagates to nginx-sys/stream and gate the module with both conditions. --- Cargo.toml | 4 ++++ src/lib.rs | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index df0d190c..c8aa9117 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,10 @@ std = [ "alloc", "allocator-api2/std" ] +# Enables the HTTP module API. +http = ["nginx-sys/http"] +# Enables the Stream module API. +stream = ["nginx-sys/stream"] # Enables the build scripts to build a copy of nginx source and link against it. vendored = ["nginx-sys/vendored"] diff --git a/src/lib.rs b/src/lib.rs index e7ec451d..076f136c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -149,7 +149,7 @@ pub mod http; /// /// This modules provides wrappers and utilities to NGINX stream APIs, such as session, /// configuration access. -#[cfg(ngx_feature = "stream")] +#[cfg(all(feature = "stream", ngx_feature = "stream"))] pub mod stream; /// The log module.