diff --git a/rodbus/src/client/channel.rs b/rodbus/src/client/channel.rs index a4d4eeca..b11eae47 100644 --- a/rodbus/src/client/channel.rs +++ b/rodbus/src/client/channel.rs @@ -15,6 +15,52 @@ pub struct Channel { pub(crate) tx: tokio::sync::mpsc::Sender, } +/// A client channel task that has been created but not yet spawned. +/// +/// This is returned, alongside its [`Channel`] handle, by the `create_*_client_task` functions. +/// Drive it to completion by awaiting [`ClientTask::run`], typically from within +/// [`tokio::spawn`]. The task completes when the associated [`Channel`] handle is dropped. +/// +/// Unlike the `spawn_*_client_task` functions, no tracing span is attached to the task, so the +/// caller is free to wrap [`run`](ClientTask::run) with their own instrumentation. +pub struct ClientTask { + inner: ClientTaskInner, +} + +enum ClientTaskInner { + Tcp(crate::tcp::client::TcpChannelTask), + #[cfg(feature = "serial")] + Serial(crate::serial::client::SerialChannelTask), +} + +impl ClientTask { + pub(crate) fn tcp(task: crate::tcp::client::TcpChannelTask) -> Self { + Self { + inner: ClientTaskInner::Tcp(task), + } + } + + #[cfg(feature = "serial")] + pub(crate) fn serial(task: crate::serial::client::SerialChannelTask) -> Self { + Self { + inner: ClientTaskInner::Serial(task), + } + } + + /// Run the channel task until the associated [`Channel`] handle is dropped. + pub async fn run(self) { + match self.inner { + ClientTaskInner::Tcp(mut task) => { + task.run().await; + } + #[cfg(feature = "serial")] + ClientTaskInner::Serial(mut task) => { + task.run().await; + } + } + } +} + /// Request parameters to dispatch the request to the proper device #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[cfg_attr( @@ -48,6 +94,8 @@ impl Channel { decode: DecodeLevel, listener: Option>>, ) -> Self { + use tracing::Instrument; + let (handle, task) = Self::create_rtu_handle_and_task( path, serial_settings, @@ -56,7 +104,10 @@ impl Channel { decode, listener, ); - tokio::spawn(task); + tokio::spawn( + task.run() + .instrument(tracing::info_span!("Modbus-Client-RTU", "port" = ?path)), + ); handle } @@ -68,25 +119,17 @@ impl Channel { retry: Box, decode: DecodeLevel, listener: Option>>, - ) -> (Self, impl std::future::Future) { - use tracing::Instrument; - - let path = path.to_string(); + ) -> (Self, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(max_queued_requests); - let task = async move { - let _ = crate::serial::client::SerialChannelTask::new( - &path, - serial_settings, - rx.into(), - retry, - decode, - listener.unwrap_or_else(|| crate::client::NullListener::create()), - ) - .run() - .instrument(tracing::info_span!("Modbus-Client-RTU", "port" = ?path)) - .await; - }; - (Channel { tx }, task) + let task = crate::serial::client::SerialChannelTask::new( + path, + serial_settings, + rx.into(), + retry, + decode, + listener.unwrap_or_else(|| crate::client::NullListener::create()), + ); + (Channel { tx }, ClientTask::serial(task)) } /// Enable communications diff --git a/rodbus/src/client/mod.rs b/rodbus/src/client/mod.rs index 08429634..f464140c 100644 --- a/rodbus/src/client/mod.rs +++ b/rodbus/src/client/mod.rs @@ -143,6 +143,34 @@ pub fn spawn_tcp_client_task_with_options( ) } +/// Creates a channel task that maintains a TCP connection and processes requests, but does +/// **not** spawn it. It is the caller's responsibility to run the returned [`ClientTask`] onto a +/// runtime, e.g. `tokio::spawn(task.run())`. The task completes when the returned [`Channel`] +/// handle is dropped. +/// +/// Unlike [`spawn_tcp_client_task_with_options`], no tracing span is attached to the task, so the +/// caller may wrap [`ClientTask::run`] with their own instrumentation before spawning it. +/// +/// The channel uses the provided [`RetryStrategy`] to pause between failed connection attempts +/// +/// * `host` - Address/port of the remote server. Can be an IP address or name on which to perform DNS resolution. +/// * `retry` - A boxed trait object that controls when the connection is retried on failure +/// * `listener` - Optional callback to monitor the TCP connection state +/// * `client_options` - A builder that contains various client options. +pub fn create_tcp_client_task_with_options( + host: HostAddr, + retry: Box, + listener: Option>>, + client_options: ClientOptions, +) -> (Channel, ClientTask) { + crate::tcp::client::create_tcp_channel( + host, + retry, + listener.unwrap_or_else(|| NullListener::create()), + client_options, + ) +} + /// Spawns a channel task onto the runtime that opens a serial port and processes /// requests. The task completes when the returned channel handle /// is dropped. @@ -177,6 +205,42 @@ pub fn spawn_rtu_client_task( ) } +/// Creates a channel task that opens a serial port and processes requests, but does **not** +/// spawn it. It is the caller's responsibility to run the returned [`ClientTask`] onto a runtime, +/// e.g. `tokio::spawn(task.run())`. The task completes when the returned [`Channel`] handle is +/// dropped. +/// +/// Unlike [`spawn_rtu_client_task`], no tracing span is attached to the task, so the caller may +/// wrap [`ClientTask::run`] with their own instrumentation before spawning it. +/// +/// The channel uses the provided [`RetryStrategy`] to pause between failed attempts to open the +/// serial port or after the serial port fails. +/// +/// * `path` - Path to the serial device. Generally `/dev/tty0` on Linux and `COM1` on Windows. +/// * `serial_settings` = Serial port settings +/// * `max_queued_requests` - The maximum size of the request queue +/// * `retry` - A boxed trait object that controls when opening the serial port is retried on failure +/// * `decode` - Decode log level +/// * `listener` - Optional callback to monitor the state of the serial port +#[cfg(feature = "serial")] +pub fn create_rtu_client_task( + path: &str, + serial_settings: crate::serial::SerialSettings, + max_queued_requests: usize, + retry: Box, + decode: DecodeLevel, + listener: Option>>, +) -> (Channel, ClientTask) { + Channel::create_rtu_handle_and_task( + path, + serial_settings, + max_queued_requests, + retry, + decode, + listener, + ) +} + /// Spawns a channel task onto the runtime that maintains a TLS connection and processes /// requests. The task completes when the returned channel handle /// is dropped. @@ -212,3 +276,35 @@ pub fn spawn_tls_client_task( listener.unwrap_or_else(|| NullListener::create()), ) } + +/// Creates a channel task that maintains a TLS connection and processes requests, but does +/// **not** spawn it. It is the caller's responsibility to run the returned [`ClientTask`] onto a +/// runtime, e.g. `tokio::spawn(task.run())`. The task completes when the returned [`Channel`] +/// handle is dropped. +/// +/// Unlike [`spawn_tls_client_task`], no tracing span is attached to the task, so the caller may +/// wrap [`ClientTask::run`] with their own instrumentation before spawning it. +/// +/// The channel uses the provided [`RetryStrategy`] to pause between failed connection attempts +/// +/// * `host` - Address/port of the remote server. Can be a IP address or name on which to perform DNS resolution. +/// * `retry` - A boxed trait object that controls when the connection is retried on failure +/// * `tls_config` - TLS configuration +/// * `listener` - Optional callback to monitor the TLS connection state +/// * `client_options` - A builder that contains various client options. +#[cfg(feature = "enable-tls")] +pub fn create_tls_client_task_with_options( + host: HostAddr, + retry: Box, + tls_config: TlsClientConfig, + listener: Option>>, + client_options: ClientOptions, +) -> (Channel, ClientTask) { + create_tls_channel( + host, + retry, + tls_config, + client_options, + listener.unwrap_or_else(|| NullListener::create()), + ) +} diff --git a/rodbus/src/tcp/client.rs b/rodbus/src/tcp/client.rs index 7729d951..d19afa77 100644 --- a/rodbus/src/tcp/client.rs +++ b/rodbus/src/tcp/client.rs @@ -1,6 +1,6 @@ use tracing::Instrument; -use crate::client::{Channel, ClientState, HostAddr, Listener}; +use crate::client::{Channel, ClientState, ClientTask, HostAddr, Listener}; use crate::common::phys::PhysLayer; use crate::client::message::Command; @@ -31,8 +31,9 @@ pub(crate) fn spawn_tcp_channel( listener: Box>, client_options: ClientOptions, ) -> Channel { + let span = tracing::info_span!("Modbus-Client-TCP", endpoint = ?host); let (handle, task) = create_tcp_channel(host, connect_retry, listener, client_options); - tokio::spawn(task); + tokio::spawn(task.run().instrument(span)); handle } @@ -41,22 +42,17 @@ pub(crate) fn create_tcp_channel( connect_retry: Box, listener: Box>, options: ClientOptions, -) -> (Channel, impl std::future::Future) { +) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); - let task = async move { - TcpChannelTask::new( - host.clone(), - rx.into(), - TcpTaskConnectionHandler::Tcp, - connect_retry, - options, - listener, - ) - .run() - .instrument(tracing::info_span!("Modbus-Client-TCP", endpoint = ?host)) - .await; - }; - (Channel { tx }, task) + let task = TcpChannelTask::new( + host, + rx.into(), + TcpTaskConnectionHandler::Tcp, + connect_retry, + options, + listener, + ); + (Channel { tx }, ClientTask::tcp(task)) } pub(crate) enum TcpTaskConnectionHandler { diff --git a/rodbus/src/tcp/tls/client.rs b/rodbus/src/tcp/tls/client.rs index 250f91fd..ba12fbc6 100644 --- a/rodbus/src/tcp/tls/client.rs +++ b/rodbus/src/tcp/tls/client.rs @@ -10,7 +10,7 @@ use tokio_rustls::rustls; use tokio_rustls::rustls::pki_types::InvalidDnsNameError; use tracing::Instrument; -use crate::client::{Channel, ClientState, HostAddr, Listener, RetryStrategy}; +use crate::client::{Channel, ClientState, ClientTask, HostAddr, Listener, RetryStrategy}; use crate::common::phys::PhysLayer; use crate::tcp::client::{TcpChannelTask, TcpTaskConnectionHandler}; use crate::tcp::tls::{CertificateMode, MinTlsVersion, TlsError}; @@ -30,8 +30,9 @@ pub(crate) fn spawn_tls_channel( options: ClientOptions, listener: Box>, ) -> Channel { + let span = tracing::info_span!("Modbus-Client-TLS", endpoint = ?host); let (handle, task) = create_tls_channel(host, connect_retry, tls_config, options, listener); - tokio::spawn(task); + tokio::spawn(task.run().instrument(span)); handle } @@ -41,22 +42,17 @@ pub(crate) fn create_tls_channel( tls_config: TlsClientConfig, options: ClientOptions, listener: Box>, -) -> (Channel, impl std::future::Future) { +) -> (Channel, ClientTask) { let (tx, rx) = tokio::sync::mpsc::channel(options.max_queued_requests); - let task = async move { - TcpChannelTask::new( - host.clone(), - rx.into(), - TcpTaskConnectionHandler::Tls(tls_config), - connect_retry, - options, - listener, - ) - .run() - .instrument(tracing::info_span!("Modbus-Client-TCP", endpoint = ?host)) - .await; - }; - (Channel { tx }, task) + let task = TcpChannelTask::new( + host, + rx.into(), + TcpTaskConnectionHandler::Tls(tls_config), + connect_retry, + options, + listener, + ); + (Channel { tx }, ClientTask::tcp(task)) } impl TlsClientConfig {