From 74a7343b68f3df88f004fdf31160605112b71625 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 29 Mar 2026 20:23:15 -0700 Subject: [PATCH 01/14] first pass at new builder API --- dnp3/src/master/builder.rs | 208 +++++++++++++++++++++++++++++++++++++ dnp3/src/master/mod.rs | 2 + dnp3/src/master/task.rs | 24 ++++- 3 files changed, 232 insertions(+), 2 deletions(-) create mode 100644 dnp3/src/master/builder.rs diff --git a/dnp3/src/master/builder.rs b/dnp3/src/master/builder.rs new file mode 100644 index 00000000..77dc2e50 --- /dev/null +++ b/dnp3/src/master/builder.rs @@ -0,0 +1,208 @@ +use std::time::Duration; + +use tracing::Instrument; + +use super::association::{Association, AssociationMap}; +use super::handler::{AssociationHandle, MasterChannel, MasterChannelConfig, MasterChannelType}; +use super::task::MasterTask; +use super::{ + AssociationConfig, AssociationError, AssociationHandler, AssociationInformation, ReadHandler, +}; +use crate::app::parse::options::ParseOptions; +use crate::app::Listener; +use crate::link::reader::LinkModes; +use crate::link::{EndpointAddress, LinkErrorMode}; +use crate::tcp::client::ClientTask; +use crate::tcp::{ClientConnectionHandler, ClientState, PostConnectionHandler}; +use crate::transport::FragmentAddr; +use crate::util::channel::Receiver; +use crate::util::phys::PhysAddr; +use crate::util::session::{Enabled, Session}; + +/// Builder for configuring a master task before binding it to a transport. +/// +/// Created via [`MasterBuilder::new`]. The returned [`MasterChannel`] can be +/// used for post-run operations, but async methods on it (and on any +/// [`AssociationHandle`] obtained from [`add_association`](Self::add_association)) +/// require the task to be running. +pub struct MasterBuilder { + channel: MasterChannel, + rx: Receiver, + config: MasterChannelConfig, + enabled: Enabled, + associations: AssociationMap, +} + +/// A master task bound to a TCP transport. Call [`run`](Self::run) to execute +/// the event loop. +#[must_use = "a MasterTcpTask does nothing unless you call .run()"] +pub struct MasterTcpTask { + inner: MasterTask, + connect_handler: Box, + listener: Box>, +} + +/// A master task bound to a serial transport. Call [`run`](Self::run) to execute +/// the event loop. +#[cfg(feature = "serial")] +#[must_use = "a MasterSerialTask does nothing unless you call .run()"] +pub struct MasterSerialTask { + inner: MasterTask, + path: String, + settings: crate::serial::SerialSettings, + retry_strategy: crate::app::RetryStrategy, + listener: Box>, +} + +impl MasterBuilder { + /// Create a new master builder and its associated [`MasterChannel`]. + /// + /// The channel can be cloned and stored for later use. Async operations + /// on the channel require the task to be running. + pub fn new(config: MasterChannelConfig) -> (Self, MasterChannel) { + let (tx, rx) = crate::util::channel::request_channel(); + let channel = MasterChannel::new(tx, MasterChannelType::Stream); + let builder = Self { + channel: channel.clone(), + rx, + config, + enabled: Enabled::No, + associations: AssociationMap::new(), + }; + (builder, channel) + } + + /// Enable communications. This is a synchronous direct mutation of the + /// master's initial state, unlike [`MasterChannel::enable`] which sends + /// a message to an already-running task. + pub fn enable(&mut self) { + self.enabled = Enabled::Yes; + } + + /// Register an association with the master before the task is running. + /// + /// Returns an [`AssociationHandle`] that can be used for post-run operations. + /// Async methods on the handle require the task to be running. + pub fn add_association( + &mut self, + address: EndpointAddress, + config: AssociationConfig, + read_handler: Box, + assoc_handler: Box, + assoc_information: Box, + ) -> Result { + let addr = FragmentAddr { + link: address, + phys: PhysAddr::None, + }; + self.associations.register(Association::new( + addr, + config, + read_handler, + assoc_handler, + assoc_information, + ))?; + Ok(AssociationHandle::new(address, self.channel.clone())) + } + + /// Bind to a TCP transport, consuming the builder. + /// + /// * `link_error_mode` controls how link-layer errors are handled + /// * `connection_handler` controls connection and reconnection behavior + /// * `listener` receives connection state updates + pub fn into_tcp( + self, + link_error_mode: LinkErrorMode, + connection_handler: Box, + listener: Box>, + ) -> MasterTcpTask { + let inner = MasterTask::with_associations( + self.enabled, + self.associations, + LinkModes::stream(link_error_mode), + ParseOptions::get_static(), + self.config, + self.rx, + ); + MasterTcpTask { + inner, + connect_handler: connection_handler, + listener, + } + } + + /// Bind to a serial transport, consuming the builder. + /// + /// * `path` is the serial port path (e.g. "/dev/ttyS0") + /// * `serial_settings` configures baud rate, parity, etc. + /// * `retry_delay` is the delay before retrying after a failed open or link error + /// * `listener` receives port state updates + #[cfg(feature = "serial")] + pub fn into_serial( + self, + path: &str, + serial_settings: crate::serial::SerialSettings, + retry_delay: Duration, + listener: Box>, + ) -> MasterSerialTask { + let inner = MasterTask::with_associations( + self.enabled, + self.associations, + LinkModes::serial(), + ParseOptions::get_static(), + self.config, + self.rx, + ); + MasterSerialTask { + inner, + path: path.to_string(), + settings: serial_settings, + retry_strategy: crate::app::RetryStrategy::new(retry_delay, retry_delay), + listener, + } + } +} + +impl MasterTcpTask { + /// Run the master TCP client event loop. + /// + /// This method consumes the task and runs until the [`MasterChannel`] (and + /// all associated handles) are dropped. + pub async fn run(self) { + let name = self.connect_handler.endpoint_span_name(); + let session = Session::master(self.inner); + let mut client = ClientTask::new( + session, + self.connect_handler, + PostConnectionHandler::Tcp, + self.listener, + ); + client + .run() + .instrument(tracing::info_span!("dnp3-master-tcp-client", "endpoint" = ?name)) + .await; + } +} + +#[cfg(feature = "serial")] +impl MasterSerialTask { + /// Run the master serial event loop. + /// + /// This method consumes the task and runs until the [`MasterChannel`] (and + /// all associated handles) are dropped. + pub async fn run(self) { + let log_path = self.path.clone(); + let session = Session::master(self.inner); + let mut serial = crate::serial::task::SerialTask::new( + &self.path, + self.settings, + session, + self.retry_strategy, + self.listener, + ); + serial + .run() + .instrument(tracing::info_span!("dnp3-master-serial", "port" = ?log_path)) + .await; + } +} diff --git a/dnp3/src/master/mod.rs b/dnp3/src/master/mod.rs index e410bbf8..53ed3606 100644 --- a/dnp3/src/master/mod.rs +++ b/dnp3/src/master/mod.rs @@ -1,4 +1,5 @@ pub use association::*; +pub use builder::*; pub use error::*; pub use file::*; pub use handler::*; @@ -7,6 +8,7 @@ pub use read_handler::*; pub use request::*; mod association; +mod builder; mod error; mod file; mod handler; diff --git a/dnp3/src/master/task.rs b/dnp3/src/master/task.rs index bc938cc1..df3c40c7 100644 --- a/dnp3/src/master/task.rs +++ b/dnp3/src/master/task.rs @@ -36,11 +36,30 @@ impl MasterTask { config: MasterChannelConfig, messages: Receiver, ) -> Self { - let session = MasterSession::new( + Self::with_associations( initial_state, + AssociationMap::new(), + link_modes, + parse_options, + config, + messages, + ) + } + + pub(crate) fn with_associations( + enabled: Enabled, + associations: AssociationMap, + link_modes: LinkModes, + parse_options: ParseOptions, + config: MasterChannelConfig, + messages: Receiver, + ) -> Self { + let session = MasterSession::new( + enabled, config.decode_level, config.tx_buffer_size, messages, + associations, ); let (reader, writer) = crate::transport::create_master_transport_layer( link_modes, @@ -105,11 +124,12 @@ impl MasterSession { decode_level: DecodeLevel, tx_buffer_size: BufferSize<249, 2048>, messages: Receiver, + associations: AssociationMap, ) -> Self { Self { enabled: initial_state, decode_level, - associations: AssociationMap::new(), + associations, messages, tx_buffer: tx_buffer_size.create_buffer(), } From a0d6c54e9370430e8afa173c5df85c3c1403c2e4 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 29 Mar 2026 21:25:06 -0700 Subject: [PATCH 02/14] add non-spawning builder API for master creation (#414) Adds MasterBuilder and UdpMasterBuilder types that separate protocol configuration from transport binding and task execution. Callers control which Tokio runtime or task set the master runs on instead of the library calling tokio::spawn internally. - MasterBuilder: stream transports (TCP, TLS, serial) - UdpMasterBuilder: UDP transport with required SocketAddr per association - Both produce a unified MasterTask with async fn run(self) - Synchronous enable() and add_association() for pre-run configuration - EndpointList::into_connect_handler() for easy ClientConnectionHandler creation - Existing spawn_* functions unchanged (parallel API) - Master example rewritten to use builder API --- dnp3/src/master/builder.rs | 322 +++++++++++++++++---- dnp3/src/tcp/endpoint_list.rs | 23 +- dnp3/src/udp/mod.rs | 2 +- examples/master/src/main.rs | 523 ++++++++++++++-------------------- 4 files changed, 510 insertions(+), 360 deletions(-) diff --git a/dnp3/src/master/builder.rs b/dnp3/src/master/builder.rs index 77dc2e50..1f198ab5 100644 --- a/dnp3/src/master/builder.rs +++ b/dnp3/src/master/builder.rs @@ -1,25 +1,29 @@ +use std::net::SocketAddr; use std::time::Duration; use tracing::Instrument; use super::association::{Association, AssociationMap}; use super::handler::{AssociationHandle, MasterChannel, MasterChannelConfig, MasterChannelType}; -use super::task::MasterTask; +use super::task::MasterTask as InnerMasterTask; use super::{ AssociationConfig, AssociationError, AssociationHandler, AssociationInformation, ReadHandler, }; use crate::app::parse::options::ParseOptions; -use crate::app::Listener; +use crate::app::{Listener, Timeout}; use crate::link::reader::LinkModes; -use crate::link::{EndpointAddress, LinkErrorMode}; +use crate::link::{EndpointAddress, LinkErrorMode, LinkReadMode}; use crate::tcp::client::ClientTask; use crate::tcp::{ClientConnectionHandler, ClientState, PostConnectionHandler}; use crate::transport::FragmentAddr; +use crate::udp::layer::UdpFactory; +use crate::udp::task::UdpTask; use crate::util::channel::Receiver; use crate::util::phys::PhysAddr; use crate::util::session::{Enabled, Session}; -/// Builder for configuring a master task before binding it to a transport. +/// Builder for configuring a stream-based master task (TCP, TLS, serial) +/// before binding it to a transport. /// /// Created via [`MasterBuilder::new`]. The returned [`MasterChannel`] can be /// used for post-run operations, but async methods on it (and on any @@ -33,27 +37,85 @@ pub struct MasterBuilder { associations: AssociationMap, } -/// A master task bound to a TCP transport. Call [`run`](Self::run) to execute +/// Builder for configuring a UDP master task before binding it to a transport. +/// +/// Created via [`UdpMasterBuilder::new`]. The returned [`MasterChannel`] can be +/// used for post-run operations, but async methods on it (and on any +/// [`AssociationHandle`] obtained from [`add_association`](Self::add_association)) +/// require the task to be running. +pub struct UdpMasterBuilder { + channel: MasterChannel, + rx: Receiver, + config: MasterChannelConfig, + enabled: Enabled, + associations: AssociationMap, +} + +/// A master task bound to a transport. Call [`run`](Self::run) to execute /// the event loop. -#[must_use = "a MasterTcpTask does nothing unless you call .run()"] -pub struct MasterTcpTask { - inner: MasterTask, +/// +/// Created by [`MasterBuilder::into_tcp`], [`MasterBuilder::into_tls`], +/// [`MasterBuilder::into_serial`], or [`UdpMasterBuilder::into_udp`]. +#[must_use = "a MasterTask does nothing unless you call .run()"] +pub struct MasterTask { + inner: MasterTaskType, +} + +enum MasterTaskType { + Tcp(TcpTask), + #[cfg(feature = "enable-tls")] + Tls(TlsTask), + #[cfg(feature = "serial")] + Serial(SerialTask), + Udp(UdpMasterTask), +} + +struct TcpTask { + inner: InnerMasterTask, connect_handler: Box, listener: Box>, } -/// A master task bound to a serial transport. Call [`run`](Self::run) to execute -/// the event loop. +#[cfg(feature = "enable-tls")] +struct TlsTask { + inner: InnerMasterTask, + connect_handler: Box, + tls_config: crate::tcp::tls::TlsClientConfig, + listener: Box>, +} + #[cfg(feature = "serial")] -#[must_use = "a MasterSerialTask does nothing unless you call .run()"] -pub struct MasterSerialTask { - inner: MasterTask, +struct SerialTask { + inner: InnerMasterTask, path: String, settings: crate::serial::SerialSettings, retry_strategy: crate::app::RetryStrategy, listener: Box>, } +struct UdpMasterTask { + inner: InnerMasterTask, + local_endpoint: SocketAddr, + retry_delay: Timeout, +} + +fn build_inner_task( + enabled: Enabled, + associations: AssociationMap, + link_modes: LinkModes, + config: MasterChannelConfig, + rx: Receiver, +) -> InnerMasterTask { + InnerMasterTask::with_associations( + enabled, + associations, + link_modes, + ParseOptions::get_static(), + config, + rx, + ) +} + impl MasterBuilder { /// Create a new master builder and its associated [`MasterChannel`]. /// @@ -115,19 +177,49 @@ impl MasterBuilder { link_error_mode: LinkErrorMode, connection_handler: Box, listener: Box>, - ) -> MasterTcpTask { - let inner = MasterTask::with_associations( - self.enabled, - self.associations, - LinkModes::stream(link_error_mode), - ParseOptions::get_static(), - self.config, - self.rx, - ); - MasterTcpTask { - inner, - connect_handler: connection_handler, - listener, + ) -> MasterTask { + MasterTask { + inner: MasterTaskType::Tcp(TcpTask { + inner: build_inner_task( + self.enabled, + self.associations, + LinkModes::stream(link_error_mode), + self.config, + self.rx, + ), + connect_handler: connection_handler, + listener, + }), + } + } + + /// Bind to a TLS transport, consuming the builder. + /// + /// * `link_error_mode` controls how link-layer errors are handled + /// * `connection_handler` controls connection and reconnection behavior + /// * `listener` receives connection state updates + /// * `tls_config` provides the TLS client configuration + #[cfg(feature = "enable-tls")] + pub fn into_tls( + self, + link_error_mode: LinkErrorMode, + connection_handler: Box, + listener: Box>, + tls_config: crate::tcp::tls::TlsClientConfig, + ) -> MasterTask { + MasterTask { + inner: MasterTaskType::Tls(TlsTask { + inner: build_inner_task( + self.enabled, + self.associations, + LinkModes::stream(link_error_mode), + self.config, + self.rx, + ), + connect_handler: connection_handler, + tls_config, + listener, + }), } } @@ -144,31 +236,130 @@ impl MasterBuilder { serial_settings: crate::serial::SerialSettings, retry_delay: Duration, listener: Box>, - ) -> MasterSerialTask { - let inner = MasterTask::with_associations( - self.enabled, - self.associations, - LinkModes::serial(), - ParseOptions::get_static(), - self.config, - self.rx, - ); - MasterSerialTask { - inner, - path: path.to_string(), - settings: serial_settings, - retry_strategy: crate::app::RetryStrategy::new(retry_delay, retry_delay), - listener, + ) -> MasterTask { + MasterTask { + inner: MasterTaskType::Serial(SerialTask { + inner: build_inner_task( + self.enabled, + self.associations, + LinkModes::serial(), + self.config, + self.rx, + ), + path: path.to_string(), + settings: serial_settings, + retry_strategy: crate::app::RetryStrategy::new(retry_delay, retry_delay), + listener, + }), + } + } +} + +impl UdpMasterBuilder { + /// Create a new UDP master builder and its associated [`MasterChannel`]. + /// + /// The channel can be cloned and stored for later use. Async operations + /// on the channel require the task to be running. + pub fn new(config: MasterChannelConfig) -> (Self, MasterChannel) { + let (tx, rx) = crate::util::channel::request_channel(); + let channel = MasterChannel::new(tx, MasterChannelType::Udp); + let builder = Self { + channel: channel.clone(), + rx, + config, + enabled: Enabled::No, + associations: AssociationMap::new(), + }; + (builder, channel) + } + + /// Enable communications. This is a synchronous direct mutation of the + /// master's initial state, unlike [`MasterChannel::enable`] which sends + /// a message to an already-running task. + pub fn enable(&mut self) { + self.enabled = Enabled::Yes; + } + + /// Register a UDP association with the master before the task is running. + /// + /// * `address` is the DNP3 link-layer address of the outstation + /// * `destination` is the IP address and port of the outstation + /// + /// Returns an [`AssociationHandle`] that can be used for post-run operations. + /// Async methods on the handle require the task to be running. + pub fn add_association( + &mut self, + address: EndpointAddress, + destination: SocketAddr, + config: AssociationConfig, + read_handler: Box, + assoc_handler: Box, + assoc_information: Box, + ) -> Result { + let addr = FragmentAddr { + link: address, + phys: PhysAddr::Udp(destination), + }; + self.associations.register(Association::new( + addr, + config, + read_handler, + assoc_handler, + assoc_information, + ))?; + Ok(AssociationHandle::new(address, self.channel.clone())) + } + + /// Bind to a UDP transport, consuming the builder. + /// + /// * `local_endpoint` is the local IP address and port to bind to + /// * `read_mode` controls how link-layer frames are read from datagrams + /// * `retry_delay` is the delay before retrying after a failed bind + pub fn into_udp( + self, + local_endpoint: SocketAddr, + read_mode: LinkReadMode, + retry_delay: Timeout, + ) -> MasterTask { + let link_modes = LinkModes { + error_mode: LinkErrorMode::Discard, + read_mode, + }; + MasterTask { + inner: MasterTaskType::Udp(UdpMasterTask { + inner: build_inner_task( + self.enabled, + self.associations, + link_modes, + self.config, + self.rx, + ), + local_endpoint, + retry_delay, + }), } } } -impl MasterTcpTask { - /// Run the master TCP client event loop. +impl MasterTask { + /// Run the master event loop. /// /// This method consumes the task and runs until the [`MasterChannel`] (and /// all associated handles) are dropped. pub async fn run(self) { + match self.inner { + MasterTaskType::Tcp(task) => task.run().await, + #[cfg(feature = "enable-tls")] + MasterTaskType::Tls(task) => task.run().await, + #[cfg(feature = "serial")] + MasterTaskType::Serial(task) => task.run().await, + MasterTaskType::Udp(task) => task.run().await, + } + } +} + +impl TcpTask { + async fn run(self) { let name = self.connect_handler.endpoint_span_name(); let session = Session::master(self.inner); let mut client = ClientTask::new( @@ -184,14 +375,27 @@ impl MasterTcpTask { } } +#[cfg(feature = "enable-tls")] +impl TlsTask { + async fn run(self) { + let name = self.connect_handler.endpoint_span_name(); + let session = Session::master(self.inner); + let mut client = ClientTask::new( + session, + self.connect_handler, + PostConnectionHandler::Tls(self.tls_config), + self.listener, + ); + client + .run() + .instrument(tracing::info_span!("dnp3-master-tls-client", "endpoint" = ?name)) + .await; + } +} + #[cfg(feature = "serial")] -impl MasterSerialTask { - /// Run the master serial event loop. - /// - /// This method consumes the task and runs until the [`MasterChannel`] (and - /// all associated handles) are dropped. - pub async fn run(self) { - let log_path = self.path.clone(); +impl SerialTask { + async fn run(self) { let session = Session::master(self.inner); let mut serial = crate::serial::task::SerialTask::new( &self.path, @@ -202,7 +406,23 @@ impl MasterSerialTask { ); serial .run() - .instrument(tracing::info_span!("dnp3-master-serial", "port" = ?log_path)) + .instrument(tracing::info_span!("dnp3-master-serial", "port" = ?self.path)) + .await; + } +} + +impl UdpMasterTask { + async fn run(self) { + let local_endpoint = self.local_endpoint; + let session = Session::master(self.inner); + let task = UdpTask { + session, + factory: UdpFactory::bound(local_endpoint), + retry_delay: self.retry_delay, + }; + let _ = task + .run() + .instrument(tracing::info_span!("dnp3-master-udp", "endpoint" = ?local_endpoint)) .await; } } diff --git a/dnp3/src/tcp/endpoint_list.rs b/dnp3/src/tcp/endpoint_list.rs index 1bd6919f..b9fafaac 100644 --- a/dnp3/src/tcp/endpoint_list.rs +++ b/dnp3/src/tcp/endpoint_list.rs @@ -1,4 +1,6 @@ -use crate::tcp::Endpoint; +use crate::app::ConnectStrategy; +use crate::tcp::connector::SimpleConnectHandler; +use crate::tcp::{ClientConnectionHandler, ConnectOptions, Endpoint}; /// List of IP endpoints /// @@ -33,4 +35,23 @@ impl EndpointList { pub fn add(&mut self, addr: String) { self.endpoints.push(addr); } + + /// Create a [`ClientConnectionHandler`] from this endpoint list with the + /// given [`ConnectStrategy`] and default [`ConnectOptions`]. + pub fn into_connect_handler( + self, + strategy: ConnectStrategy, + ) -> Box { + SimpleConnectHandler::create(self, ConnectOptions::default(), strategy) + } + + /// Create a [`ClientConnectionHandler`] from this endpoint list with the + /// given [`ConnectStrategy`] and [`ConnectOptions`]. + pub fn into_connect_handler_with_options( + self, + strategy: ConnectStrategy, + options: ConnectOptions, + ) -> Box { + SimpleConnectHandler::create(self, options, strategy) + } } diff --git a/dnp3/src/udp/mod.rs b/dnp3/src/udp/mod.rs index 9b9d69aa..d3fc6bf5 100644 --- a/dnp3/src/udp/mod.rs +++ b/dnp3/src/udp/mod.rs @@ -1,7 +1,7 @@ pub(crate) mod layer; mod master; mod outstation; -mod task; +pub(crate) mod task; pub use master::*; pub use outstation::*; diff --git a/examples/master/src/main.rs b/examples/master/src/main.rs index aaec0dde..7a90aa6c 100644 --- a/examples/master/src/main.rs +++ b/examples/master/src/main.rs @@ -17,9 +17,8 @@ use dnp3::tcp::*; use clap::{Parser, Subcommand}; use dnp3::outstation::FreezeInterval; -use dnp3::udp::spawn_master_udp; use std::net::SocketAddr; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use dnp3_cli_utils::serial::{DataBitsArg, FlowControlArg, ParityArg, StopBitsArg}; use dnp3_cli_utils::LogLevel; @@ -44,113 +43,128 @@ struct CliArgs { transport: TransportCommand, } +type SetupResult = + Result<(MasterChannel, AssociationHandle, MasterTask), Box>; + #[derive(Debug, Subcommand)] enum TransportCommand { /// Use TCP client transport - TcpClient { - /// IP address and port to connect to - #[arg(short, long, default_value = "127.0.0.1:20000")] - endpoint: SocketAddr, - - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, - }, + TcpClient(TcpClientArgs), /// Use UDP transport - Udp { - /// Local IP address and port to bind to - #[arg(short, long, default_value = "127.0.0.1:20001")] - local_endpoint: SocketAddr, + Udp(UdpArgs), + /// Use serial transport + Serial(SerialArgs), + /// Use TLS with CA chain transport + TlsCa(TlsCaArgs), + /// Use TLS with self-signed certificates + TlsSelfSigned(TlsSelfSignedArgs), +} - /// Remote IP address and port to send to - #[arg(short, long, default_value = "127.0.0.1:20000")] - remote_endpoint: SocketAddr, +#[derive(Debug, Parser)] +struct TcpClientArgs { + /// IP address and port to connect to + #[arg(short, long, default_value = "127.0.0.1:20000")] + endpoint: SocketAddr, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, - }, + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, +} - /// Use serial transport - Serial { - /// Serial port name - #[arg(short, long, default_value = "/dev/ttyS0")] - port: String, +#[derive(Debug, Parser)] +struct UdpArgs { + /// Local IP address and port to bind to + #[arg(short, long, default_value = "127.0.0.1:20001")] + local_endpoint: SocketAddr, - /// Baud rate - #[arg(short, long, default_value = "9600")] - baud_rate: u32, + /// Remote IP address and port to send to + #[arg(short, long, default_value = "127.0.0.1:20000")] + remote_endpoint: SocketAddr, - /// Data bits - #[arg(long, value_enum, default_value_t = DataBitsArg::Eight)] - data_bits: DataBitsArg, + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, +} + +#[derive(Debug, Parser)] +struct SerialArgs { + /// Serial port name + #[arg(short, long, default_value = "/dev/ttyS0")] + port: String, + + /// Baud rate + #[arg(short, long, default_value = "9600")] + baud_rate: u32, - /// Stop bits - #[arg(long, value_enum, default_value_t = StopBitsArg::One)] - stop_bits: StopBitsArg, + /// Data bits + #[arg(long, value_enum, default_value_t = DataBitsArg::Eight)] + data_bits: DataBitsArg, - /// Parity - #[arg(long, value_enum, default_value_t = ParityArg::None)] - parity: ParityArg, + /// Stop bits + #[arg(long, value_enum, default_value_t = StopBitsArg::One)] + stop_bits: StopBitsArg, - /// Flow control - #[arg(long, value_enum, default_value_t = FlowControlArg::None)] - flow_control: FlowControlArg, + /// Parity + #[arg(long, value_enum, default_value_t = ParityArg::None)] + parity: ParityArg, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, - }, + /// Flow control + #[arg(long, value_enum, default_value_t = FlowControlArg::None)] + flow_control: FlowControlArg, - /// Use TLS with CA chain transport - TlsCa { - /// IP address and port to connect to - #[arg(short, long, default_value = "127.0.0.1:20001")] - endpoint: SocketAddr, + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, +} - /// Domain name to verify - #[arg(long, default_value = "test.com")] - domain: String, +#[derive(Debug, Parser)] +struct TlsCaArgs { + /// IP address and port to connect to + #[arg(short, long, default_value = "127.0.0.1:20001")] + endpoint: SocketAddr, - /// Path to CA certificate file - #[arg(long, default_value = "./certs/ca_chain/ca_cert.pem")] - ca_cert: PathBuf, + /// Domain name to verify + #[arg(long, default_value = "test.com")] + domain: String, - /// Path to entity certificate file - #[arg(long, default_value = "./certs/ca_chain/entity1_cert.pem")] - entity_cert: PathBuf, + /// Path to CA certificate file + #[arg(long, default_value = "./certs/ca_chain/ca_cert.pem")] + ca_cert: PathBuf, - /// Path to entity private key file - #[arg(long, default_value = "./certs/ca_chain/entity1_key.pem")] - entity_key: PathBuf, + /// Path to entity certificate file + #[arg(long, default_value = "./certs/ca_chain/entity1_cert.pem")] + entity_cert: PathBuf, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, - }, + /// Path to entity private key file + #[arg(long, default_value = "./certs/ca_chain/entity1_key.pem")] + entity_key: PathBuf, - /// Use TLS with self-signed certificates - TlsSelfSigned { - /// IP address and port to connect to - #[arg(short, long, default_value = "127.0.0.1:20001")] - endpoint: SocketAddr, - - /// Path to peer certificate file - #[arg(long, default_value = "./certs/self_signed/entity2_cert.pem")] - peer_cert: PathBuf, - - /// Path to entity certificate file - #[arg(long, default_value = "./certs/self_signed/entity1_cert.pem")] - entity_cert: PathBuf, - - /// Path to entity private key file - #[arg(long, default_value = "./certs/self_signed/entity1_key.pem")] - entity_key: PathBuf, - - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, - }, + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, +} + +#[derive(Debug, Parser)] +struct TlsSelfSignedArgs { + /// IP address and port to connect to + #[arg(short, long, default_value = "127.0.0.1:20001")] + endpoint: SocketAddr, + + /// Path to peer certificate file + #[arg(long, default_value = "./certs/self_signed/entity2_cert.pem")] + peer_cert: PathBuf, + + /// Path to entity certificate file + #[arg(long, default_value = "./certs/self_signed/entity1_cert.pem")] + entity_cert: PathBuf, + + /// Path to entity private key file + #[arg(long, default_value = "./certs/self_signed/entity1_key.pem")] + entity_key: PathBuf, + + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, } /// read handler that does nothing @@ -383,8 +397,9 @@ async fn main() -> Result<(), Box> { .init(); // ANCHOR_END: logging - // spawn the master channel based on the command line argument - let (mut channel, mut association) = create_channel_and_association(&args).await?; + // create the master channel based on the command line argument + let (channel, mut association, task) = setup(&args)?; + tokio::spawn(task.run()); // create an event poll // ANCHOR: add_poll @@ -396,9 +411,6 @@ async fn main() -> Result<(), Box> { .await?; // ANCHOR_END: add_poll - // enable communications - channel.enable().await?; - let mut handler = CliHandler { poll, channel, @@ -607,120 +619,30 @@ impl CliHandler { } } -// create the specified channel based on the command line argument -async fn create_channel_and_association( - cli: &CliArgs, -) -> Result<(MasterChannel, AssociationHandle), Box> { +fn setup(cli: &CliArgs) -> SetupResult { match &cli.transport { - TransportCommand::TcpClient { - endpoint, - outstation_address, - } => { - let mut channel = create_tcp_channel(cli.master_address, *endpoint)?; - let assoc = add_association(&mut channel, *outstation_address).await?; - Ok((channel, assoc)) - } - TransportCommand::Udp { - local_endpoint, - remote_endpoint, - outstation_address, - } => { - let mut channel = create_udp_channel(cli.master_address, *local_endpoint)?; - let assoc = - add_udp_association(&mut channel, *remote_endpoint, *outstation_address).await?; - Ok((channel, assoc)) - } - TransportCommand::Serial { - port, - baud_rate, - data_bits, - stop_bits, - parity, - flow_control, - outstation_address, - } => { - let mut channel = create_serial_channel( - cli.master_address, - port, - *baud_rate, - *data_bits, - *stop_bits, - *parity, - *flow_control, - )?; - let assoc = add_association(&mut channel, *outstation_address).await?; - Ok((channel, assoc)) - } - TransportCommand::TlsCa { - endpoint, - domain, - ca_cert, - entity_cert, - entity_key, - outstation_address, - } => { - let mut channel = create_tls_channel( - cli.master_address, - *endpoint, - get_tls_authority_config(domain, ca_cert, entity_cert, entity_key)?, - )?; - let assoc = add_association(&mut channel, *outstation_address).await?; - Ok((channel, assoc)) - } - TransportCommand::TlsSelfSigned { - endpoint, - peer_cert, - entity_cert, - entity_key, - outstation_address, - } => { - let mut channel = create_tls_channel( - cli.master_address, - *endpoint, - get_tls_self_signed_config(peer_cert, entity_cert, entity_key)?, - )?; - let assoc = add_association(&mut channel, *outstation_address).await?; - Ok((channel, assoc)) - } + TransportCommand::TcpClient(args) => args.setup(cli.master_address), + TransportCommand::Udp(args) => args.setup(cli.master_address), + TransportCommand::Serial(args) => args.setup(cli.master_address), + TransportCommand::TlsCa(args) => args.setup(cli.master_address), + TransportCommand::TlsSelfSigned(args) => args.setup(cli.master_address), } } -async fn add_association( - channel: &mut MasterChannel, - outstation_address: EndpointAddress, -) -> Result> { - // ANCHOR: association_create - let association = channel - .add_association( - outstation_address, - get_association_config(), - ExampleReadHandler::boxed(), - Box::new(ExampleAssociationHandler), - Box::new(ExampleAssociationInformation), - ) - .await?; - // ANCHOR_END: association_create - Ok(association) -} - -async fn add_udp_association( - channel: &mut MasterChannel, - remote_endpoint: SocketAddr, +fn create_stream_builder( + master_address: EndpointAddress, outstation_address: EndpointAddress, -) -> Result> { - // ANCHOR: association_create_udp - let association = channel - .add_udp_association( - outstation_address, - remote_endpoint, - get_association_config(), - ExampleReadHandler::boxed(), - Box::new(ExampleAssociationHandler), - Box::new(ExampleAssociationInformation), - ) - .await?; - // ANCHOR_END: association_create_udp - Ok(association) +) -> Result<(MasterBuilder, MasterChannel, AssociationHandle), Box> { + let (mut builder, channel) = MasterBuilder::new(get_master_channel_config(master_address)?); + builder.enable(); + let association = builder.add_association( + outstation_address, + get_association_config(), + ExampleReadHandler::boxed(), + Box::new(ExampleAssociationHandler), + Box::new(ExampleAssociationInformation), + )?; + Ok((builder, channel, association)) } // ANCHOR: master_channel_config @@ -751,118 +673,105 @@ fn get_association_config() -> AssociationConfig { } // ANCHOR_END: association_config -fn get_tls_self_signed_config( - peer_cert: &Path, - entity_cert: &Path, - entity_key: &Path, -) -> Result> { - // ANCHOR: tls_self_signed_config - let config = TlsClientConfig::self_signed( - peer_cert, - entity_cert, - entity_key, - None, // no password - MinTlsVersion::V12, - )?; - // ANCHOR_END: tls_self_signed_config - Ok(config) -} - -fn get_tls_authority_config( - domain: &str, - ca_cert: &Path, - entity_cert: &Path, - entity_key: &Path, -) -> Result> { - // ANCHOR: tls_ca_chain_config - let config = TlsClientConfig::full_pki( - Some(domain.to_string()), - ca_cert, - entity_cert, - entity_key, - None, // no password - MinTlsVersion::V12, - )?; - // ANCHOR_END: tls_ca_chain_config - Ok(config) +impl TcpClientArgs { + fn setup(&self, master_address: EndpointAddress) -> SetupResult { + let (builder, channel, association) = + create_stream_builder(master_address, self.outstation_address)?; + let handler = EndpointList::new(self.endpoint.to_string(), &[]) + .into_connect_handler(ConnectStrategy::default()); + let task = builder.into_tcp(LinkErrorMode::Close, handler, NullListener::create()); + Ok((channel, association, task)) + } } -fn create_tcp_channel( - master_address: EndpointAddress, - endpoint: SocketAddr, -) -> Result> { - // ANCHOR: create_master_tcp_channel - let channel = spawn_master_tcp_client( - LinkErrorMode::Close, - get_master_channel_config(master_address)?, - EndpointList::new(endpoint.to_string(), &[]), - ConnectStrategy::default(), - NullListener::create(), - ); - // ANCHOR_END: create_master_tcp_channel - Ok(channel) +impl UdpArgs { + fn setup(&self, master_address: EndpointAddress) -> SetupResult { + let (mut builder, channel) = + UdpMasterBuilder::new(get_master_channel_config(master_address)?); + builder.enable(); + let association = builder.add_association( + self.outstation_address, + self.remote_endpoint, + get_association_config(), + ExampleReadHandler::boxed(), + Box::new(ExampleAssociationHandler), + Box::new(ExampleAssociationInformation), + )?; + let task = builder.into_udp( + self.local_endpoint, + LinkReadMode::Datagram, + Timeout::from_secs(5)?, + ); + Ok((channel, association, task)) + } } -fn create_udp_channel( - master_address: EndpointAddress, - local_endpoint: SocketAddr, -) -> Result> { - // ANCHOR: create_master_udp_channel - let channel = spawn_master_udp( - local_endpoint, - LinkReadMode::Datagram, - Timeout::from_secs(5)?, - get_master_channel_config(master_address)?, - ); - // ANCHOR_END: create_master_udp_channel - Ok(channel) +impl SerialArgs { + fn setup(&self, master_address: EndpointAddress) -> SetupResult { + let (builder, channel, association) = + create_stream_builder(master_address, self.outstation_address)?; + let settings = SerialSettings { + baud_rate: self.baud_rate, + data_bits: self.data_bits.into(), + stop_bits: self.stop_bits.into(), + parity: self.parity.into(), + flow_control: self.flow_control.into(), + }; + let task = builder.into_serial( + &self.port, + settings, + Duration::from_secs(1), + NullListener::create(), + ); + Ok((channel, association, task)) + } } -fn create_serial_channel( - master_address: EndpointAddress, - port: &str, - baud_rate: u32, - data_bits: DataBitsArg, - stop_bits: StopBitsArg, - parity: ParityArg, - flow_control: FlowControlArg, -) -> Result> { - // ANCHOR: create_master_serial_channel - let settings = SerialSettings { - baud_rate, - data_bits: data_bits.into(), - stop_bits: stop_bits.into(), - parity: parity.into(), - flow_control: flow_control.into(), - }; - - let channel = spawn_master_serial( - get_master_channel_config(master_address)?, - port, - settings, - Duration::from_secs(1), - NullListener::create(), - ); - // ANCHOR_END: create_master_serial_channel - Ok(channel) +impl TlsCaArgs { + fn setup(&self, master_address: EndpointAddress) -> SetupResult { + let (builder, channel, association) = + create_stream_builder(master_address, self.outstation_address)?; + let handler = EndpointList::new(self.endpoint.to_string(), &[]) + .into_connect_handler(ConnectStrategy::default()); + let tls_config = TlsClientConfig::full_pki( + Some(self.domain.to_string()), + &self.ca_cert, + &self.entity_cert, + &self.entity_key, + None, + MinTlsVersion::V12, + )?; + let task = builder.into_tls( + LinkErrorMode::Close, + handler, + NullListener::create(), + tls_config, + ); + Ok((channel, association, task)) + } } -fn create_tls_channel( - master_address: EndpointAddress, - endpoint: SocketAddr, - tls_config: TlsClientConfig, -) -> Result> { - // ANCHOR: create_master_tls_channel - let channel = spawn_master_tls_client( - LinkErrorMode::Close, - get_master_channel_config(master_address)?, - EndpointList::new(endpoint.to_string(), &[]), - ConnectStrategy::default(), - NullListener::create(), - tls_config, - ); - // ANCHOR_END: create_master_tls_channel - Ok(channel) +impl TlsSelfSignedArgs { + fn setup(&self, master_address: EndpointAddress) -> SetupResult { + let (builder, channel, association) = + create_stream_builder(master_address, self.outstation_address)?; + let handler = EndpointList::new(self.endpoint.to_string(), &[]) + .into_connect_handler(ConnectStrategy::default()); + let tls_config = TlsClientConfig::self_signed( + &self.peer_cert, + &self.entity_cert, + &self.entity_key, + None, + MinTlsVersion::V12, + )?; + let task = builder.into_tls( + LinkErrorMode::Close, + handler, + NullListener::create(), + tls_config, + ); + Ok((channel, association, task)) + } } fn print_file_info(info: FileInfo) { From 8407f51dbcf3f10e5e347790d9c7802d6c05dd12 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 29 Mar 2026 21:55:27 -0700 Subject: [PATCH 03/14] add association polls without async --- dnp3/src/master/association.rs | 5 ++ dnp3/src/master/builder.rs | 113 +++++++++++++++++------- examples/master/src/main.rs | 154 +++++++++++++++++++-------------- 3 files changed, 177 insertions(+), 95 deletions(-) diff --git a/dnp3/src/master/association.rs b/dnp3/src/master/association.rs index e1fe626e..b468e14f 100644 --- a/dnp3/src/master/association.rs +++ b/dnp3/src/master/association.rs @@ -14,6 +14,7 @@ use crate::master::extract::extract_measurements; use crate::master::handler::AssociationHandler; use crate::master::messages::AssociationMsgType; use crate::master::poll::{PollHandle, PollMap, PollMsg}; +use crate::master::request::ReadRequest; use crate::master::request::{Classes, EventClasses, TimeSyncProcedure}; use crate::master::tasks::auto::AutoTask; use crate::master::tasks::time::TimeSyncTask; @@ -348,6 +349,10 @@ impl Association { } } + pub(crate) fn add_poll(&mut self, request: ReadRequest, period: Duration) -> u64 { + self.polls.add(request, period) + } + pub(crate) fn process_message(&mut self, msg: AssociationMsgType, is_connected: bool) { match msg { AssociationMsgType::QueueTask(task) => { diff --git a/dnp3/src/master/builder.rs b/dnp3/src/master/builder.rs index 1f198ab5..5b284573 100644 --- a/dnp3/src/master/builder.rs +++ b/dnp3/src/master/builder.rs @@ -5,6 +5,8 @@ use tracing::Instrument; use super::association::{Association, AssociationMap}; use super::handler::{AssociationHandle, MasterChannel, MasterChannelConfig, MasterChannelType}; +use super::poll::PollHandle; +use super::request::ReadRequest; use super::task::MasterTask as InnerMasterTask; use super::{ AssociationConfig, AssociationError, AssociationHandler, AssociationInformation, ReadHandler, @@ -51,6 +53,17 @@ pub struct UdpMasterBuilder { associations: AssociationMap, } +/// Builder for configuring an association and its polls before registration. +/// +/// Created via [`MasterBuilder::new_association`] or [`UdpMasterBuilder::new_association`]. +/// Add polls with [`add_poll`](Self::add_poll), then register with +/// [`MasterBuilder::add_association`] or [`UdpMasterBuilder::add_association`]. +pub struct AssociationBuilder { + address: EndpointAddress, + channel: MasterChannel, + association: Association, +} + /// A master task bound to a transport. Call [`run`](Self::run) to execute /// the event loop. /// @@ -141,30 +154,43 @@ impl MasterBuilder { self.enabled = Enabled::Yes; } - /// Register an association with the master before the task is running. - /// - /// Returns an [`AssociationHandle`] that can be used for post-run operations. - /// Async methods on the handle require the task to be running. - pub fn add_association( - &mut self, + /// Create an [`AssociationBuilder`] for configuring an association and its + /// polls before registration. + pub fn new_association( + &self, address: EndpointAddress, config: AssociationConfig, read_handler: Box, assoc_handler: Box, assoc_information: Box, - ) -> Result { + ) -> AssociationBuilder { let addr = FragmentAddr { link: address, phys: PhysAddr::None, }; - self.associations.register(Association::new( - addr, - config, - read_handler, - assoc_handler, - assoc_information, - ))?; - Ok(AssociationHandle::new(address, self.channel.clone())) + AssociationBuilder { + address, + channel: self.channel.clone(), + association: Association::new( + addr, + config, + read_handler, + assoc_handler, + assoc_information, + ), + } + } + + /// Register a configured association with the master. + /// + /// Returns an [`AssociationHandle`] that can be used for post-run operations. + /// Async methods on the handle require the task to be running. + pub fn add_association( + &mut self, + builder: AssociationBuilder, + ) -> Result { + self.associations.register(builder.association)?; + Ok(AssociationHandle::new(builder.address, builder.channel)) } /// Bind to a TCP transport, consuming the builder. @@ -280,34 +306,47 @@ impl UdpMasterBuilder { self.enabled = Enabled::Yes; } - /// Register a UDP association with the master before the task is running. + /// Create an [`AssociationBuilder`] for configuring a UDP association and + /// its polls before registration. /// /// * `address` is the DNP3 link-layer address of the outstation /// * `destination` is the IP address and port of the outstation - /// - /// Returns an [`AssociationHandle`] that can be used for post-run operations. - /// Async methods on the handle require the task to be running. - pub fn add_association( - &mut self, + pub fn new_association( + &self, address: EndpointAddress, destination: SocketAddr, config: AssociationConfig, read_handler: Box, assoc_handler: Box, assoc_information: Box, - ) -> Result { + ) -> AssociationBuilder { let addr = FragmentAddr { link: address, phys: PhysAddr::Udp(destination), }; - self.associations.register(Association::new( - addr, - config, - read_handler, - assoc_handler, - assoc_information, - ))?; - Ok(AssociationHandle::new(address, self.channel.clone())) + AssociationBuilder { + address, + channel: self.channel.clone(), + association: Association::new( + addr, + config, + read_handler, + assoc_handler, + assoc_information, + ), + } + } + + /// Register a configured association with the master. + /// + /// Returns an [`AssociationHandle`] that can be used for post-run operations. + /// Async methods on the handle require the task to be running. + pub fn add_association( + &mut self, + builder: AssociationBuilder, + ) -> Result { + self.associations.register(builder.association)?; + Ok(AssociationHandle::new(builder.address, builder.channel)) } /// Bind to a UDP transport, consuming the builder. @@ -341,6 +380,20 @@ impl UdpMasterBuilder { } } +impl AssociationBuilder { + /// Add a periodic poll to the association. + /// + /// Returns a [`PollHandle`] that can be used for post-run operations. + /// Async methods on the handle require the task to be running. + pub fn add_poll(&mut self, request: ReadRequest, period: Duration) -> PollHandle { + let id = self.association.add_poll(request, period); + PollHandle::new( + AssociationHandle::new(self.address, self.channel.clone()), + id, + ) + } +} + impl MasterTask { /// Run the master event loop. /// diff --git a/examples/master/src/main.rs b/examples/master/src/main.rs index 7a90aa6c..7077e62a 100644 --- a/examples/master/src/main.rs +++ b/examples/master/src/main.rs @@ -43,9 +43,6 @@ struct CliArgs { transport: TransportCommand, } -type SetupResult = - Result<(MasterChannel, AssociationHandle, MasterTask), Box>; - #[derive(Debug, Subcommand)] enum TransportCommand { /// Use TCP client transport @@ -398,25 +395,9 @@ async fn main() -> Result<(), Box> { // ANCHOR_END: logging // create the master channel based on the command line argument - let (channel, mut association, task) = setup(&args)?; + let (mut handler, task) = setup(&args)?; tokio::spawn(task.run()); - // create an event poll - // ANCHOR: add_poll - let poll = association - .add_poll( - ReadRequest::ClassScan(Classes::class123()), - Duration::from_secs(5), - ) - .await?; - // ANCHOR_END: add_poll - - let mut handler = CliHandler { - poll, - channel, - association, - }; - let mut reader = FramedRead::new(tokio::io::stdin(), LinesCodec::new()); loop { @@ -619,7 +600,7 @@ impl CliHandler { } } -fn setup(cli: &CliArgs) -> SetupResult { +fn setup(cli: &CliArgs) -> Result<(CliHandler, MasterTask), Box> { match &cli.transport { TransportCommand::TcpClient(args) => args.setup(cli.master_address), TransportCommand::Udp(args) => args.setup(cli.master_address), @@ -632,17 +613,29 @@ fn setup(cli: &CliArgs) -> SetupResult { fn create_stream_builder( master_address: EndpointAddress, outstation_address: EndpointAddress, -) -> Result<(MasterBuilder, MasterChannel, AssociationHandle), Box> { +) -> Result<(MasterBuilder, CliHandler), Box> { let (mut builder, channel) = MasterBuilder::new(get_master_channel_config(master_address)?); builder.enable(); - let association = builder.add_association( + let mut assoc = builder.new_association( outstation_address, get_association_config(), ExampleReadHandler::boxed(), Box::new(ExampleAssociationHandler), Box::new(ExampleAssociationInformation), - )?; - Ok((builder, channel, association)) + ); + let poll = assoc.add_poll( + ReadRequest::ClassScan(Classes::class123()), + Duration::from_secs(5), + ); + let association = builder.add_association(assoc)?; + Ok(( + builder, + CliHandler { + poll, + channel, + association, + }, + )) } // ANCHOR: master_channel_config @@ -674,42 +667,63 @@ fn get_association_config() -> AssociationConfig { // ANCHOR_END: association_config impl TcpClientArgs { - fn setup(&self, master_address: EndpointAddress) -> SetupResult { - let (builder, channel, association) = - create_stream_builder(master_address, self.outstation_address)?; - let handler = EndpointList::new(self.endpoint.to_string(), &[]) + fn setup( + &self, + master_address: EndpointAddress, + ) -> Result<(CliHandler, MasterTask), Box> { + let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; + let connect = EndpointList::new(self.endpoint.to_string(), &[]) .into_connect_handler(ConnectStrategy::default()); - let task = builder.into_tcp(LinkErrorMode::Close, handler, NullListener::create()); - Ok((channel, association, task)) + Ok(( + handler, + builder.into_tcp(LinkErrorMode::Close, connect, NullListener::create()), + )) } } impl UdpArgs { - fn setup(&self, master_address: EndpointAddress) -> SetupResult { + fn setup( + &self, + master_address: EndpointAddress, + ) -> Result<(CliHandler, MasterTask), Box> { let (mut builder, channel) = UdpMasterBuilder::new(get_master_channel_config(master_address)?); builder.enable(); - let association = builder.add_association( + let mut assoc = builder.new_association( self.outstation_address, self.remote_endpoint, get_association_config(), ExampleReadHandler::boxed(), Box::new(ExampleAssociationHandler), Box::new(ExampleAssociationInformation), - )?; + ); + let poll = assoc.add_poll( + ReadRequest::ClassScan(Classes::class123()), + Duration::from_secs(5), + ); + let association = builder.add_association(assoc)?; let task = builder.into_udp( self.local_endpoint, LinkReadMode::Datagram, Timeout::from_secs(5)?, ); - Ok((channel, association, task)) + Ok(( + CliHandler { + poll, + channel, + association, + }, + task, + )) } } impl SerialArgs { - fn setup(&self, master_address: EndpointAddress) -> SetupResult { - let (builder, channel, association) = - create_stream_builder(master_address, self.outstation_address)?; + fn setup( + &self, + master_address: EndpointAddress, + ) -> Result<(CliHandler, MasterTask), Box> { + let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; let settings = SerialSettings { baud_rate: self.baud_rate, data_bits: self.data_bits.into(), @@ -717,21 +731,25 @@ impl SerialArgs { parity: self.parity.into(), flow_control: self.flow_control.into(), }; - let task = builder.into_serial( - &self.port, - settings, - Duration::from_secs(1), - NullListener::create(), - ); - Ok((channel, association, task)) + Ok(( + handler, + builder.into_serial( + &self.port, + settings, + Duration::from_secs(1), + NullListener::create(), + ), + )) } } impl TlsCaArgs { - fn setup(&self, master_address: EndpointAddress) -> SetupResult { - let (builder, channel, association) = - create_stream_builder(master_address, self.outstation_address)?; - let handler = EndpointList::new(self.endpoint.to_string(), &[]) + fn setup( + &self, + master_address: EndpointAddress, + ) -> Result<(CliHandler, MasterTask), Box> { + let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; + let connect = EndpointList::new(self.endpoint.to_string(), &[]) .into_connect_handler(ConnectStrategy::default()); let tls_config = TlsClientConfig::full_pki( Some(self.domain.to_string()), @@ -741,21 +759,25 @@ impl TlsCaArgs { None, MinTlsVersion::V12, )?; - let task = builder.into_tls( - LinkErrorMode::Close, + Ok(( handler, - NullListener::create(), - tls_config, - ); - Ok((channel, association, task)) + builder.into_tls( + LinkErrorMode::Close, + connect, + NullListener::create(), + tls_config, + ), + )) } } impl TlsSelfSignedArgs { - fn setup(&self, master_address: EndpointAddress) -> SetupResult { - let (builder, channel, association) = - create_stream_builder(master_address, self.outstation_address)?; - let handler = EndpointList::new(self.endpoint.to_string(), &[]) + fn setup( + &self, + master_address: EndpointAddress, + ) -> Result<(CliHandler, MasterTask), Box> { + let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; + let connect = EndpointList::new(self.endpoint.to_string(), &[]) .into_connect_handler(ConnectStrategy::default()); let tls_config = TlsClientConfig::self_signed( &self.peer_cert, @@ -764,13 +786,15 @@ impl TlsSelfSignedArgs { None, MinTlsVersion::V12, )?; - let task = builder.into_tls( - LinkErrorMode::Close, + Ok(( handler, - NullListener::create(), - tls_config, - ); - Ok((channel, association, task)) + builder.into_tls( + LinkErrorMode::Close, + connect, + NullListener::create(), + tls_config, + ), + )) } } From a5d06a45a5f540c093115ebe9c69845cf8bf208f Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 13:21:46 -0700 Subject: [PATCH 04/14] Improve no-spawn master builder APIs --- dnp3/src/master/association.rs | 113 ++++++- dnp3/src/master/builder.rs | 317 ++++++++++++++---- dnp3/src/master/poll.rs | 92 +++++- dnp3/src/master/task.rs | 4 + dnp3/src/serial/task.rs | 8 +- dnp3/src/tcp/client.rs | 8 +- examples/master/src/main.rs | 579 ++++++++++++++++++--------------- 7 files changed, 792 insertions(+), 329 deletions(-) diff --git a/dnp3/src/master/association.rs b/dnp3/src/master/association.rs index b468e14f..4826236d 100644 --- a/dnp3/src/master/association.rs +++ b/dnp3/src/master/association.rs @@ -1,3 +1,4 @@ +use std::collections::btree_map::Entry; use std::collections::{BTreeMap, VecDeque}; use std::time::Duration; @@ -349,6 +350,39 @@ impl Association { } } + pub(crate) fn new_deferred( + address: FragmentAddr, + config: AssociationConfig, + read_handler: Box, + assoc_handler: Box, + assoc_info: Box, + ) -> Self { + Self { + response_timeout: config.response_timeout, + address, + seq: Sequence::default(), + last_unsol_frag: None, + request_queue: VecDeque::new(), + max_request_queue_size: config.max_queued_user_requests, + auto_tasks: TaskStates::new(), + read_handler, + assoc_handler, + assoc_info, + config, + polls: PollMap::new_deferred(), + next_link_status_deadline: None, + startup_integrity_done: false, + events_available: EventClasses::none(), + } + } + + pub(crate) fn start(&mut self, now: Instant) { + if self.polls.start(now) { + self.next_link_status_deadline = + self.config.keep_alive_timeout.map(|delay| now + delay); + } + } + pub(crate) fn add_poll(&mut self, request: ReadRequest, period: Duration) -> u64 { self.polls.add(request, period) } @@ -755,6 +789,12 @@ impl AssociationMap { } } + pub(crate) fn start(&mut self, now: Instant) { + for association in self.map.values_mut() { + association.start(now); + } + } + pub(crate) fn get_timeout(&self, address: EndpointAddress) -> Result { match self.map.get(&address) { Some(x) => Ok(x.response_timeout), @@ -769,13 +809,21 @@ impl AssociationMap { } pub(crate) fn register(&mut self, session: Association) -> Result<(), AssociationError> { - if self.map.contains_key(&session.address.link) { - return Err(AssociationError::DuplicateAddress(session.address.link)); - } + self.register_and_get(session).map(|_| ()) + } - self.priority.push_back(session.address.link); - self.map.insert(session.address.link, session); - Ok(()) + pub(crate) fn register_and_get( + &mut self, + session: Association, + ) -> Result<&mut Association, AssociationError> { + let address = session.address.link; + match self.map.entry(address) { + Entry::Occupied(_) => Err(AssociationError::DuplicateAddress(address)), + Entry::Vacant(entry) => { + self.priority.push_back(address); + Ok(entry.insert(session)) + } + } } pub(crate) fn remove(&mut self, address: EndpointAddress) { @@ -841,3 +889,56 @@ impl AssociationMap { Next::None } } + +#[cfg(test)] +mod timer_tests { + use super::*; + use crate::master::{AssociationHandler, AssociationInformation, ReadHandler}; + use crate::util::phys::PhysAddr; + + struct NullReadHandler; + impl ReadHandler for NullReadHandler {} + + struct NullAssociationHandler; + impl AssociationHandler for NullAssociationHandler {} + + struct NullAssociationInformation; + impl AssociationInformation for NullAssociationInformation {} + + fn deferred_association(keep_alive_timeout: Duration) -> Association { + let mut config = AssociationConfig::quiet(); + config.keep_alive_timeout = Some(keep_alive_timeout); + Association::new_deferred( + FragmentAddr { + link: EndpointAddress::try_new(1).unwrap(), + phys: PhysAddr::None, + }, + config, + Box::new(NullReadHandler), + Box::new(NullAssociationHandler), + Box::new(NullAssociationInformation), + ) + } + + #[test] + fn deferred_keep_alive_is_scheduled_from_start_time() { + let timeout = Duration::from_secs(5); + let start = Instant::now(); + let mut association = deferred_association(timeout); + + assert_eq!(association.next_link_status_deadline, None); + association.start(start); + assert_eq!(association.next_link_status_deadline, Some(start + timeout)); + } + + #[test] + fn starting_twice_does_not_reset_keep_alive_deadline() { + let timeout = Duration::from_secs(5); + let start = Instant::now(); + let mut association = deferred_association(timeout); + + association.start(start); + association.start(start + Duration::from_secs(1)); + assert_eq!(association.next_link_status_deadline, Some(start + timeout)); + } +} diff --git a/dnp3/src/master/builder.rs b/dnp3/src/master/builder.rs index 5b284573..04e0ed8c 100644 --- a/dnp3/src/master/builder.rs +++ b/dnp3/src/master/builder.rs @@ -1,6 +1,7 @@ use std::net::SocketAddr; use std::time::Duration; +use tokio::time::Instant; use tracing::Instrument; use super::association::{Association, AssociationMap}; @@ -53,15 +54,17 @@ pub struct UdpMasterBuilder { associations: AssociationMap, } -/// Builder for configuring an association and its polls before registration. +/// Builder for configuring an association and its polls before the master task +/// starts. /// -/// Created via [`MasterBuilder::new_association`] or [`UdpMasterBuilder::new_association`]. -/// Add polls with [`add_poll`](Self::add_poll), then register with -/// [`MasterBuilder::add_association`] or [`UdpMasterBuilder::add_association`]. -pub struct AssociationBuilder { - address: EndpointAddress, - channel: MasterChannel, - association: Association, +/// Created by [`MasterBuilder::add_association`] or +/// [`UdpMasterBuilder::add_association`] after the association has been +/// registered. Its lifetime binds it to the parent master builder, preventing +/// it from being registered with a different master or transport. +#[must_use = "call .into_handle() if runtime access to the association is needed"] +pub struct AssociationBuilder<'a> { + handle: AssociationHandle, + association: &'a mut Association, } /// A master task bound to a transport. Call [`run`](Self::run) to execute @@ -154,43 +157,34 @@ impl MasterBuilder { self.enabled = Enabled::Yes; } - /// Create an [`AssociationBuilder`] for configuring an association and its - /// polls before registration. - pub fn new_association( - &self, + /// Register an association and return a scoped builder for adding polls + /// before the master task starts. + pub fn add_association( + &mut self, address: EndpointAddress, config: AssociationConfig, read_handler: Box, assoc_handler: Box, assoc_information: Box, - ) -> AssociationBuilder { + ) -> Result, AssociationError> { let addr = FragmentAddr { link: address, phys: PhysAddr::None, }; - AssociationBuilder { - address, - channel: self.channel.clone(), - association: Association::new( + let handle = AssociationHandle::new(address, self.channel.clone()); + let association = self + .associations + .register_and_get(Association::new_deferred( addr, config, read_handler, assoc_handler, assoc_information, - ), - } - } - - /// Register a configured association with the master. - /// - /// Returns an [`AssociationHandle`] that can be used for post-run operations. - /// Async methods on the handle require the task to be running. - pub fn add_association( - &mut self, - builder: AssociationBuilder, - ) -> Result { - self.associations.register(builder.association)?; - Ok(AssociationHandle::new(builder.address, builder.channel)) + ))?; + Ok(AssociationBuilder { + handle, + association, + }) } /// Bind to a TCP transport, consuming the builder. @@ -306,47 +300,38 @@ impl UdpMasterBuilder { self.enabled = Enabled::Yes; } - /// Create an [`AssociationBuilder`] for configuring a UDP association and - /// its polls before registration. + /// Register a UDP association and return a scoped builder for adding polls + /// before the master task starts. /// /// * `address` is the DNP3 link-layer address of the outstation /// * `destination` is the IP address and port of the outstation - pub fn new_association( - &self, + pub fn add_association( + &mut self, address: EndpointAddress, destination: SocketAddr, config: AssociationConfig, read_handler: Box, assoc_handler: Box, assoc_information: Box, - ) -> AssociationBuilder { + ) -> Result, AssociationError> { let addr = FragmentAddr { link: address, phys: PhysAddr::Udp(destination), }; - AssociationBuilder { - address, - channel: self.channel.clone(), - association: Association::new( + let handle = AssociationHandle::new(address, self.channel.clone()); + let association = self + .associations + .register_and_get(Association::new_deferred( addr, config, read_handler, assoc_handler, assoc_information, - ), - } - } - - /// Register a configured association with the master. - /// - /// Returns an [`AssociationHandle`] that can be used for post-run operations. - /// Async methods on the handle require the task to be running. - pub fn add_association( - &mut self, - builder: AssociationBuilder, - ) -> Result { - self.associations.register(builder.association)?; - Ok(AssociationHandle::new(builder.address, builder.channel)) + ))?; + Ok(AssociationBuilder { + handle, + association, + }) } /// Bind to a UDP transport, consuming the builder. @@ -380,17 +365,22 @@ impl UdpMasterBuilder { } } -impl AssociationBuilder { +impl AssociationBuilder<'_> { /// Add a periodic poll to the association. /// /// Returns a [`PollHandle`] that can be used for post-run operations. /// Async methods on the handle require the task to be running. pub fn add_poll(&mut self, request: ReadRequest, period: Duration) -> PollHandle { let id = self.association.add_poll(request, period); - PollHandle::new( - AssociationHandle::new(self.address, self.channel.clone()), - id, - ) + PollHandle::new(self.handle.clone(), id) + } + + /// Convert this scoped builder into an [`AssociationHandle`]. + /// + /// This releases the mutable borrow of the parent master builder. Async + /// methods on the returned handle require the master task to be running. + pub fn into_handle(self) -> AssociationHandle { + self.handle } } @@ -399,7 +389,8 @@ impl MasterTask { /// /// This method consumes the task and runs until the [`MasterChannel`] (and /// all associated handles) are dropped. - pub async fn run(self) { + pub async fn run(mut self) { + self.inner.start(Instant::now()); match self.inner { MasterTaskType::Tcp(task) => task.run().await, #[cfg(feature = "enable-tls")] @@ -411,6 +402,19 @@ impl MasterTask { } } +impl MasterTaskType { + fn start(&mut self, now: Instant) { + match self { + MasterTaskType::Tcp(task) => task.inner.start(now), + #[cfg(feature = "enable-tls")] + MasterTaskType::Tls(task) => task.inner.start(now), + #[cfg(feature = "serial")] + MasterTaskType::Serial(task) => task.inner.start(now), + MasterTaskType::Udp(task) => task.inner.start(now), + } + } +} + impl TcpTask { async fn run(self) { let name = self.connect_handler.endpoint_span_name(); @@ -479,3 +483,196 @@ impl UdpMasterTask { .await; } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::app::MaybeAsync; + use crate::master::messages::{AssociationMsgType, Message}; + use crate::master::poll::PollMsg; + use std::marker::PhantomData; + + struct ChannelListener { + tx: tokio::sync::mpsc::UnboundedSender, + } + + impl Listener for ChannelListener { + fn update(&mut self, value: T) -> MaybeAsync<()> { + let _ = self.tx.send(value); + MaybeAsync::ready(()) + } + } + + fn channel_listener() -> ( + Box>, + tokio::sync::mpsc::UnboundedReceiver, + ) { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + (Box::new(ChannelListener { tx }), rx) + } + + struct NeverConnect { + _not_constructible: PhantomData<()>, + } + + impl NeverConnect { + fn create() -> Box { + Box::new(Self { + _not_constructible: PhantomData, + }) + } + } + + impl ClientConnectionHandler for NeverConnect { + fn endpoint_span_name(&self) -> String { + "test".to_string() + } + + fn disconnected(&mut self, _: SocketAddr, _: Option<&str>) -> Duration { + Duration::from_secs(60) + } + + fn next(&mut self) -> Result { + Err(Duration::from_secs(60)) + } + } + + struct NullReadHandler; + impl ReadHandler for NullReadHandler {} + + struct NullAssociationHandler; + impl AssociationHandler for NullAssociationHandler {} + + struct NullAssociationInformation; + impl AssociationInformation for NullAssociationInformation {} + + fn endpoint(value: u16) -> EndpointAddress { + EndpointAddress::try_new(value).unwrap() + } + + fn add_stream_association( + builder: &mut MasterBuilder, + address: EndpointAddress, + ) -> Result, AssociationError> { + builder.add_association( + address, + AssociationConfig::quiet(), + Box::new(NullReadHandler), + Box::new(NullAssociationHandler), + Box::new(NullAssociationInformation), + ) + } + + #[tokio::test] + async fn preconfigured_poll_handle_targets_parent_builder() { + let (mut builder, _channel) = MasterBuilder::new(MasterChannelConfig::new(endpoint(1))); + let address = endpoint(1024); + + let mut association = add_stream_association(&mut builder, address).unwrap(); + let mut poll = association.add_poll( + ReadRequest::ClassScan(crate::master::Classes::all()), + Duration::from_secs(5), + ); + let handle = association.into_handle(); + + assert_eq!(handle.address(), address); + poll.demand().await.unwrap(); + + match builder.rx.receive().await.unwrap() { + Message::Association(msg) => { + assert_eq!(msg.address, address); + assert!(matches!( + msg.details, + AssociationMsgType::Poll(PollMsg::Demand(0)) + )); + } + _ => panic!("expected an association message"), + } + } + + #[test] + fn duplicate_address_fails_before_a_scoped_builder_is_returned() { + let (mut builder, _channel) = MasterBuilder::new(MasterChannelConfig::new(endpoint(1))); + let address = endpoint(1024); + + let association = add_stream_association(&mut builder, address).unwrap(); + let _handle = association.into_handle(); + + assert!(matches!( + add_stream_association(&mut builder, address), + Err(AssociationError::DuplicateAddress(x)) if x == address + )); + } + + #[tokio::test] + async fn enabled_tcp_task_does_not_report_disabled() { + let (mut builder, channel) = MasterBuilder::new(MasterChannelConfig::new(endpoint(1))); + builder.enable(); + let (listener, mut states) = channel_listener(); + let task = builder.into_tcp(LinkErrorMode::Close, NeverConnect::create(), listener); + + let join = tokio::spawn(task.run()); + assert_eq!(states.recv().await, Some(ClientState::Connecting)); + + drop(channel); + join.await.unwrap(); + } + + #[tokio::test] + async fn disabled_tcp_task_reports_disabled() { + let (builder, channel) = MasterBuilder::new(MasterChannelConfig::new(endpoint(1))); + let (listener, mut states) = channel_listener(); + let task = builder.into_tcp(LinkErrorMode::Close, NeverConnect::create(), listener); + + let join = tokio::spawn(task.run()); + assert_eq!(states.recv().await, Some(ClientState::Disabled)); + + drop(channel); + join.await.unwrap(); + } + + #[cfg(feature = "serial")] + #[tokio::test] + async fn enabled_serial_task_does_not_report_disabled() { + let (mut builder, channel) = MasterBuilder::new(MasterChannelConfig::new(endpoint(1))); + builder.enable(); + let (listener, mut states) = channel_listener(); + let task = builder.into_serial( + "/path/that/does/not/exist/dnp3-test", + crate::serial::SerialSettings::default(), + Duration::from_secs(60), + listener, + ); + + let join = tokio::spawn(task.run()); + assert!(matches!( + states.recv().await, + Some(crate::serial::PortState::Wait(_)) + )); + + drop(channel); + join.await.unwrap(); + } + + #[cfg(feature = "serial")] + #[tokio::test] + async fn disabled_serial_task_reports_disabled() { + let (builder, channel) = MasterBuilder::new(MasterChannelConfig::new(endpoint(1))); + let (listener, mut states) = channel_listener(); + let task = builder.into_serial( + "/path/that/does/not/exist/dnp3-test", + crate::serial::SerialSettings::default(), + Duration::from_secs(60), + listener, + ); + + let join = tokio::spawn(task.run()); + assert_eq!( + states.recv().await, + Some(crate::serial::PortState::Disabled) + ); + + drop(channel); + join.await.unwrap(); + } +} diff --git a/dnp3/src/master/poll.rs b/dnp3/src/master/poll.rs index 194a6c32..440cf464 100644 --- a/dnp3/src/master/poll.rs +++ b/dnp3/src/master/poll.rs @@ -27,13 +27,29 @@ pub(crate) struct Poll { /// Map of all the polls of an association pub(crate) struct PollMap { + timer_state: TimerState, id: u64, polls: BTreeMap, } +#[derive(Copy, Clone, PartialEq)] +enum TimerState { + Deferred, + Running, +} + impl PollMap { pub(crate) fn new() -> Self { Self { + timer_state: TimerState::Running, + id: 0, + polls: BTreeMap::new(), + } + } + + pub(crate) fn new_deferred() -> Self { + Self { + timer_state: TimerState::Deferred, id: 0, polls: BTreeMap::new(), } @@ -42,10 +58,29 @@ impl PollMap { pub(crate) fn add(&mut self, request: ReadRequest, period: Duration) -> u64 { let id = self.id; self.id += 1; - self.polls.insert(id, Poll::new(id, request, period)); + let poll = match self.timer_state { + TimerState::Deferred => Poll::new_deferred(id, request, period), + TimerState::Running => Poll::new(id, request, period), + }; + self.polls.insert(id, poll); id } + /// Start all timers that were deferred during configuration. + /// + /// Returns `true` only when this call transitions the map to running. + pub(crate) fn start(&mut self, now: Instant) -> bool { + if self.timer_state == TimerState::Running { + return false; + } + + self.timer_state = TimerState::Running; + for poll in self.polls.values_mut() { + poll.start(now); + } + true + } + pub(crate) fn remove(&mut self, id: u64) -> bool { self.polls.remove(&id).is_some() } @@ -96,6 +131,19 @@ impl Poll { } } + fn new_deferred(id: u64, request: ReadRequest, period: Duration) -> Self { + Self { + id, + request, + period, + next: None, + } + } + + fn start(&mut self, now: Instant) { + self.next = now.checked_add(self.period); + } + pub(crate) fn format(&self, writer: &mut HeaderWriter) -> Result<(), scursor::WriteError> { self.request.format(writer) } @@ -181,3 +229,45 @@ impl PollHandle { .await } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::master::Classes; + + fn request() -> ReadRequest { + ReadRequest::ClassScan(Classes::all()) + } + + #[test] + fn deferred_polls_are_scheduled_from_start_time() { + let period = Duration::from_secs(5); + let start = Instant::now(); + let mut polls = PollMap::new_deferred(); + polls.add(request(), period); + + assert!(matches!(polls.next(start), Next::None)); + assert!(polls.start(start)); + + match polls.next(start) { + Next::NotBefore(deadline) => assert_eq!(deadline, start + period), + _ => panic!("expected a scheduled poll"), + } + } + + #[test] + fn starting_twice_does_not_reset_poll_deadlines() { + let period = Duration::from_secs(5); + let start = Instant::now(); + let mut polls = PollMap::new_deferred(); + polls.add(request(), period); + + assert!(polls.start(start)); + assert!(!polls.start(start + Duration::from_secs(1))); + + match polls.next(start) { + Next::NotBefore(deadline) => assert_eq!(deadline, start + period), + _ => panic!("expected a scheduled poll"), + } + } +} diff --git a/dnp3/src/master/task.rs b/dnp3/src/master/task.rs index df3c40c7..1d58a708 100644 --- a/dnp3/src/master/task.rs +++ b/dnp3/src/master/task.rs @@ -87,6 +87,10 @@ impl MasterTask { self.session.enabled } + pub(crate) fn start(&mut self, now: Instant) { + self.session.associations.start(now); + } + pub(crate) async fn run(&mut self, io: &mut PhysLayer) -> RunError { let ret = self .session diff --git a/dnp3/src/serial/task.rs b/dnp3/src/serial/task.rs index 5355c21d..aa8b5721 100644 --- a/dnp3/src/serial/task.rs +++ b/dnp3/src/serial/task.rs @@ -1,7 +1,7 @@ use crate::app::{ExponentialBackOff, Listener, RetryStrategy, Shutdown}; use crate::serial::{PortState, SerialSettings}; use crate::util::phys::PhysLayer; -use crate::util::session::{RunError, Session, StopReason}; +use crate::util::session::{Enabled, RunError, Session, StopReason}; pub(crate) struct SerialTask { path: String, @@ -35,8 +35,10 @@ impl SerialTask { async fn run_inner(&mut self) -> Result<(), Shutdown> { loop { - self.listener.update(PortState::Disabled).get().await; - self.session.wait_for_enabled().await?; + if self.session.enabled() == Enabled::No { + self.listener.update(PortState::Disabled).get().await; + self.session.wait_for_enabled().await?; + } if let Err(StopReason::Shutdown) = self.run_enabled().await { return Err(Shutdown); } diff --git a/dnp3/src/tcp/client.rs b/dnp3/src/tcp/client.rs index 522358e2..3566189d 100644 --- a/dnp3/src/tcp/client.rs +++ b/dnp3/src/tcp/client.rs @@ -3,7 +3,7 @@ use crate::tcp::{ ClientConnectionHandler, ClientState, ConnectionInfo, EndpointInner, PostConnectionHandler, }; use crate::util::phys::PhysLayer; -use crate::util::session::{RunError, Session, StopReason}; +use crate::util::session::{Enabled, RunError, Session, StopReason}; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; @@ -38,8 +38,10 @@ impl ClientTask { async fn run_impl(&mut self) -> Result<(), Shutdown> { loop { - self.listener.update(ClientState::Disabled).get().await; - self.session.wait_for_enabled().await?; + if self.session.enabled() == Enabled::No { + self.listener.update(ClientState::Disabled).get().await; + self.session.wait_for_enabled().await?; + } if let Err(StopReason::Shutdown) = self.run_connection().await { return Err(Shutdown); } diff --git a/examples/master/src/main.rs b/examples/master/src/main.rs index 7077e62a..aaec0dde 100644 --- a/examples/master/src/main.rs +++ b/examples/master/src/main.rs @@ -17,8 +17,9 @@ use dnp3::tcp::*; use clap::{Parser, Subcommand}; use dnp3::outstation::FreezeInterval; +use dnp3::udp::spawn_master_udp; use std::net::SocketAddr; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use dnp3_cli_utils::serial::{DataBitsArg, FlowControlArg, ParityArg, StopBitsArg}; use dnp3_cli_utils::LogLevel; @@ -46,122 +47,110 @@ struct CliArgs { #[derive(Debug, Subcommand)] enum TransportCommand { /// Use TCP client transport - TcpClient(TcpClientArgs), + TcpClient { + /// IP address and port to connect to + #[arg(short, long, default_value = "127.0.0.1:20000")] + endpoint: SocketAddr, + + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, + }, /// Use UDP transport - Udp(UdpArgs), - /// Use serial transport - Serial(SerialArgs), - /// Use TLS with CA chain transport - TlsCa(TlsCaArgs), - /// Use TLS with self-signed certificates - TlsSelfSigned(TlsSelfSignedArgs), -} + Udp { + /// Local IP address and port to bind to + #[arg(short, long, default_value = "127.0.0.1:20001")] + local_endpoint: SocketAddr, -#[derive(Debug, Parser)] -struct TcpClientArgs { - /// IP address and port to connect to - #[arg(short, long, default_value = "127.0.0.1:20000")] - endpoint: SocketAddr, + /// Remote IP address and port to send to + #[arg(short, long, default_value = "127.0.0.1:20000")] + remote_endpoint: SocketAddr, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, -} - -#[derive(Debug, Parser)] -struct UdpArgs { - /// Local IP address and port to bind to - #[arg(short, long, default_value = "127.0.0.1:20001")] - local_endpoint: SocketAddr, - - /// Remote IP address and port to send to - #[arg(short, long, default_value = "127.0.0.1:20000")] - remote_endpoint: SocketAddr, + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, + }, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, -} - -#[derive(Debug, Parser)] -struct SerialArgs { - /// Serial port name - #[arg(short, long, default_value = "/dev/ttyS0")] - port: String, - - /// Baud rate - #[arg(short, long, default_value = "9600")] - baud_rate: u32, - - /// Data bits - #[arg(long, value_enum, default_value_t = DataBitsArg::Eight)] - data_bits: DataBitsArg, - - /// Stop bits - #[arg(long, value_enum, default_value_t = StopBitsArg::One)] - stop_bits: StopBitsArg, - - /// Parity - #[arg(long, value_enum, default_value_t = ParityArg::None)] - parity: ParityArg, + /// Use serial transport + Serial { + /// Serial port name + #[arg(short, long, default_value = "/dev/ttyS0")] + port: String, - /// Flow control - #[arg(long, value_enum, default_value_t = FlowControlArg::None)] - flow_control: FlowControlArg, + /// Baud rate + #[arg(short, long, default_value = "9600")] + baud_rate: u32, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, -} + /// Data bits + #[arg(long, value_enum, default_value_t = DataBitsArg::Eight)] + data_bits: DataBitsArg, -#[derive(Debug, Parser)] -struct TlsCaArgs { - /// IP address and port to connect to - #[arg(short, long, default_value = "127.0.0.1:20001")] - endpoint: SocketAddr, + /// Stop bits + #[arg(long, value_enum, default_value_t = StopBitsArg::One)] + stop_bits: StopBitsArg, - /// Domain name to verify - #[arg(long, default_value = "test.com")] - domain: String, + /// Parity + #[arg(long, value_enum, default_value_t = ParityArg::None)] + parity: ParityArg, - /// Path to CA certificate file - #[arg(long, default_value = "./certs/ca_chain/ca_cert.pem")] - ca_cert: PathBuf, + /// Flow control + #[arg(long, value_enum, default_value_t = FlowControlArg::None)] + flow_control: FlowControlArg, - /// Path to entity certificate file - #[arg(long, default_value = "./certs/ca_chain/entity1_cert.pem")] - entity_cert: PathBuf, + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, + }, - /// Path to entity private key file - #[arg(long, default_value = "./certs/ca_chain/entity1_key.pem")] - entity_key: PathBuf, + /// Use TLS with CA chain transport + TlsCa { + /// IP address and port to connect to + #[arg(short, long, default_value = "127.0.0.1:20001")] + endpoint: SocketAddr, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, -} + /// Domain name to verify + #[arg(long, default_value = "test.com")] + domain: String, -#[derive(Debug, Parser)] -struct TlsSelfSignedArgs { - /// IP address and port to connect to - #[arg(short, long, default_value = "127.0.0.1:20001")] - endpoint: SocketAddr, + /// Path to CA certificate file + #[arg(long, default_value = "./certs/ca_chain/ca_cert.pem")] + ca_cert: PathBuf, - /// Path to peer certificate file - #[arg(long, default_value = "./certs/self_signed/entity2_cert.pem")] - peer_cert: PathBuf, + /// Path to entity certificate file + #[arg(long, default_value = "./certs/ca_chain/entity1_cert.pem")] + entity_cert: PathBuf, - /// Path to entity certificate file - #[arg(long, default_value = "./certs/self_signed/entity1_cert.pem")] - entity_cert: PathBuf, + /// Path to entity private key file + #[arg(long, default_value = "./certs/ca_chain/entity1_key.pem")] + entity_key: PathBuf, - /// Path to entity private key file - #[arg(long, default_value = "./certs/self_signed/entity1_key.pem")] - entity_key: PathBuf, + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, + }, - /// Outstation address (DNP3 address of the outstation) - #[arg(short, long, default_value = "1024")] - outstation_address: EndpointAddress, + /// Use TLS with self-signed certificates + TlsSelfSigned { + /// IP address and port to connect to + #[arg(short, long, default_value = "127.0.0.1:20001")] + endpoint: SocketAddr, + + /// Path to peer certificate file + #[arg(long, default_value = "./certs/self_signed/entity2_cert.pem")] + peer_cert: PathBuf, + + /// Path to entity certificate file + #[arg(long, default_value = "./certs/self_signed/entity1_cert.pem")] + entity_cert: PathBuf, + + /// Path to entity private key file + #[arg(long, default_value = "./certs/self_signed/entity1_key.pem")] + entity_key: PathBuf, + + /// Outstation address (DNP3 address of the outstation) + #[arg(short, long, default_value = "1024")] + outstation_address: EndpointAddress, + }, } /// read handler that does nothing @@ -394,9 +383,27 @@ async fn main() -> Result<(), Box> { .init(); // ANCHOR_END: logging - // create the master channel based on the command line argument - let (mut handler, task) = setup(&args)?; - tokio::spawn(task.run()); + // spawn the master channel based on the command line argument + let (mut channel, mut association) = create_channel_and_association(&args).await?; + + // create an event poll + // ANCHOR: add_poll + let poll = association + .add_poll( + ReadRequest::ClassScan(Classes::class123()), + Duration::from_secs(5), + ) + .await?; + // ANCHOR_END: add_poll + + // enable communications + channel.enable().await?; + + let mut handler = CliHandler { + poll, + channel, + association, + }; let mut reader = FramedRead::new(tokio::io::stdin(), LinesCodec::new()); @@ -600,42 +607,120 @@ impl CliHandler { } } -fn setup(cli: &CliArgs) -> Result<(CliHandler, MasterTask), Box> { +// create the specified channel based on the command line argument +async fn create_channel_and_association( + cli: &CliArgs, +) -> Result<(MasterChannel, AssociationHandle), Box> { match &cli.transport { - TransportCommand::TcpClient(args) => args.setup(cli.master_address), - TransportCommand::Udp(args) => args.setup(cli.master_address), - TransportCommand::Serial(args) => args.setup(cli.master_address), - TransportCommand::TlsCa(args) => args.setup(cli.master_address), - TransportCommand::TlsSelfSigned(args) => args.setup(cli.master_address), + TransportCommand::TcpClient { + endpoint, + outstation_address, + } => { + let mut channel = create_tcp_channel(cli.master_address, *endpoint)?; + let assoc = add_association(&mut channel, *outstation_address).await?; + Ok((channel, assoc)) + } + TransportCommand::Udp { + local_endpoint, + remote_endpoint, + outstation_address, + } => { + let mut channel = create_udp_channel(cli.master_address, *local_endpoint)?; + let assoc = + add_udp_association(&mut channel, *remote_endpoint, *outstation_address).await?; + Ok((channel, assoc)) + } + TransportCommand::Serial { + port, + baud_rate, + data_bits, + stop_bits, + parity, + flow_control, + outstation_address, + } => { + let mut channel = create_serial_channel( + cli.master_address, + port, + *baud_rate, + *data_bits, + *stop_bits, + *parity, + *flow_control, + )?; + let assoc = add_association(&mut channel, *outstation_address).await?; + Ok((channel, assoc)) + } + TransportCommand::TlsCa { + endpoint, + domain, + ca_cert, + entity_cert, + entity_key, + outstation_address, + } => { + let mut channel = create_tls_channel( + cli.master_address, + *endpoint, + get_tls_authority_config(domain, ca_cert, entity_cert, entity_key)?, + )?; + let assoc = add_association(&mut channel, *outstation_address).await?; + Ok((channel, assoc)) + } + TransportCommand::TlsSelfSigned { + endpoint, + peer_cert, + entity_cert, + entity_key, + outstation_address, + } => { + let mut channel = create_tls_channel( + cli.master_address, + *endpoint, + get_tls_self_signed_config(peer_cert, entity_cert, entity_key)?, + )?; + let assoc = add_association(&mut channel, *outstation_address).await?; + Ok((channel, assoc)) + } } } -fn create_stream_builder( - master_address: EndpointAddress, +async fn add_association( + channel: &mut MasterChannel, outstation_address: EndpointAddress, -) -> Result<(MasterBuilder, CliHandler), Box> { - let (mut builder, channel) = MasterBuilder::new(get_master_channel_config(master_address)?); - builder.enable(); - let mut assoc = builder.new_association( - outstation_address, - get_association_config(), - ExampleReadHandler::boxed(), - Box::new(ExampleAssociationHandler), - Box::new(ExampleAssociationInformation), - ); - let poll = assoc.add_poll( - ReadRequest::ClassScan(Classes::class123()), - Duration::from_secs(5), - ); - let association = builder.add_association(assoc)?; - Ok(( - builder, - CliHandler { - poll, - channel, - association, - }, - )) +) -> Result> { + // ANCHOR: association_create + let association = channel + .add_association( + outstation_address, + get_association_config(), + ExampleReadHandler::boxed(), + Box::new(ExampleAssociationHandler), + Box::new(ExampleAssociationInformation), + ) + .await?; + // ANCHOR_END: association_create + Ok(association) +} + +async fn add_udp_association( + channel: &mut MasterChannel, + remote_endpoint: SocketAddr, + outstation_address: EndpointAddress, +) -> Result> { + // ANCHOR: association_create_udp + let association = channel + .add_udp_association( + outstation_address, + remote_endpoint, + get_association_config(), + ExampleReadHandler::boxed(), + Box::new(ExampleAssociationHandler), + Box::new(ExampleAssociationInformation), + ) + .await?; + // ANCHOR_END: association_create_udp + Ok(association) } // ANCHOR: master_channel_config @@ -666,136 +751,118 @@ fn get_association_config() -> AssociationConfig { } // ANCHOR_END: association_config -impl TcpClientArgs { - fn setup( - &self, - master_address: EndpointAddress, - ) -> Result<(CliHandler, MasterTask), Box> { - let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; - let connect = EndpointList::new(self.endpoint.to_string(), &[]) - .into_connect_handler(ConnectStrategy::default()); - Ok(( - handler, - builder.into_tcp(LinkErrorMode::Close, connect, NullListener::create()), - )) - } +fn get_tls_self_signed_config( + peer_cert: &Path, + entity_cert: &Path, + entity_key: &Path, +) -> Result> { + // ANCHOR: tls_self_signed_config + let config = TlsClientConfig::self_signed( + peer_cert, + entity_cert, + entity_key, + None, // no password + MinTlsVersion::V12, + )?; + // ANCHOR_END: tls_self_signed_config + Ok(config) } -impl UdpArgs { - fn setup( - &self, - master_address: EndpointAddress, - ) -> Result<(CliHandler, MasterTask), Box> { - let (mut builder, channel) = - UdpMasterBuilder::new(get_master_channel_config(master_address)?); - builder.enable(); - let mut assoc = builder.new_association( - self.outstation_address, - self.remote_endpoint, - get_association_config(), - ExampleReadHandler::boxed(), - Box::new(ExampleAssociationHandler), - Box::new(ExampleAssociationInformation), - ); - let poll = assoc.add_poll( - ReadRequest::ClassScan(Classes::class123()), - Duration::from_secs(5), - ); - let association = builder.add_association(assoc)?; - let task = builder.into_udp( - self.local_endpoint, - LinkReadMode::Datagram, - Timeout::from_secs(5)?, - ); - Ok(( - CliHandler { - poll, - channel, - association, - }, - task, - )) - } +fn get_tls_authority_config( + domain: &str, + ca_cert: &Path, + entity_cert: &Path, + entity_key: &Path, +) -> Result> { + // ANCHOR: tls_ca_chain_config + let config = TlsClientConfig::full_pki( + Some(domain.to_string()), + ca_cert, + entity_cert, + entity_key, + None, // no password + MinTlsVersion::V12, + )?; + // ANCHOR_END: tls_ca_chain_config + Ok(config) } -impl SerialArgs { - fn setup( - &self, - master_address: EndpointAddress, - ) -> Result<(CliHandler, MasterTask), Box> { - let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; - let settings = SerialSettings { - baud_rate: self.baud_rate, - data_bits: self.data_bits.into(), - stop_bits: self.stop_bits.into(), - parity: self.parity.into(), - flow_control: self.flow_control.into(), - }; - Ok(( - handler, - builder.into_serial( - &self.port, - settings, - Duration::from_secs(1), - NullListener::create(), - ), - )) - } +fn create_tcp_channel( + master_address: EndpointAddress, + endpoint: SocketAddr, +) -> Result> { + // ANCHOR: create_master_tcp_channel + let channel = spawn_master_tcp_client( + LinkErrorMode::Close, + get_master_channel_config(master_address)?, + EndpointList::new(endpoint.to_string(), &[]), + ConnectStrategy::default(), + NullListener::create(), + ); + // ANCHOR_END: create_master_tcp_channel + Ok(channel) } -impl TlsCaArgs { - fn setup( - &self, - master_address: EndpointAddress, - ) -> Result<(CliHandler, MasterTask), Box> { - let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; - let connect = EndpointList::new(self.endpoint.to_string(), &[]) - .into_connect_handler(ConnectStrategy::default()); - let tls_config = TlsClientConfig::full_pki( - Some(self.domain.to_string()), - &self.ca_cert, - &self.entity_cert, - &self.entity_key, - None, - MinTlsVersion::V12, - )?; - Ok(( - handler, - builder.into_tls( - LinkErrorMode::Close, - connect, - NullListener::create(), - tls_config, - ), - )) - } +fn create_udp_channel( + master_address: EndpointAddress, + local_endpoint: SocketAddr, +) -> Result> { + // ANCHOR: create_master_udp_channel + let channel = spawn_master_udp( + local_endpoint, + LinkReadMode::Datagram, + Timeout::from_secs(5)?, + get_master_channel_config(master_address)?, + ); + // ANCHOR_END: create_master_udp_channel + Ok(channel) } -impl TlsSelfSignedArgs { - fn setup( - &self, - master_address: EndpointAddress, - ) -> Result<(CliHandler, MasterTask), Box> { - let (builder, handler) = create_stream_builder(master_address, self.outstation_address)?; - let connect = EndpointList::new(self.endpoint.to_string(), &[]) - .into_connect_handler(ConnectStrategy::default()); - let tls_config = TlsClientConfig::self_signed( - &self.peer_cert, - &self.entity_cert, - &self.entity_key, - None, - MinTlsVersion::V12, - )?; - Ok(( - handler, - builder.into_tls( - LinkErrorMode::Close, - connect, - NullListener::create(), - tls_config, - ), - )) - } +fn create_serial_channel( + master_address: EndpointAddress, + port: &str, + baud_rate: u32, + data_bits: DataBitsArg, + stop_bits: StopBitsArg, + parity: ParityArg, + flow_control: FlowControlArg, +) -> Result> { + // ANCHOR: create_master_serial_channel + let settings = SerialSettings { + baud_rate, + data_bits: data_bits.into(), + stop_bits: stop_bits.into(), + parity: parity.into(), + flow_control: flow_control.into(), + }; + + let channel = spawn_master_serial( + get_master_channel_config(master_address)?, + port, + settings, + Duration::from_secs(1), + NullListener::create(), + ); + // ANCHOR_END: create_master_serial_channel + Ok(channel) +} + +fn create_tls_channel( + master_address: EndpointAddress, + endpoint: SocketAddr, + tls_config: TlsClientConfig, +) -> Result> { + // ANCHOR: create_master_tls_channel + let channel = spawn_master_tls_client( + LinkErrorMode::Close, + get_master_channel_config(master_address)?, + EndpointList::new(endpoint.to_string(), &[]), + ConnectStrategy::default(), + NullListener::create(), + tls_config, + ); + // ANCHOR_END: create_master_tls_channel + Ok(channel) } fn print_file_info(info: FileInfo) { From d74840e94266d9c7bdc22a18e7e6bb9d99dc4212 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 13:32:50 -0700 Subject: [PATCH 05/14] Gate new master builder APIs as unstable --- dnp3/Cargo.toml | 1 + dnp3/src/master/association.rs | 7 ++++++- dnp3/src/master/builder.rs | 3 +-- dnp3/src/master/mod.rs | 2 ++ dnp3/src/master/poll.rs | 12 +++++++++++- dnp3/src/master/task.rs | 1 + 6 files changed, 22 insertions(+), 4 deletions(-) diff --git a/dnp3/Cargo.toml b/dnp3/Cargo.toml index 55ac8c07..e458563c 100644 --- a/dnp3/Cargo.toml +++ b/dnp3/Cargo.toml @@ -46,6 +46,7 @@ tokio = { version = "1", features = ["test-util"] } [features] default = ["tls", "serial"] +unstable = [] enable-tls = [] # not enabled directly ffi = [] # this feature flag is only used when building the FFI tls = ["enable-tls", "sfio-rustls-config/crypto-ring", "tokio-rustls"] diff --git a/dnp3/src/master/association.rs b/dnp3/src/master/association.rs index 4826236d..e8996f39 100644 --- a/dnp3/src/master/association.rs +++ b/dnp3/src/master/association.rs @@ -15,6 +15,7 @@ use crate::master::extract::extract_measurements; use crate::master::handler::AssociationHandler; use crate::master::messages::AssociationMsgType; use crate::master::poll::{PollHandle, PollMap, PollMsg}; +#[cfg(feature = "unstable")] use crate::master::request::ReadRequest; use crate::master::request::{Classes, EventClasses, TimeSyncProcedure}; use crate::master::tasks::auto::AutoTask; @@ -350,6 +351,7 @@ impl Association { } } + #[cfg(feature = "unstable")] pub(crate) fn new_deferred( address: FragmentAddr, config: AssociationConfig, @@ -376,6 +378,7 @@ impl Association { } } + #[cfg(feature = "unstable")] pub(crate) fn start(&mut self, now: Instant) { if self.polls.start(now) { self.next_link_status_deadline = @@ -383,6 +386,7 @@ impl Association { } } + #[cfg(feature = "unstable")] pub(crate) fn add_poll(&mut self, request: ReadRequest, period: Duration) -> u64 { self.polls.add(request, period) } @@ -789,6 +793,7 @@ impl AssociationMap { } } + #[cfg(feature = "unstable")] pub(crate) fn start(&mut self, now: Instant) { for association in self.map.values_mut() { association.start(now); @@ -890,7 +895,7 @@ impl AssociationMap { } } -#[cfg(test)] +#[cfg(all(test, feature = "unstable"))] mod timer_tests { use super::*; use crate::master::{AssociationHandler, AssociationInformation, ReadHandler}; diff --git a/dnp3/src/master/builder.rs b/dnp3/src/master/builder.rs index 04e0ed8c..ab217693 100644 --- a/dnp3/src/master/builder.rs +++ b/dnp3/src/master/builder.rs @@ -70,8 +70,7 @@ pub struct AssociationBuilder<'a> { /// A master task bound to a transport. Call [`run`](Self::run) to execute /// the event loop. /// -/// Created by [`MasterBuilder::into_tcp`], [`MasterBuilder::into_tls`], -/// [`MasterBuilder::into_serial`], or [`UdpMasterBuilder::into_udp`]. +/// Created by binding a [`MasterBuilder`] or [`UdpMasterBuilder`] to a transport. #[must_use = "a MasterTask does nothing unless you call .run()"] pub struct MasterTask { inner: MasterTaskType, diff --git a/dnp3/src/master/mod.rs b/dnp3/src/master/mod.rs index 53ed3606..61fb560a 100644 --- a/dnp3/src/master/mod.rs +++ b/dnp3/src/master/mod.rs @@ -1,4 +1,5 @@ pub use association::*; +#[cfg(feature = "unstable")] pub use builder::*; pub use error::*; pub use file::*; @@ -8,6 +9,7 @@ pub use read_handler::*; pub use request::*; mod association; +#[cfg(feature = "unstable")] mod builder; mod error; mod file; diff --git a/dnp3/src/master/poll.rs b/dnp3/src/master/poll.rs index 440cf464..0c4dd97e 100644 --- a/dnp3/src/master/poll.rs +++ b/dnp3/src/master/poll.rs @@ -27,11 +27,13 @@ pub(crate) struct Poll { /// Map of all the polls of an association pub(crate) struct PollMap { + #[cfg(feature = "unstable")] timer_state: TimerState, id: u64, polls: BTreeMap, } +#[cfg(feature = "unstable")] #[derive(Copy, Clone, PartialEq)] enum TimerState { Deferred, @@ -41,12 +43,14 @@ enum TimerState { impl PollMap { pub(crate) fn new() -> Self { Self { + #[cfg(feature = "unstable")] timer_state: TimerState::Running, id: 0, polls: BTreeMap::new(), } } + #[cfg(feature = "unstable")] pub(crate) fn new_deferred() -> Self { Self { timer_state: TimerState::Deferred, @@ -58,10 +62,13 @@ impl PollMap { pub(crate) fn add(&mut self, request: ReadRequest, period: Duration) -> u64 { let id = self.id; self.id += 1; + #[cfg(feature = "unstable")] let poll = match self.timer_state { TimerState::Deferred => Poll::new_deferred(id, request, period), TimerState::Running => Poll::new(id, request, period), }; + #[cfg(not(feature = "unstable"))] + let poll = Poll::new(id, request, period); self.polls.insert(id, poll); id } @@ -69,6 +76,7 @@ impl PollMap { /// Start all timers that were deferred during configuration. /// /// Returns `true` only when this call transitions the map to running. + #[cfg(feature = "unstable")] pub(crate) fn start(&mut self, now: Instant) -> bool { if self.timer_state == TimerState::Running { return false; @@ -131,6 +139,7 @@ impl Poll { } } + #[cfg(feature = "unstable")] fn new_deferred(id: u64, request: ReadRequest, period: Duration) -> Self { Self { id, @@ -140,6 +149,7 @@ impl Poll { } } + #[cfg(feature = "unstable")] fn start(&mut self, now: Instant) { self.next = now.checked_add(self.period); } @@ -230,7 +240,7 @@ impl PollHandle { } } -#[cfg(test)] +#[cfg(all(test, feature = "unstable"))] mod tests { use super::*; use crate::master::Classes; diff --git a/dnp3/src/master/task.rs b/dnp3/src/master/task.rs index 1d58a708..8c274a03 100644 --- a/dnp3/src/master/task.rs +++ b/dnp3/src/master/task.rs @@ -87,6 +87,7 @@ impl MasterTask { self.session.enabled } + #[cfg(feature = "unstable")] pub(crate) fn start(&mut self, now: Instant) { self.session.associations.start(now); } From 3f55e26ae441cf25c6d937bb576336e3d52421d6 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 14:11:45 -0700 Subject: [PATCH 06/14] Document unstable feature policy --- dnp3/Cargo.toml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/dnp3/Cargo.toml b/dnp3/Cargo.toml index e458563c..a29b85a6 100644 --- a/dnp3/Cargo.toml +++ b/dnp3/Cargo.toml @@ -46,6 +46,8 @@ tokio = { version = "1", features = ["test-util"] } [features] default = ["tls", "serial"] +# APIs behind this feature are exempt from semantic versioning guarantees and +# may change or be removed in any release, including patch releases. unstable = [] enable-tls = [] # not enabled directly ffi = [] # this feature flag is only used when building the FFI From 5ff94440bc3919a3b59192533288d7ec0b56733c Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 14:22:52 -0700 Subject: [PATCH 07/14] Add non-spawning TCP server task API --- .github/workflows/ci.yml | 3 ++- dnp3/src/tcp/outstation/server.rs | 44 ++++++++++++++++++++++++++++--- 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4ad2ef66..0db2508e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,6 +36,7 @@ jobs: - "--no-default-features --features serial" - "--no-default-features --features tls" - "--no-default-features --features tls-aws-lc" + - "--no-default-features --features unstable" runs-on: ubuntu-latest steps: - name: Checkout @@ -65,7 +66,7 @@ jobs: with: toolchain: ${{ matrix.rust }} - name: Run Rust unit tests - run: cargo test + run: cargo test --features unstable # Build API documentation packages documentation: runs-on: ubuntu-latest diff --git a/dnp3/src/tcp/outstation/server.rs b/dnp3/src/tcp/outstation/server.rs index 76346f84..e9c5b8ff 100644 --- a/dnp3/src/tcp/outstation/server.rs +++ b/dnp3/src/tcp/outstation/server.rs @@ -158,16 +158,33 @@ impl Server { /// /// This may be called outside the Tokio runtime and allows for manual spawning pub async fn bind_no_spawn( - mut self, + self, ) -> Result<(ServerHandle, impl std::future::Future), tokio::io::Error> { let listener = tokio::net::TcpListener::bind(self.address).await?; + Ok(self.create_task(listener)) + } + /// Consume this server and create a task using an already-bound TCP listener. + /// + /// The returned future does nothing until polled and may be run on any Tokio runtime. + #[cfg(feature = "unstable")] + pub fn into_task( + self, + listener: tokio::net::TcpListener, + ) -> (ServerHandle, impl std::future::Future) { + self.create_task(listener) + } + + fn create_task( + mut self, + listener: tokio::net::TcpListener, + ) -> (ServerHandle, impl std::future::Future) { let addr = listener.local_addr().ok(); + let local = addr.unwrap_or(self.address); let (token, shutdown_rx) = crate::util::shutdown::shutdown_token(); let task = async move { - let local = self.address; self.run(listener, shutdown_rx) .instrument(tracing::info_span!("tcp-server", "listen" = ?local)) .await @@ -178,7 +195,7 @@ impl Server { _token: token, }; - Ok((handle, task)) + (handle, task) } /// Consume the `TcpServer` builder object, bind it to pre-specified port, and spawn the server @@ -267,3 +284,24 @@ impl Server { } } } + +#[cfg(all(test, feature = "unstable"))] +mod tests { + use super::*; + + #[tokio::test] + async fn into_task_uses_supplied_listener_and_stops_with_handle() { + let configured_address = "127.0.0.1:0".parse().unwrap(); + let listener = tokio::net::TcpListener::bind(configured_address) + .await + .unwrap(); + let listener_address = listener.local_addr().unwrap(); + let server = Server::new_tcp_server(LinkErrorMode::Close, configured_address); + + let (handle, task) = server.into_task(listener); + + assert_eq!(handle.local_addr(), Some(listener_address)); + drop(handle); + task.await; + } +} From 426e04a8ca6f75d795d0763b194d74ae9253c484 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 14:41:14 -0700 Subject: [PATCH 08/14] Add non-spawning serial outstation task API --- dnp3/src/serial/outstation.rs | 167 ++++++++++++++++++++++++++++------ 1 file changed, 137 insertions(+), 30 deletions(-) diff --git a/dnp3/src/serial/outstation.rs b/dnp3/src/serial/outstation.rs index 5012551b..f365d4e6 100644 --- a/dnp3/src/serial/outstation.rs +++ b/dnp3/src/serial/outstation.rs @@ -12,6 +12,96 @@ use crate::util::phys::{PhysAddr, PhysLayer}; use crate::util::session::{Enabled, Session}; use tracing::Instrument; +/// A fully configured serial outstation that has not yet opened its serial port. +#[cfg(feature = "unstable")] +pub struct SerialOutstation { + task: OutstationTask, +} + +/// A serial outstation task that owns an open and configured serial port. +#[cfg(feature = "unstable")] +#[must_use = "a SerialOutstationTask does nothing unless you call .run()"] +pub struct SerialOutstationTask { + inner: OneShotSerialOutstationTask, +} + +struct OneShotSerialOutstationTask { + path: String, + serial: tokio_serial::SerialStream, + task: OutstationTask, +} + +#[cfg(feature = "unstable")] +impl SerialOutstation { + /// Create a fully configured serial outstation without opening a serial port. + pub fn new( + config: OutstationConfig, + application: Box, + information: Box, + control_handler: Box, + ) -> (Self, OutstationHandle) { + let (task, handle) = create_outstation(config, application, information, control_handler); + (Self { task }, handle) + } + + /// Open and configure the serial port, producing a task that is ready to run. + pub async fn open( + self, + path: &str, + settings: SerialSettings, + ) -> std::io::Result { + let serial = crate::serial::open(path, settings)?; + Ok(SerialOutstationTask { + inner: OneShotSerialOutstationTask::new(path, serial, self.task), + }) + } +} + +#[cfg(feature = "unstable")] +impl SerialOutstationTask { + /// Run the outstation until it is shut down or the serial port fails. + pub async fn run(self) { + self.inner.run().await; + } +} + +impl OneShotSerialOutstationTask { + fn new(path: &str, serial: tokio_serial::SerialStream, task: OutstationTask) -> Self { + Self { + path: path.to_owned(), + serial, + task, + } + } + + async fn run(mut self) { + let mut io = PhysLayer::Serial(self.serial); + let _ = self + .task + .run(&mut io) + .instrument(tracing::info_span!("dnp3-outstation-serial", "port" = ?self.path)) + .await; + } +} + +fn create_outstation( + config: OutstationConfig, + application: Box, + information: Box, + control_handler: Box, +) -> (OutstationTask, OutstationHandle) { + OutstationTask::create( + Enabled::Yes, + LinkModes::serial(), + ParseOptions::get_static(), + config, + PhysAddr::None, + application, + information, + control_handler, + ) +} + /// Spawn an outstation task onto the `Tokio` runtime. The task runs until the returned handle is dropped or /// a serial port error occurs, e.g. a serial port is removed from the OS. It attempts to open /// the serial port immediately, and fails if it cannot. @@ -30,26 +120,8 @@ pub fn spawn_outstation_serial( control_handler: Box, ) -> std::io::Result { let serial = crate::serial::open(path, settings)?; - let (mut task, handle) = OutstationTask::create( - Enabled::Yes, - LinkModes::serial(), - ParseOptions::get_static(), - config, - PhysAddr::None, - application, - information, - control_handler, - ); - - let log_path = path.to_owned(); - let future = async move { - let mut io = PhysLayer::Serial(serial); - let _ = task - .run(&mut io) - .instrument(tracing::info_span!("dnp3-outstation-serial", "port" = ?log_path)) - .await; - }; - tokio::spawn(future); + let (task, handle) = create_outstation(config, application, information, control_handler); + tokio::spawn(OneShotSerialOutstationTask::new(path, serial, task).run()); Ok(handle) } @@ -82,16 +154,7 @@ pub fn spawn_outstation_serial_2( control_handler: Box, listener: Box>, ) -> OutstationHandle { - let (task, handle) = OutstationTask::create( - Enabled::Yes, - LinkModes::serial(), - ParseOptions::get_static(), - config, - PhysAddr::None, - application, - information, - control_handler, - ); + let (task, handle) = create_outstation(config, application, information, control_handler); let mut serial = SerialTask::new(path, settings, Session::outstation(task), retry, listener); @@ -106,6 +169,50 @@ pub fn spawn_outstation_serial_2( handle } +#[cfg(all(test, feature = "unstable"))] +mod tests { + use super::*; + use crate::link::EndpointAddress; + use crate::outstation::database::EventBufferConfig; + + struct NullApplication; + impl OutstationApplication for NullApplication {} + + struct NullInformation; + impl OutstationInformation for NullInformation {} + + fn create() -> (SerialOutstation, OutstationHandle) { + let config = OutstationConfig::new( + EndpointAddress::try_new(10).unwrap(), + EndpointAddress::try_new(1).unwrap(), + EventBufferConfig::all_types(0), + ); + SerialOutstation::new( + config, + Box::new(NullApplication), + Box::new(NullInformation), + crate::outstation::DefaultControlHandler::create(), + ) + } + + #[test] + fn construction_does_not_require_a_runtime() { + let (_outstation, _handle) = create(); + } + + #[tokio::test] + async fn failed_open_does_not_produce_a_task() { + let (outstation, _handle) = create(); + let result = outstation + .open( + "/path/that/does/not/exist/dnp3-test", + SerialSettings::default(), + ) + .await; + assert!(result.is_err()); + } +} + /// This function was added post 1.0 to provide fault tolerance for outstation serial ports. /// /// This function is implemented by calling [`spawn_outstation_serial_2`] with a post listener that does nothing. From e0723f1a0d52b987ad98fb11f718018deaef0ab8 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 16:52:15 -0700 Subject: [PATCH 09/14] Leave tracing spans to non-spawning callers --- dnp3/src/master/builder.rs | 26 +++++--------------------- dnp3/src/serial/outstation.rs | 24 ++++++++++-------------- dnp3/src/tcp/outstation/server.rs | 26 ++++++++++---------------- 3 files changed, 25 insertions(+), 51 deletions(-) diff --git a/dnp3/src/master/builder.rs b/dnp3/src/master/builder.rs index ab217693..93a29c15 100644 --- a/dnp3/src/master/builder.rs +++ b/dnp3/src/master/builder.rs @@ -1,9 +1,6 @@ use std::net::SocketAddr; use std::time::Duration; -use tokio::time::Instant; -use tracing::Instrument; - use super::association::{Association, AssociationMap}; use super::handler::{AssociationHandle, MasterChannel, MasterChannelConfig, MasterChannelType}; use super::poll::PollHandle; @@ -24,6 +21,7 @@ use crate::udp::task::UdpTask; use crate::util::channel::Receiver; use crate::util::phys::PhysAddr; use crate::util::session::{Enabled, Session}; +use tokio::time::Instant; /// Builder for configuring a stream-based master task (TCP, TLS, serial) /// before binding it to a transport. @@ -416,7 +414,6 @@ impl MasterTaskType { impl TcpTask { async fn run(self) { - let name = self.connect_handler.endpoint_span_name(); let session = Session::master(self.inner); let mut client = ClientTask::new( session, @@ -424,17 +421,13 @@ impl TcpTask { PostConnectionHandler::Tcp, self.listener, ); - client - .run() - .instrument(tracing::info_span!("dnp3-master-tcp-client", "endpoint" = ?name)) - .await; + client.run().await; } } #[cfg(feature = "enable-tls")] impl TlsTask { async fn run(self) { - let name = self.connect_handler.endpoint_span_name(); let session = Session::master(self.inner); let mut client = ClientTask::new( session, @@ -442,10 +435,7 @@ impl TlsTask { PostConnectionHandler::Tls(self.tls_config), self.listener, ); - client - .run() - .instrument(tracing::info_span!("dnp3-master-tls-client", "endpoint" = ?name)) - .await; + client.run().await; } } @@ -460,10 +450,7 @@ impl SerialTask { self.retry_strategy, self.listener, ); - serial - .run() - .instrument(tracing::info_span!("dnp3-master-serial", "port" = ?self.path)) - .await; + serial.run().await; } } @@ -476,10 +463,7 @@ impl UdpMasterTask { factory: UdpFactory::bound(local_endpoint), retry_delay: self.retry_delay, }; - let _ = task - .run() - .instrument(tracing::info_span!("dnp3-master-udp", "endpoint" = ?local_endpoint)) - .await; + let _ = task.run().await; } } diff --git a/dnp3/src/serial/outstation.rs b/dnp3/src/serial/outstation.rs index f365d4e6..693c5834 100644 --- a/dnp3/src/serial/outstation.rs +++ b/dnp3/src/serial/outstation.rs @@ -26,7 +26,6 @@ pub struct SerialOutstationTask { } struct OneShotSerialOutstationTask { - path: String, serial: tokio_serial::SerialStream, task: OutstationTask, } @@ -52,7 +51,7 @@ impl SerialOutstation { ) -> std::io::Result { let serial = crate::serial::open(path, settings)?; Ok(SerialOutstationTask { - inner: OneShotSerialOutstationTask::new(path, serial, self.task), + inner: OneShotSerialOutstationTask::new(serial, self.task), }) } } @@ -66,21 +65,13 @@ impl SerialOutstationTask { } impl OneShotSerialOutstationTask { - fn new(path: &str, serial: tokio_serial::SerialStream, task: OutstationTask) -> Self { - Self { - path: path.to_owned(), - serial, - task, - } + fn new(serial: tokio_serial::SerialStream, task: OutstationTask) -> Self { + Self { serial, task } } async fn run(mut self) { let mut io = PhysLayer::Serial(self.serial); - let _ = self - .task - .run(&mut io) - .instrument(tracing::info_span!("dnp3-outstation-serial", "port" = ?self.path)) - .await; + let _ = self.task.run(&mut io).await; } } @@ -121,7 +112,12 @@ pub fn spawn_outstation_serial( ) -> std::io::Result { let serial = crate::serial::open(path, settings)?; let (task, handle) = create_outstation(config, application, information, control_handler); - tokio::spawn(OneShotSerialOutstationTask::new(path, serial, task).run()); + let log_path = path.to_owned(); + tokio::spawn( + OneShotSerialOutstationTask::new(serial, task) + .run() + .instrument(tracing::info_span!("dnp3-outstation-serial", "port" = ?log_path)), + ); Ok(handle) } diff --git a/dnp3/src/tcp/outstation/server.rs b/dnp3/src/tcp/outstation/server.rs index e9c5b8ff..34fe1104 100644 --- a/dnp3/src/tcp/outstation/server.rs +++ b/dnp3/src/tcp/outstation/server.rs @@ -117,14 +117,8 @@ impl Server { }; self.outstations.push(outstation); - let endpoint = self.address; - let address = config.outstation_address.raw_value(); let future = async move { - let _ = adapter.run() - .instrument( - tracing::info_span!("dnp3-outstation-tcp", "listen" = ?endpoint, "addr" = address), - ) - .await; + let _ = adapter.run().await; }; Ok((handle, future)) } @@ -141,6 +135,8 @@ impl Server { listener: Box>, filter: AddressFilter, ) -> Result { + let endpoint = self.address; + let address = config.outstation_address.raw_value(); let (handle, future) = self.add_outstation_no_spawn( config, application, @@ -149,7 +145,9 @@ impl Server { listener, filter, )?; - tokio::spawn(future); + tokio::spawn(future.instrument( + tracing::info_span!("dnp3-outstation-tcp", "listen" = ?endpoint, "addr" = address), + )); Ok(handle) } @@ -180,15 +178,9 @@ impl Server { listener: tokio::net::TcpListener, ) -> (ServerHandle, impl std::future::Future) { let addr = listener.local_addr().ok(); - let local = addr.unwrap_or(self.address); - let (token, shutdown_rx) = crate::util::shutdown::shutdown_token(); - let task = async move { - self.run(listener, shutdown_rx) - .instrument(tracing::info_span!("tcp-server", "listen" = ?local)) - .await - }; + let task = async move { self.run(listener, shutdown_rx).await }; let handle = ServerHandle { addr, @@ -205,8 +197,10 @@ impl Server { /// /// This must be called from within the Tokio runtime pub async fn bind(self) -> Result { + let configured_address = self.address; let (handle, future) = self.bind_no_spawn().await?; - tokio::spawn(future); + let local = handle.local_addr().unwrap_or(configured_address); + tokio::spawn(future.instrument(tracing::info_span!("tcp-server", "listen" = ?local))); Ok(handle) } From 855b335b14b13fdbe3a9925ab6ed7cdd133250ad Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 17:27:37 -0700 Subject: [PATCH 10/14] Make serial outstation open synchronous --- dnp3/src/serial/outstation.rs | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/dnp3/src/serial/outstation.rs b/dnp3/src/serial/outstation.rs index 693c5834..dcaed790 100644 --- a/dnp3/src/serial/outstation.rs +++ b/dnp3/src/serial/outstation.rs @@ -44,7 +44,7 @@ impl SerialOutstation { } /// Open and configure the serial port, producing a task that is ready to run. - pub async fn open( + pub fn open( self, path: &str, settings: SerialSettings, @@ -196,15 +196,13 @@ mod tests { let (_outstation, _handle) = create(); } - #[tokio::test] - async fn failed_open_does_not_produce_a_task() { + #[test] + fn failed_open_does_not_produce_a_task() { let (outstation, _handle) = create(); - let result = outstation - .open( - "/path/that/does/not/exist/dnp3-test", - SerialSettings::default(), - ) - .await; + let result = outstation.open( + "/path/that/does/not/exist/dnp3-test", + SerialSettings::default(), + ); assert!(result.is_err()); } } From 55c1b4f46200a616ee38cdb126cbc3b89eec04e6 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Sun, 19 Jul 2026 19:47:13 -0700 Subject: [PATCH 11/14] Add concrete no-spawn outstation tasks --- dnp3/src/outstation/mod.rs | 4 + dnp3/src/outstation/runner.rs | 41 ++++++++ dnp3/src/serial/mod.rs | 2 +- dnp3/src/serial/outstation.rs | 39 +++----- dnp3/src/tcp/outstation/server.rs | 159 ++++++++++++++++++++++++++---- 5 files changed, 197 insertions(+), 48 deletions(-) create mode 100644 dnp3/src/outstation/runner.rs diff --git a/dnp3/src/outstation/mod.rs b/dnp3/src/outstation/mod.rs index 455c98a4..32d86336 100644 --- a/dnp3/src/outstation/mod.rs +++ b/dnp3/src/outstation/mod.rs @@ -1,4 +1,6 @@ pub use config::*; +#[cfg(feature = "unstable")] +pub use runner::OutstationTask; pub use traits::*; use crate::app::Shutdown; @@ -15,6 +17,8 @@ mod config; pub(crate) mod control; /// handling of deferred read requests pub(crate) mod deferred; +#[cfg(feature = "unstable")] +mod runner; /// outstation session pub(crate) mod session; /// async outstation task diff --git a/dnp3/src/outstation/runner.rs b/dnp3/src/outstation/runner.rs new file mode 100644 index 00000000..94a2c37e --- /dev/null +++ b/dnp3/src/outstation/runner.rs @@ -0,0 +1,41 @@ +/// A fully configured DNP3 outstation that is ready to run. +/// +/// This task represents one outstation endpoint. It is the caller's responsibility to run it on +/// a Tokio runtime. No tracing span is attached automatically, so the caller may instrument the +/// [`run`](Self::run) future as desired. +#[must_use = "an OutstationTask does nothing unless you call .run()"] +pub struct OutstationTask { + inner: OutstationTaskType, +} + +enum OutstationTaskType { + #[cfg(feature = "serial")] + Serial(crate::serial::outstation::OneShotSerialOutstationTask), + Tcp(crate::tcp::server_task::ServerTask), +} + +impl OutstationTask { + #[cfg(feature = "serial")] + pub(crate) fn serial(task: crate::serial::outstation::OneShotSerialOutstationTask) -> Self { + Self { + inner: OutstationTaskType::Serial(task), + } + } + + pub(crate) fn tcp(task: crate::tcp::server_task::ServerTask) -> Self { + Self { + inner: OutstationTaskType::Tcp(task), + } + } + + /// Run the outstation until it is shut down or the transport fails. + pub async fn run(self) { + match self.inner { + #[cfg(feature = "serial")] + OutstationTaskType::Serial(task) => task.run().await, + OutstationTaskType::Tcp(mut task) => { + let _ = task.run().await; + } + } + } +} diff --git a/dnp3/src/serial/mod.rs b/dnp3/src/serial/mod.rs index 5167cbc5..b1a7c1ef 100644 --- a/dnp3/src/serial/mod.rs +++ b/dnp3/src/serial/mod.rs @@ -51,7 +51,7 @@ pub use master::*; pub use outstation::*; mod master; -mod outstation; +pub(crate) mod outstation; pub(crate) mod task; /// State of the serial port diff --git a/dnp3/src/serial/outstation.rs b/dnp3/src/serial/outstation.rs index dcaed790..5cbddf8c 100644 --- a/dnp3/src/serial/outstation.rs +++ b/dnp3/src/serial/outstation.rs @@ -1,7 +1,7 @@ use crate::app::parse::options::ParseOptions; use crate::app::{Listener, MaybeAsync, RetryStrategy}; use crate::link::reader::LinkModes; -use crate::outstation::task::OutstationTask; +use crate::outstation::task::OutstationTask as ProtocolOutstationTask; use crate::outstation::{ ControlHandler, OutstationApplication, OutstationConfig, OutstationHandle, OutstationInformation, @@ -15,19 +15,12 @@ use tracing::Instrument; /// A fully configured serial outstation that has not yet opened its serial port. #[cfg(feature = "unstable")] pub struct SerialOutstation { - task: OutstationTask, + task: ProtocolOutstationTask, } -/// A serial outstation task that owns an open and configured serial port. -#[cfg(feature = "unstable")] -#[must_use = "a SerialOutstationTask does nothing unless you call .run()"] -pub struct SerialOutstationTask { - inner: OneShotSerialOutstationTask, -} - -struct OneShotSerialOutstationTask { +pub(crate) struct OneShotSerialOutstationTask { serial: tokio_serial::SerialStream, - task: OutstationTask, + task: ProtocolOutstationTask, } #[cfg(feature = "unstable")] @@ -48,28 +41,20 @@ impl SerialOutstation { self, path: &str, settings: SerialSettings, - ) -> std::io::Result { + ) -> std::io::Result { let serial = crate::serial::open(path, settings)?; - Ok(SerialOutstationTask { - inner: OneShotSerialOutstationTask::new(serial, self.task), - }) - } -} - -#[cfg(feature = "unstable")] -impl SerialOutstationTask { - /// Run the outstation until it is shut down or the serial port fails. - pub async fn run(self) { - self.inner.run().await; + Ok(crate::outstation::OutstationTask::serial( + OneShotSerialOutstationTask::new(serial, self.task), + )) } } impl OneShotSerialOutstationTask { - fn new(serial: tokio_serial::SerialStream, task: OutstationTask) -> Self { + fn new(serial: tokio_serial::SerialStream, task: ProtocolOutstationTask) -> Self { Self { serial, task } } - async fn run(mut self) { + pub(crate) async fn run(mut self) { let mut io = PhysLayer::Serial(self.serial); let _ = self.task.run(&mut io).await; } @@ -80,8 +65,8 @@ fn create_outstation( application: Box, information: Box, control_handler: Box, -) -> (OutstationTask, OutstationHandle) { - OutstationTask::create( +) -> (ProtocolOutstationTask, OutstationHandle) { + ProtocolOutstationTask::create( Enabled::Yes, LinkModes::serial(), ParseOptions::get_static(), diff --git a/dnp3/src/tcp/outstation/server.rs b/dnp3/src/tcp/outstation/server.rs index 34fe1104..e4662243 100644 --- a/dnp3/src/tcp/outstation/server.rs +++ b/dnp3/src/tcp/outstation/server.rs @@ -7,7 +7,7 @@ use crate::outstation::{ ConnectionState, ControlHandler, OutstationApplication, OutstationConfig, OutstationHandle, OutstationInformation, }; -use crate::tcp::server_task::{NewSession, ServerTask}; +use crate::tcp::server_task::{NewSession, ServerTask as OutstationServerTask}; use crate::tcp::{AddressFilter, FilterError, ServerHandle}; use crate::util::channel::Sender; use crate::util::phys::{PhysAddr, PhysLayer}; @@ -33,6 +33,24 @@ pub struct Server { connection_handler: ServerConnectionHandler, } +struct AcceptTask { + server: Server, + listener: tokio::net::TcpListener, + shutdown_rx: ShutdownListener, +} + +/// A TCP server task that accepts connections and routes them to outstations. +/// +/// It is the caller's responsibility to run this task on a Tokio runtime. Each outstation must be +/// run independently using the task returned by [`Server::add_outstation_task`] or the future +/// returned by [`Server::add_outstation_no_spawn`]. No tracing span is attached automatically, so +/// the caller may instrument the [`run`](Self::run) future as desired. +#[cfg(feature = "unstable")] +#[must_use = "a TcpServerTask does nothing unless you call .run()"] +pub struct TcpServerTask { + inner: AcceptTask, +} + enum ServerConnectionHandler { Tcp, #[cfg(feature = "enable-tls")] @@ -91,6 +109,58 @@ impl Server { listener: Box>, filter: AddressFilter, ) -> Result<(OutstationHandle, impl std::future::Future), FilterError> { + let (handle, mut adapter) = self.register_outstation( + config, + application, + information, + control_handler, + listener, + filter, + )?; + + let future = async move { + let _ = adapter.run().await; + }; + Ok((handle, future)) + } + + /// Associate an outstation with the TCP server and return a concrete task without spawning it. + /// + /// The caller is responsible for running the returned [`crate::outstation::OutstationTask`] + /// independently from the TCP server task. + #[cfg(feature = "unstable")] + #[allow(clippy::too_many_arguments)] + pub fn add_outstation_task( + &mut self, + config: OutstationConfig, + application: Box, + information: Box, + control_handler: Box, + listener: Box>, + filter: AddressFilter, + ) -> Result<(OutstationHandle, crate::outstation::OutstationTask), FilterError> { + let (handle, adapter) = self.register_outstation( + config, + application, + information, + control_handler, + listener, + filter, + )?; + + Ok((handle, crate::outstation::OutstationTask::tcp(adapter))) + } + + #[allow(clippy::too_many_arguments)] + fn register_outstation( + &mut self, + config: OutstationConfig, + application: Box, + information: Box, + control_handler: Box, + listener: Box>, + filter: AddressFilter, + ) -> Result<(OutstationHandle, OutstationServerTask), FilterError> { for item in self.outstations.iter() { if filter.conflicts_with(&item.filter) { return Err(FilterError::Conflict); @@ -108,7 +178,7 @@ impl Server { control_handler, ); - let (mut adapter, tx) = ServerTask::create(task, listener); + let (adapter, tx) = OutstationServerTask::create(task, listener); let outstation = OutstationInfo { filter, @@ -116,11 +186,7 @@ impl Server { sender: tx, }; self.outstations.push(outstation); - - let future = async move { - let _ = adapter.run().await; - }; - Ok((handle, future)) + Ok((handle, adapter)) } /// associate an outstation with the TcpServer and spawn it @@ -159,28 +225,29 @@ impl Server { self, ) -> Result<(ServerHandle, impl std::future::Future), tokio::io::Error> { let listener = tokio::net::TcpListener::bind(self.address).await?; - Ok(self.create_task(listener)) + let (handle, task) = self.create_task(listener); + Ok((handle, async move { task.run().await })) } /// Consume this server and create a task using an already-bound TCP listener. /// - /// The returned future does nothing until polled and may be run on any Tokio runtime. + /// The returned task does nothing until [`TcpServerTask::run`] is called and may be run on any + /// Tokio runtime. #[cfg(feature = "unstable")] - pub fn into_task( - self, - listener: tokio::net::TcpListener, - ) -> (ServerHandle, impl std::future::Future) { - self.create_task(listener) + pub fn into_task(self, listener: tokio::net::TcpListener) -> (ServerHandle, TcpServerTask) { + let (handle, task) = self.create_task(listener); + (handle, TcpServerTask { inner: task }) } - fn create_task( - mut self, - listener: tokio::net::TcpListener, - ) -> (ServerHandle, impl std::future::Future) { + fn create_task(self, listener: tokio::net::TcpListener) -> (ServerHandle, AcceptTask) { let addr = listener.local_addr().ok(); let (token, shutdown_rx) = crate::util::shutdown::shutdown_token(); - let task = async move { self.run(listener, shutdown_rx).await }; + let task = AcceptTask { + server: self, + listener, + shutdown_rx, + }; let handle = ServerHandle { addr, @@ -279,9 +346,40 @@ impl Server { } } +impl AcceptTask { + async fn run(mut self) -> Shutdown { + self.server.run(self.listener, self.shutdown_rx).await + } +} + +#[cfg(feature = "unstable")] +impl TcpServerTask { + /// Run until the associated [`ServerHandle`] is dropped or the accept loop terminates. + pub async fn run(self) { + self.inner.run().await; + } +} + #[cfg(all(test, feature = "unstable"))] mod tests { use super::*; + use crate::app::NullListener; + use crate::link::EndpointAddress; + use crate::outstation::database::EventBufferConfig; + + struct NullApplication; + impl OutstationApplication for NullApplication {} + + struct NullInformation; + impl OutstationInformation for NullInformation {} + + fn outstation_config() -> OutstationConfig { + OutstationConfig::new( + EndpointAddress::try_new(10).unwrap(), + EndpointAddress::try_new(1).unwrap(), + EventBufferConfig::all_types(0), + ) + } #[tokio::test] async fn into_task_uses_supplied_listener_and_stops_with_handle() { @@ -296,6 +394,27 @@ mod tests { assert_eq!(handle.local_addr(), Some(listener_address)); drop(handle); - task.await; + tokio::spawn(task.run()).await.unwrap(); + } + + #[tokio::test] + async fn add_outstation_task_returns_independently_runnable_task() { + let configured_address = "127.0.0.1:0".parse().unwrap(); + let mut server = Server::new_tcp_server(LinkErrorMode::Close, configured_address); + + let (handle, task) = server + .add_outstation_task( + outstation_config(), + Box::new(NullApplication), + Box::new(NullInformation), + crate::outstation::DefaultControlHandler::create(), + NullListener::create(), + AddressFilter::Any, + ) + .unwrap(); + + drop(handle); + drop(server); + tokio::spawn(task.run()).await.unwrap(); } } From 244f5cfa25a2dcdc68abff54a2906c07d0b199bd Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Mon, 20 Jul 2026 11:25:39 -0700 Subject: [PATCH 12/14] Keep FFI outstation tracing spans on the spawning variants Move the FFI TCP outstation server to the spawning add_outstation/bind methods, driven via block_on to enter the runtime context, so the library-defined dnp3-outstation-tcp and tcp-server spans are reused instead of being lost. This restores 1.6.0 FFI logging behavior. Document that the no-spawn APIs (add_outstation_no_spawn, bind_no_spawn) and the new unstable MasterTask intentionally attach no span so callers control instrumentation, and add a changelog note for the Rust-facing behavior change. --- CHANGELOG.md | 1 + dnp3/src/master/builder.rs | 4 ++++ dnp3/src/tcp/outstation/server.rs | 10 ++++++++++ ffi/dnp3-ffi/src/outstation/mod.rs | 29 +++++++++++++++++------------ 4 files changed, 32 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b21edd4b..6b726d80 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ ### 1.7.0-RC2 ### * :star: Add `Database::discard_unselected_events()` to drop undelivered events of selected classes on demand, e.g. from a disconnect handler. Returns per-class discard counts. Opt-in and lossy by design. See [#427](https://github.com/stepfunc/dnp3/issues/427). +* :wrench: The no-spawn TCP outstation server APIs (`Server::add_outstation_no_spawn` and `Server::bind_no_spawn`) no longer attach a tracing span to the futures they return. This is intentional so that non-spawning callers retain full control over instrumentation. Rust callers that relied on the previous automatic `dnp3-outstation-tcp` / `tcp-server` spans should now wrap the returned future in their own span before spawning it. The prebuilt bindings (C/C++, .NET, Java) are unaffected. See [#433](https://github.com/stepfunc/dnp3/pull/433). * :shield: Update `rustls-webpki` to 0.103.13 to resolve [RUSTSEC-2026-0098](https://rustsec.org/advisories/RUSTSEC-2026-0098) and [RUSTSEC-2026-0099](https://rustsec.org/advisories/RUSTSEC-2026-0099), both concerning incorrect acceptance of X.509 name constraints. Exposure is limited to TLS configurations using `CertificateMode::AuthorityBased`; `SelfSigned` mode bypasses the affected code path. * :bell: **This update only affects the prebuilt binary distributions of the bindings (C/C++, .NET, Java).** Rust consumers of the `dnp3` crate pick up the patched `rustls-webpki` automatically on rebuild. diff --git a/dnp3/src/master/builder.rs b/dnp3/src/master/builder.rs index 93a29c15..8ff3ac08 100644 --- a/dnp3/src/master/builder.rs +++ b/dnp3/src/master/builder.rs @@ -69,6 +69,10 @@ pub struct AssociationBuilder<'a> { /// the event loop. /// /// Created by binding a [`MasterBuilder`] or [`UdpMasterBuilder`] to a transport. +/// +/// No tracing span is attached automatically. This is by design so that the caller retains full +/// control over instrumentation and may wrap the [`run`](Self::run) future in whatever span it +/// chooses before spawning it. #[must_use = "a MasterTask does nothing unless you call .run()"] pub struct MasterTask { inner: MasterTaskType, diff --git a/dnp3/src/tcp/outstation/server.rs b/dnp3/src/tcp/outstation/server.rs index e4662243..2478b3cb 100644 --- a/dnp3/src/tcp/outstation/server.rs +++ b/dnp3/src/tcp/outstation/server.rs @@ -100,6 +100,11 @@ impl Server { } /// associate an outstation with the TcpServer, but do not spawn it + /// + /// The returned future is intentionally uninstrumented: unlike + /// [`add_outstation`](Self::add_outstation), it attaches no tracing span. This is by design so + /// that the caller retains full control over instrumentation and may wrap the future in + /// whatever span (e.g. `dnp3-outstation-tcp`) it chooses before spawning it. pub fn add_outstation_no_spawn( &mut self, config: OutstationConfig, @@ -221,6 +226,11 @@ impl Server { /// tuple. /// /// This may be called outside the Tokio runtime and allows for manual spawning + /// + /// The returned future is intentionally uninstrumented: unlike [`bind`](Self::bind), it + /// attaches no tracing span. This is by design so that the caller retains full control over + /// instrumentation and may wrap the future in whatever span (e.g. `tcp-server`) it chooses + /// before spawning it. pub async fn bind_no_spawn( self, ) -> Result<(ServerHandle, impl std::future::Future), tokio::io::Error> { diff --git a/ffi/dnp3-ffi/src/outstation/mod.rs b/ffi/dnp3-ffi/src/outstation/mod.rs index 6d7e2a77..1449ea3c 100644 --- a/ffi/dnp3-ffi/src/outstation/mod.rs +++ b/ffi/dnp3-ffi/src/outstation/mod.rs @@ -155,16 +155,20 @@ pub unsafe fn outstation_server_add_outstation( .ok_or(ffi::ParamError::NullParameter)? .into(); - let (outstation, task) = server_handle.add_outstation_no_spawn( - config, - Box::new(application), - Box::new(information), - Box::new(control_handler), - Box::new(listener), - filter, - )?; - - server.runtime.spawn(task)?; + // Drive the spawning variant inside the runtime context (via block_on) so that its + // internal tokio::spawn succeeds and the library-defined tracing span is applied. This + // avoids re-defining that instrumentation here. add_outstation is synchronous, so the + // future completes immediately. + let outstation = server.runtime.block_on(async move { + server_handle.add_outstation( + config, + Box::new(application), + Box::new(information), + Box::new(control_handler), + Box::new(listener), + filter, + ) + })??; Ok(Box::into_raw(Box::new(Outstation { handle: outstation, @@ -183,9 +187,10 @@ pub unsafe fn outstation_server_bind(server: *mut OutstationServer) -> Result<() OutstationServerState::Running(_) => return Err(ffi::ParamError::ServerAlreadyStarted), }; - let (handle, task) = server.runtime.block_on(server_handle.bind_no_spawn())??; + // Use the spawning variant inside the runtime context so its internal tokio::spawn and + // library-defined tracing span are applied, rather than re-defining instrumentation here. + let handle = server.runtime.block_on(server_handle.bind())??; - server.runtime.spawn(task)?; server.state = OutstationServerState::Running(handle); Box::leak(server); Ok(()) From 82bc9692147791e0abfe580ed6c9d597ed41feef Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Mon, 20 Jul 2026 11:58:37 -0700 Subject: [PATCH 13/14] Bump version to 1.7.0-RC3 and update changelog Add a 1.7.0-RC3 changelog section documenting the non-spawning master and outstation task APIs, the new EndpointList::into_connect_handler helpers, and the no-spawn tracing span behavior change. Bump all crate versions from 1.7.0-RC2 to 1.7.0-RC3. --- CHANGELOG.md | 6 +++++- Cargo.lock | 10 +++++----- dnp3/Cargo.toml | 2 +- ffi/dnp3-bindings/Cargo.toml | 2 +- ffi/dnp3-ffi-java/Cargo.toml | 2 +- ffi/dnp3-ffi/Cargo.toml | 2 +- ffi/dnp3-schema/Cargo.toml | 2 +- 7 files changed, 15 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b726d80..2de3c31e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,11 @@ +### 1.7.0-RC3 ### +* :star: Add unstable, non-spawning master and outstation task APIs behind the semver-exempt `unstable` feature, allowing applications to obtain runnable tasks/futures without the library calling `tokio::spawn` internally. See [#433](https://github.com/stepfunc/dnp3/pull/433). +* :star: Add `EndpointList::into_connect_handler()` and `EndpointList::into_connect_handler_with_options()` to build a `ClientConnectionHandler` from an endpoint list, e.g. for use with `spawn_master_tcp_client_3`. See [#433](https://github.com/stepfunc/dnp3/pull/433). +* :wrench: The no-spawn TCP outstation server APIs (`Server::add_outstation_no_spawn` and `Server::bind_no_spawn`) no longer attach a tracing span to the futures they return. This is intentional so that non-spawning callers retain full control over instrumentation. Rust callers that relied on the previous automatic `dnp3-outstation-tcp` / `tcp-server` spans should now wrap the returned future in their own span before spawning it. The prebuilt bindings (C/C++, .NET, Java) are unaffected. See [#433](https://github.com/stepfunc/dnp3/pull/433). + ### 1.7.0-RC2 ### * :star: Add `Database::discard_unselected_events()` to drop undelivered events of selected classes on demand, e.g. from a disconnect handler. Returns per-class discard counts. Opt-in and lossy by design. See [#427](https://github.com/stepfunc/dnp3/issues/427). -* :wrench: The no-spawn TCP outstation server APIs (`Server::add_outstation_no_spawn` and `Server::bind_no_spawn`) no longer attach a tracing span to the futures they return. This is intentional so that non-spawning callers retain full control over instrumentation. Rust callers that relied on the previous automatic `dnp3-outstation-tcp` / `tcp-server` spans should now wrap the returned future in their own span before spawning it. The prebuilt bindings (C/C++, .NET, Java) are unaffected. See [#433](https://github.com/stepfunc/dnp3/pull/433). * :shield: Update `rustls-webpki` to 0.103.13 to resolve [RUSTSEC-2026-0098](https://rustsec.org/advisories/RUSTSEC-2026-0098) and [RUSTSEC-2026-0099](https://rustsec.org/advisories/RUSTSEC-2026-0099), both concerning incorrect acceptance of X.509 name constraints. Exposure is limited to TLS configurations using `CertificateMode::AuthorityBased`; `SelfSigned` mode bypasses the affected code path. * :bell: **This update only affects the prebuilt binary distributions of the bindings (C/C++, .NET, Java).** Rust consumers of the `dnp3` crate pick up the patched `rustls-webpki` automatically on rebuild. diff --git a/Cargo.lock b/Cargo.lock index dbbce8e0..75296a35 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -391,7 +391,7 @@ dependencies = [ [[package]] name = "dnp3" -version = "1.7.0-RC2" +version = "1.7.0-RC3" dependencies = [ "assert_matches", "chrono", @@ -409,7 +409,7 @@ dependencies = [ [[package]] name = "dnp3-bindings" -version = "1.7.0-RC2" +version = "1.7.0-RC3" dependencies = [ "dnp3-schema", "oo-bindgen", @@ -428,7 +428,7 @@ dependencies = [ [[package]] name = "dnp3-ffi" -version = "1.7.0-RC2" +version = "1.7.0-RC3" dependencies = [ "dnp3", "dnp3-schema", @@ -446,7 +446,7 @@ dependencies = [ [[package]] name = "dnp3-ffi-java" -version = "1.7.0-RC2" +version = "1.7.0-RC3" dependencies = [ "dnp3-ffi", "dnp3-schema", @@ -456,7 +456,7 @@ dependencies = [ [[package]] name = "dnp3-schema" -version = "1.7.0-RC2" +version = "1.7.0-RC3" dependencies = [ "oo-bindgen", "sfio-tokio-ffi", diff --git a/dnp3/Cargo.toml b/dnp3/Cargo.toml index a29b85a6..3740b70d 100644 --- a/dnp3/Cargo.toml +++ b/dnp3/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "dnp3" -version = "1.7.0-RC2" +version = "1.7.0-RC3" authors = ["Step Function I/O LLC "] description = "Rust implementation of DNP3 (IEEE 1815) with idiomatic bindings for C, C++, .NET, and Java" diff --git a/ffi/dnp3-bindings/Cargo.toml b/ffi/dnp3-bindings/Cargo.toml index 02fb622c..a8c4ddc1 100644 --- a/ffi/dnp3-bindings/Cargo.toml +++ b/ffi/dnp3-bindings/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dnp3-bindings" -version = "1.7.0-RC2" +version = "1.7.0-RC3" # inherit from workspace rust-version.workspace = true diff --git a/ffi/dnp3-ffi-java/Cargo.toml b/ffi/dnp3-ffi-java/Cargo.toml index 9e971c55..c0a02800 100644 --- a/ffi/dnp3-ffi-java/Cargo.toml +++ b/ffi/dnp3-ffi-java/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dnp3-ffi-java" -version = "1.7.0-RC2" +version = "1.7.0-RC3" build = "build.rs" # inherit from workspace diff --git a/ffi/dnp3-ffi/Cargo.toml b/ffi/dnp3-ffi/Cargo.toml index a47e3676..2cd035f0 100644 --- a/ffi/dnp3-ffi/Cargo.toml +++ b/ffi/dnp3-ffi/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dnp3-ffi" -version = "1.7.0-RC2" +version = "1.7.0-RC3" # inherit from workspace rust-version.workspace = true diff --git a/ffi/dnp3-schema/Cargo.toml b/ffi/dnp3-schema/Cargo.toml index e6b83b7c..d74141b8 100644 --- a/ffi/dnp3-schema/Cargo.toml +++ b/ffi/dnp3-schema/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "dnp3-schema" # this version is what gets applied to the FFI libraries -version = "1.7.0-RC2" +version = "1.7.0-RC3" # inherit from workspace rust-version.workspace = true From 940addc5d26fb8659ccbe7ab5f663471a632d588 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Mon, 20 Jul 2026 12:04:49 -0700 Subject: [PATCH 14/14] Revert version bump and changelog to keep this PR feature-only The 1.7.0-RC3 version bump and changelog belong in a dedicated "prepare 1.7.0-RC3" PR per RELEASE.md (using bump-version.sh, which also updates the poms, csproj, CMakeLists, and sitedata.json). Revert them here so this PR contains only the feature work and the FFI tracing-span fix; the release prep will follow after this lands on main. --- CHANGELOG.md | 5 ----- Cargo.lock | 10 +++++----- dnp3/Cargo.toml | 2 +- ffi/dnp3-bindings/Cargo.toml | 2 +- ffi/dnp3-ffi-java/Cargo.toml | 2 +- ffi/dnp3-ffi/Cargo.toml | 2 +- ffi/dnp3-schema/Cargo.toml | 2 +- 7 files changed, 10 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2de3c31e..b21edd4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,9 +1,4 @@ -### 1.7.0-RC3 ### -* :star: Add unstable, non-spawning master and outstation task APIs behind the semver-exempt `unstable` feature, allowing applications to obtain runnable tasks/futures without the library calling `tokio::spawn` internally. See [#433](https://github.com/stepfunc/dnp3/pull/433). -* :star: Add `EndpointList::into_connect_handler()` and `EndpointList::into_connect_handler_with_options()` to build a `ClientConnectionHandler` from an endpoint list, e.g. for use with `spawn_master_tcp_client_3`. See [#433](https://github.com/stepfunc/dnp3/pull/433). -* :wrench: The no-spawn TCP outstation server APIs (`Server::add_outstation_no_spawn` and `Server::bind_no_spawn`) no longer attach a tracing span to the futures they return. This is intentional so that non-spawning callers retain full control over instrumentation. Rust callers that relied on the previous automatic `dnp3-outstation-tcp` / `tcp-server` spans should now wrap the returned future in their own span before spawning it. The prebuilt bindings (C/C++, .NET, Java) are unaffected. See [#433](https://github.com/stepfunc/dnp3/pull/433). - ### 1.7.0-RC2 ### * :star: Add `Database::discard_unselected_events()` to drop undelivered events of selected classes on demand, e.g. from a disconnect handler. Returns per-class discard counts. Opt-in and lossy by design. See [#427](https://github.com/stepfunc/dnp3/issues/427). * :shield: Update `rustls-webpki` to 0.103.13 to resolve [RUSTSEC-2026-0098](https://rustsec.org/advisories/RUSTSEC-2026-0098) and [RUSTSEC-2026-0099](https://rustsec.org/advisories/RUSTSEC-2026-0099), both concerning incorrect acceptance of X.509 name constraints. Exposure is limited to TLS configurations using `CertificateMode::AuthorityBased`; `SelfSigned` mode bypasses the affected code path. diff --git a/Cargo.lock b/Cargo.lock index 75296a35..dbbce8e0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -391,7 +391,7 @@ dependencies = [ [[package]] name = "dnp3" -version = "1.7.0-RC3" +version = "1.7.0-RC2" dependencies = [ "assert_matches", "chrono", @@ -409,7 +409,7 @@ dependencies = [ [[package]] name = "dnp3-bindings" -version = "1.7.0-RC3" +version = "1.7.0-RC2" dependencies = [ "dnp3-schema", "oo-bindgen", @@ -428,7 +428,7 @@ dependencies = [ [[package]] name = "dnp3-ffi" -version = "1.7.0-RC3" +version = "1.7.0-RC2" dependencies = [ "dnp3", "dnp3-schema", @@ -446,7 +446,7 @@ dependencies = [ [[package]] name = "dnp3-ffi-java" -version = "1.7.0-RC3" +version = "1.7.0-RC2" dependencies = [ "dnp3-ffi", "dnp3-schema", @@ -456,7 +456,7 @@ dependencies = [ [[package]] name = "dnp3-schema" -version = "1.7.0-RC3" +version = "1.7.0-RC2" dependencies = [ "oo-bindgen", "sfio-tokio-ffi", diff --git a/dnp3/Cargo.toml b/dnp3/Cargo.toml index 3740b70d..a29b85a6 100644 --- a/dnp3/Cargo.toml +++ b/dnp3/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "dnp3" -version = "1.7.0-RC3" +version = "1.7.0-RC2" authors = ["Step Function I/O LLC "] description = "Rust implementation of DNP3 (IEEE 1815) with idiomatic bindings for C, C++, .NET, and Java" diff --git a/ffi/dnp3-bindings/Cargo.toml b/ffi/dnp3-bindings/Cargo.toml index a8c4ddc1..02fb622c 100644 --- a/ffi/dnp3-bindings/Cargo.toml +++ b/ffi/dnp3-bindings/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dnp3-bindings" -version = "1.7.0-RC3" +version = "1.7.0-RC2" # inherit from workspace rust-version.workspace = true diff --git a/ffi/dnp3-ffi-java/Cargo.toml b/ffi/dnp3-ffi-java/Cargo.toml index c0a02800..9e971c55 100644 --- a/ffi/dnp3-ffi-java/Cargo.toml +++ b/ffi/dnp3-ffi-java/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dnp3-ffi-java" -version = "1.7.0-RC3" +version = "1.7.0-RC2" build = "build.rs" # inherit from workspace diff --git a/ffi/dnp3-ffi/Cargo.toml b/ffi/dnp3-ffi/Cargo.toml index 2cd035f0..a47e3676 100644 --- a/ffi/dnp3-ffi/Cargo.toml +++ b/ffi/dnp3-ffi/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dnp3-ffi" -version = "1.7.0-RC3" +version = "1.7.0-RC2" # inherit from workspace rust-version.workspace = true diff --git a/ffi/dnp3-schema/Cargo.toml b/ffi/dnp3-schema/Cargo.toml index d74141b8..e6b83b7c 100644 --- a/ffi/dnp3-schema/Cargo.toml +++ b/ffi/dnp3-schema/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "dnp3-schema" # this version is what gets applied to the FFI libraries -version = "1.7.0-RC3" +version = "1.7.0-RC2" # inherit from workspace rust-version.workspace = true