From a54a6398e35d3a687a715bf30074b39ba93387ce Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 16:23:57 -0700 Subject: [PATCH] add non-spawning server task APIs --- integration/tests/integration_test.rs | 14 +- rodbus/src/server/mod.rs | 259 +++++++++++++++++++++----- 2 files changed, 222 insertions(+), 51 deletions(-) diff --git a/integration/tests/integration_test.rs b/integration/tests/integration_test.rs index 5ee6af1e..dc0cfeba 100644 --- a/integration/tests/integration_test.rs +++ b/integration/tests/integration_test.rs @@ -115,17 +115,19 @@ impl RequestHandler for Handler { async fn test_requests_and_responses() { let handler = Handler::new().wrap(); - let addr = SocketAddr::from_str("127.0.0.1:40000").unwrap(); + let listener = tokio::net::TcpListener::bind(SocketAddr::from_str("127.0.0.1:0").unwrap()) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); - let _server = spawn_tcp_server_task( + let (_server, task) = create_tcp_server_task( 1, - addr, + listener, ServerHandlerMap::single(UnitId::new(1), handler.clone()), AddressFilter::Any, DecodeLevel::default(), - ) - .await - .unwrap(); + ); + tokio::spawn(task.run()); let (tx, mut rx) = tokio::sync::mpsc::channel(8); let listener = ClientStateListener { tx }; diff --git a/rodbus/src/server/mod.rs b/rodbus/src/server/mod.rs index caa4c587..9d7854cb 100644 --- a/rodbus/src/server/mod.rs +++ b/rodbus/src/server/mod.rs @@ -4,7 +4,7 @@ use tracing::Instrument; use crate::decode::DecodeLevel; use crate::server::task::ServerSetting; -use crate::tcp::server::{ServerTask, TcpServerConnectionHandler}; +use crate::tcp::server::{ServerTask as TcpServerTask, TcpServerConnectionHandler}; /// server handling mod address_filter; @@ -35,6 +35,53 @@ pub struct ServerHandle { tx: tokio::sync::mpsc::Sender, } +/// A server task that has been created but not yet spawned. +/// +/// This is returned, alongside its [`ServerHandle`], by the `create_*_server_task` functions. +/// Drive it to completion by awaiting [`ServerTask::run`], typically from within +/// [`tokio::spawn`]. The task completes when the associated [`ServerHandle`] is dropped. +/// +/// Unlike the `spawn_*_server_task` functions, no tracing span is attached to the task, so the +/// caller is free to wrap [`run`](ServerTask::run) with their own instrumentation. +pub struct ServerTask { + inner: ServerTaskInner, +} + +enum ServerTaskInner { + Tcp( + Box>, + tokio::sync::mpsc::Receiver, + ), + #[cfg(feature = "serial")] + Rtu(Box>), +} + +impl ServerTask { + fn tcp(task: TcpServerTask, commands: tokio::sync::mpsc::Receiver) -> Self { + Self { + inner: ServerTaskInner::Tcp(Box::new(task), commands), + } + } + + #[cfg(feature = "serial")] + fn rtu(task: crate::serial::server::RtuServerTask) -> Self { + Self { + inner: ServerTaskInner::Rtu(Box::new(task)), + } + } + + /// Run the server task until the associated [`ServerHandle`] is dropped. + pub async fn run(self) { + match self.inner { + ServerTaskInner::Tcp(mut task, commands) => task.run(commands).await, + #[cfg(feature = "serial")] + ServerTaskInner::Rtu(mut task) => { + task.run().await; + } + } + } +} + impl ServerHandle { /// Construct a [ServerHandle] from its fields /// @@ -70,26 +117,44 @@ pub async fn spawn_tcp_server_task( decode: DecodeLevel, ) -> Result { let listener = tokio::net::TcpListener::bind(addr).await?; + let (handle, task) = create_tcp_server_task(max_sessions, listener, handlers, filter, decode); - let (tx, rx) = tokio::sync::mpsc::channel(SERVER_SETTING_CHANNEL_CAPACITY); + tokio::spawn( + task.run() + .instrument(tracing::info_span!("Modbus-Server-TCP", "listen" = ?addr)), + ); - let task = async move { - ServerTask::new( - max_sessions, - listener, - handlers, - TcpServerConnectionHandler::Tcp, - filter, - decode, - ) - .run(rx) - .instrument(tracing::info_span!("Modbus-Server-TCP", "listen" = ?addr)) - .await; - }; + Ok(handle) +} - tokio::spawn(task); +/// Creates a TCP server task for a pre-bound listener, but does **not** spawn it. +/// +/// It is the caller's responsibility to run the returned [`ServerTask`] on a runtime, e.g. +/// `tokio::spawn(task.run())`. Each incoming connection will spawn a new task to handle it. +/// +/// * `max_sessions` - Maximum number of concurrent sessions +/// * `listener` - A pre-bound TCP listener +/// * `handlers` - A map of handlers keyed by a unit id +/// * `filter` - Address filter which may be used to restrict the connecting IP address +/// * `decode` - Decode log level +pub fn create_tcp_server_task( + max_sessions: usize, + listener: tokio::net::TcpListener, + handlers: ServerHandlerMap, + filter: AddressFilter, + decode: DecodeLevel, +) -> (ServerHandle, ServerTask) { + let (tx, rx) = tokio::sync::mpsc::channel(SERVER_SETTING_CHANNEL_CAPACITY); + let task = TcpServerTask::new( + max_sessions, + listener, + handlers, + TcpServerConnectionHandler::Tcp, + filter, + decode, + ); - Ok(ServerHandle::new(tx)) + (ServerHandle::new(tx), ServerTask::tcp(task, rx)) } /// Spawns a RTU server task onto the runtime. @@ -109,6 +174,35 @@ pub fn spawn_rtu_server_task( handlers: ServerHandlerMap, decode: DecodeLevel, ) -> Result { + let (handle, task) = create_rtu_server_task(path, settings, retry, handlers, decode); + let path = path.to_string(); + + tokio::spawn( + task.run() + .instrument(tracing::info_span!("Modbus-Server-RTU", "port" = ?path)), + ); + + Ok(handle) +} + +/// Creates an RTU server task, but does **not** spawn it or open the serial port. +/// +/// It is the caller's responsibility to run the returned [`ServerTask`] on a runtime, e.g. +/// `tokio::spawn(task.run())`. +/// +/// * `path` - Path to the serial device. Generally `/dev/tty0` on Linux and `COM1` on Windows. +/// * `settings` - Serial port settings +/// * `retry` - A boxed trait object that controls when opening the serial port is retried after a failure +/// * `handlers` - A map of handlers keyed by a unit id +/// * `decode` - Decode log level +#[cfg(feature = "serial")] +pub fn create_rtu_server_task( + path: &str, + settings: crate::serial::SerialSettings, + retry: Box, + handlers: ServerHandlerMap, + decode: DecodeLevel, +) -> (ServerHandle, ServerTask) { let (tx, rx) = tokio::sync::mpsc::channel(SERVER_SETTING_CHANNEL_CAPACITY); let session = crate::server::task::SessionTask::new( handlers, @@ -119,24 +213,14 @@ pub fn spawn_rtu_server_task( decode, ); - let mut rtu = crate::serial::server::RtuServerTask { + let rtu = crate::serial::server::RtuServerTask { port: path.to_string(), retry, settings, session, }; - let path = path.to_string(); - - let task = async move { - rtu.run() - .instrument(tracing::info_span!("Modbus-Server-RTU", "port" = ?path)) - .await - }; - - tokio::spawn(task); - - Ok(ServerHandle::new(tx)) + (ServerHandle::new(tx), ServerTask::rtu(rtu)) } /// Spawns a "raw" TLS server task onto the runtime. This TLS server does NOT require that @@ -174,6 +258,38 @@ pub async fn spawn_tls_server_task( .await } +/// Creates a "raw" TLS server task for a pre-bound listener, but does **not** spawn it. +/// +/// This TLS server does not require that the client certificate contain the Role extension and +/// allows all operations for any authenticated client. It is the caller's responsibility to run +/// the returned [`ServerTask`] on a runtime, e.g. `tokio::spawn(task.run())`. +/// +/// * `max_sessions` - Maximum number of concurrent sessions +/// * `listener` - A pre-bound TCP listener +/// * `handlers` - A map of handlers keyed by a unit id +/// * `tls_config` - TLS configuration +/// * `filter` - Address filter which may be used to restrict the connecting IP address +/// * `decode` - Decode log level +#[cfg(feature = "enable-tls")] +pub fn create_tls_server_task( + max_sessions: usize, + listener: tokio::net::TcpListener, + handlers: ServerHandlerMap, + tls_config: TlsServerConfig, + filter: AddressFilter, + decode: DecodeLevel, +) -> (ServerHandle, ServerTask) { + create_tls_server_task_impl( + max_sessions, + listener, + handlers, + None, + tls_config, + filter, + decode, + ) +} + /// Spawns a "Secure Modbus" TLS server task onto the runtime. This TLS server requires that /// the client certificate contain the Role extension and checks the authorization of requests against /// the supplied handler. @@ -212,6 +328,40 @@ pub async fn spawn_tls_server_task_with_authz( .await } +/// Creates a "Secure Modbus" TLS server task for a pre-bound listener, but does **not** spawn it. +/// +/// This TLS server requires that the client certificate contain the Role extension and checks the +/// authorization of requests against the supplied handler. It is the caller's responsibility to +/// run the returned [`ServerTask`] on a runtime, e.g. `tokio::spawn(task.run())`. +/// +/// * `max_sessions` - Maximum number of concurrent sessions +/// * `listener` - A pre-bound TCP listener +/// * `handlers` - A map of handlers keyed by a unit id +/// * `auth_handler` - Handler used to authorize requests +/// * `tls_config` - TLS configuration +/// * `filter` - Address filter which may be used to restrict the connecting IP address +/// * `decode` - Decode log level +#[cfg(feature = "enable-tls")] +pub fn create_tls_server_task_with_authz( + max_sessions: usize, + listener: tokio::net::TcpListener, + handlers: ServerHandlerMap, + auth_handler: std::sync::Arc, + tls_config: TlsServerConfig, + filter: AddressFilter, + decode: DecodeLevel, +) -> (ServerHandle, ServerTask) { + create_tls_server_task_impl( + max_sessions, + listener, + handlers, + Some(auth_handler), + tls_config, + filter, + decode, + ) +} + #[cfg(feature = "enable-tls")] async fn spawn_tls_server_task_impl( max_sessions: usize, @@ -223,24 +373,43 @@ async fn spawn_tls_server_task_impl( decode: DecodeLevel, ) -> Result { let listener = tokio::net::TcpListener::bind(addr).await?; + let (handle, task) = create_tls_server_task_impl( + max_sessions, + listener, + handlers, + auth_handler, + tls_config, + filter, + decode, + ); - let (tx, rx) = tokio::sync::mpsc::channel(SERVER_SETTING_CHANNEL_CAPACITY); + tokio::spawn( + task.run() + .instrument(tracing::info_span!("Modbus-Server-TLS", "listen" = ?addr)), + ); - let task = async move { - ServerTask::new( - max_sessions, - listener, - handlers, - TcpServerConnectionHandler::Tls(tls_config, auth_handler), - filter, - decode, - ) - .run(rx) - .instrument(tracing::info_span!("Modbus-Server-TLS", "listen" = ?addr)) - .await - }; + Ok(handle) +} - tokio::spawn(task); +#[cfg(feature = "enable-tls")] +fn create_tls_server_task_impl( + max_sessions: usize, + listener: tokio::net::TcpListener, + handlers: ServerHandlerMap, + auth_handler: Option>, + tls_config: TlsServerConfig, + filter: AddressFilter, + decode: DecodeLevel, +) -> (ServerHandle, ServerTask) { + let (tx, rx) = tokio::sync::mpsc::channel(SERVER_SETTING_CHANNEL_CAPACITY); + let task = TcpServerTask::new( + max_sessions, + listener, + handlers, + TcpServerConnectionHandler::Tls(tls_config, auth_handler), + filter, + decode, + ); - Ok(ServerHandle::new(tx)) + (ServerHandle::new(tx), ServerTask::tcp(task, rx)) }