Add non-spawning create variants for TCP/TLS/serial clients

rodbus/src/client/channel.rs

@@ -15,6 +15,52 @@ pub struct Channel {
     pub(crate) tx: tokio::sync::mpsc::Sender<Command>,
 }
 
+/// 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<Box<dyn crate::client::Listener<crate::client::PortState>>>,
     ) -> 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<dyn crate::retry::RetryStrategy>,
         decode: DecodeLevel,
         listener: Option<Box<dyn crate::client::Listener<crate::client::PortState>>>,
-    ) -> (Self, impl std::future::Future<Output = ()>) {
-        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

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<dyn RetryStrategy>,
+    listener: Option<Box<dyn Listener<ClientState>>>,
+    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<dyn RetryStrategy>,
+    decode: DecodeLevel,
+    listener: Option<Box<dyn Listener<PortState>>>,
+) -> (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<dyn RetryStrategy>,
+    tls_config: TlsClientConfig,
+    listener: Option<Box<dyn Listener<ClientState>>>,
+    client_options: ClientOptions,
+) -> (Channel, ClientTask) {
+    create_tls_channel(
+        host,
+        retry,
+        tls_config,
+        client_options,
+        listener.unwrap_or_else(|| NullListener::create()),
+    )
+}

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<dyn Listener<ClientState>>,
     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<dyn RetryStrategy>,
     listener: Box<dyn Listener<ClientState>>,
     options: ClientOptions,
-) -> (Channel, impl std::future::Future<Output = ()>) {
+) -> (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 {

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<dyn Listener<ClientState>>,
 ) -> 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<dyn Listener<ClientState>>,
-) -> (Channel, impl std::future::Future<Output = ()>) {
+) -> (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 {