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/Cargo.toml b/dnp3/Cargo.toml index 55ac8c07..a29b85a6 100644 --- a/dnp3/Cargo.toml +++ b/dnp3/Cargo.toml @@ -46,6 +46,9 @@ 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 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 e1fe626e..e8996f39 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; @@ -14,6 +15,8 @@ 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; use crate::master::tasks::time::TimeSyncTask; @@ -348,6 +351,46 @@ impl Association { } } + #[cfg(feature = "unstable")] + 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(), + } + } + + #[cfg(feature = "unstable")] + 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); + } + } + + #[cfg(feature = "unstable")] + 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) => { @@ -750,6 +793,13 @@ impl AssociationMap { } } + #[cfg(feature = "unstable")] + 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), @@ -764,13 +814,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) { @@ -836,3 +894,56 @@ impl AssociationMap { Next::None } } + +#[cfg(all(test, feature = "unstable"))] +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 new file mode 100644 index 00000000..8ff3ac08 --- /dev/null +++ b/dnp3/src/master/builder.rs @@ -0,0 +1,665 @@ +use std::net::SocketAddr; +use std::time::Duration; + +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, +}; +use crate::app::parse::options::ParseOptions; +use crate::app::{Listener, Timeout}; +use crate::link::reader::LinkModes; +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}; +use tokio::time::Instant; + +/// 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 +/// [`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, +} + +/// 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, +} + +/// Builder for configuring an association and its polls before the master task +/// starts. +/// +/// 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 +/// 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, +} + +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>, +} + +#[cfg(feature = "enable-tls")] +struct TlsTask { + inner: InnerMasterTask, + connect_handler: Box, + tls_config: crate::tcp::tls::TlsClientConfig, + listener: Box>, +} + +#[cfg(feature = "serial")] +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`]. + /// + /// 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 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, + ) -> Result, AssociationError> { + let addr = FragmentAddr { + link: address, + phys: PhysAddr::None, + }; + 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, + ))?; + Ok(AssociationBuilder { + handle, + association, + }) + } + + /// 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>, + ) -> 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, + }), + } + } + + /// 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>, + ) -> 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 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 add_association( + &mut self, + address: EndpointAddress, + destination: SocketAddr, + config: AssociationConfig, + read_handler: Box, + assoc_handler: Box, + assoc_information: Box, + ) -> Result, AssociationError> { + let addr = FragmentAddr { + link: address, + phys: PhysAddr::Udp(destination), + }; + 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, + ))?; + Ok(AssociationBuilder { + handle, + association, + }) + } + + /// 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 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(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 + } +} + +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(mut self) { + self.inner.start(Instant::now()); + 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 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 session = Session::master(self.inner); + let mut client = ClientTask::new( + session, + self.connect_handler, + PostConnectionHandler::Tcp, + self.listener, + ); + client.run().await; + } +} + +#[cfg(feature = "enable-tls")] +impl TlsTask { + async fn run(self) { + let session = Session::master(self.inner); + let mut client = ClientTask::new( + session, + self.connect_handler, + PostConnectionHandler::Tls(self.tls_config), + self.listener, + ); + client.run().await; + } +} + +#[cfg(feature = "serial")] +impl SerialTask { + async fn run(self) { + 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().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().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/mod.rs b/dnp3/src/master/mod.rs index e410bbf8..61fb560a 100644 --- a/dnp3/src/master/mod.rs +++ b/dnp3/src/master/mod.rs @@ -1,4 +1,6 @@ pub use association::*; +#[cfg(feature = "unstable")] +pub use builder::*; pub use error::*; pub use file::*; pub use handler::*; @@ -7,6 +9,8 @@ pub use read_handler::*; pub use request::*; mod association; +#[cfg(feature = "unstable")] +mod builder; mod error; mod file; mod handler; diff --git a/dnp3/src/master/poll.rs b/dnp3/src/master/poll.rs index 194a6c32..0c4dd97e 100644 --- a/dnp3/src/master/poll.rs +++ b/dnp3/src/master/poll.rs @@ -27,13 +27,33 @@ 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, + Running, +} + 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, id: 0, polls: BTreeMap::new(), } @@ -42,10 +62,33 @@ 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)); + #[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 } + /// 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; + } + + 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 +139,21 @@ impl Poll { } } + #[cfg(feature = "unstable")] + fn new_deferred(id: u64, request: ReadRequest, period: Duration) -> Self { + Self { + id, + request, + period, + next: None, + } + } + + #[cfg(feature = "unstable")] + 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 +239,45 @@ impl PollHandle { .await } } + +#[cfg(all(test, feature = "unstable"))] +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 bc938cc1..8c274a03 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, @@ -68,6 +87,11 @@ impl MasterTask { self.session.enabled } + #[cfg(feature = "unstable")] + 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 @@ -105,11 +129,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(), } 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 5012551b..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, @@ -12,6 +12,72 @@ 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: ProtocolOutstationTask, +} + +pub(crate) struct OneShotSerialOutstationTask { + serial: tokio_serial::SerialStream, + task: ProtocolOutstationTask, +} + +#[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 fn open( + self, + path: &str, + settings: SerialSettings, + ) -> std::io::Result { + let serial = crate::serial::open(path, settings)?; + Ok(crate::outstation::OutstationTask::serial( + OneShotSerialOutstationTask::new(serial, self.task), + )) + } +} + +impl OneShotSerialOutstationTask { + fn new(serial: tokio_serial::SerialStream, task: ProtocolOutstationTask) -> Self { + Self { serial, task } + } + + pub(crate) async fn run(mut self) { + let mut io = PhysLayer::Serial(self.serial); + let _ = self.task.run(&mut io).await; + } +} + +fn create_outstation( + config: OutstationConfig, + application: Box, + information: Box, + control_handler: Box, +) -> (ProtocolOutstationTask, OutstationHandle) { + ProtocolOutstationTask::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 +96,13 @@ 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 (task, handle) = create_outstation(config, 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); + tokio::spawn( + OneShotSerialOutstationTask::new(serial, task) + .run() + .instrument(tracing::info_span!("dnp3-outstation-serial", "port" = ?log_path)), + ); Ok(handle) } @@ -82,16 +135,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 +150,48 @@ 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(); + } + + #[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(), + ); + 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. 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/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/tcp/outstation/server.rs b/dnp3/src/tcp/outstation/server.rs index 76346f84..2478b3cb 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")] @@ -82,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, @@ -91,6 +114,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 +183,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,17 +191,7 @@ impl Server { sender: tx, }; 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; - }; - Ok((handle, future)) + Ok((handle, adapter)) } /// associate an outstation with the TcpServer and spawn it @@ -141,6 +206,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 +216,9 @@ impl Server { listener, filter, )?; - tokio::spawn(future); + tokio::spawn(future.instrument( + tracing::info_span!("dnp3-outstation-tcp", "listen" = ?endpoint, "addr" = address), + )); Ok(handle) } @@ -157,20 +226,37 @@ 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( - mut self, + self, ) -> Result<(ServerHandle, impl std::future::Future), tokio::io::Error> { let listener = tokio::net::TcpListener::bind(self.address).await?; + let (handle, task) = self.create_task(listener); + Ok((handle, async move { task.run().await })) + } - let addr = listener.local_addr().ok(); + /// Consume this server and create a task using an already-bound TCP listener. + /// + /// 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, TcpServerTask) { + let (handle, task) = self.create_task(listener); + (handle, TcpServerTask { inner: task }) + } + 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 { - let local = self.address; - self.run(listener, shutdown_rx) - .instrument(tracing::info_span!("tcp-server", "listen" = ?local)) - .await + let task = AcceptTask { + server: self, + listener, + shutdown_rx, }; let handle = ServerHandle { @@ -178,7 +264,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 @@ -188,8 +274,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) } @@ -267,3 +355,76 @@ 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() { + 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); + 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(); + } +} 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/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(())