UDP Support

dnp3/Cargo.toml

@@ -17,7 +17,7 @@ tracing = "0.1"
 chrono = "0.4"
 tokio = { version = "1", features = ["net", "sync", "io-util", "io-std", "time", "rt", "rt-multi-thread", "macros"] }
 xxhash-rust = { version = "0.8", features = ["xxh64"] }
-scursor = "0.1.0"
+scursor = "0.2.0"
 
 # TLS dependencies
 sfio-rustls-config = { version = "0.3.2", optional = true }

dnp3/examples/master.rs

@@ -18,6 +18,7 @@ use dnp3::serial::*;
 use dnp3::tcp::tls::*;
 
 use dnp3::outstation::FreezeInterval;
+use dnp3::udp::spawn_master_udp;
 use std::process::exit;
 
 /// read handler that does nothing
@@ -225,7 +226,7 @@ impl FileReader for FileLogger {
   The program initializes a master channel based on the command line argument and then enters a loop
   reading console input allowing the user to perform common tasks interactively.
 
-  All of the configuration values are hard-coded but can be changed with a recompile.
+  All the configuration values are hard-coded but can be changed with a recompile.
 */
 // ANCHOR: runtime_init
 #[tokio::main(flavor = "multi_thread")]
@@ -243,17 +244,35 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // spawn the master channel based on the command line argument
     let mut channel = create_channel()?;
 
-    // ANCHOR: association_create
-    let mut association = channel
-        .add_association(
-            EndpointAddress::try_new(1024)?,
-            get_association_config(),
-            ExampleReadHandler::boxed(),
-            Box::new(ExampleAssociationHandler),
-            Box::new(ExampleAssociationInformation),
-        )
-        .await?;
-    // ANCHOR_END: association_create
+    let mut association = match channel.get_channel_type() {
+        MasterChannelType::Udp => {
+            // ANCHOR: association_create_udp
+            channel
+                .add_udp_association(
+                    EndpointAddress::try_new(1024)?,
+                    "127.0.0.1:20000".parse()?,
+                    get_association_config(),
+                    ExampleReadHandler::boxed(),
+                    Box::new(ExampleAssociationHandler),
+                    Box::new(ExampleAssociationInformation),
+                )
+                .await?
+            // ANCHOR_END: association_create_udp
+        }
+        MasterChannelType::Stream => {
+            // ANCHOR: association_create
+            channel
+                .add_association(
+                    EndpointAddress::try_new(1024)?,
+                    get_association_config(),
+                    ExampleReadHandler::boxed(),
+                    Box::new(ExampleAssociationHandler),
+                    Box::new(ExampleAssociationInformation),
+                )
+                .await?
+            // ANCHOR_END: association_create
+        }
+    };
 
     // create an event poll
     // ANCHOR: add_poll
@@ -483,12 +502,13 @@ fn create_channel() -> Result<MasterChannel, Box<dyn std::error::Error>> {
         [_, x] => x,
         _ => {
             eprintln!("please specify a transport:");
-            eprintln!("usage: master <transport> (tcp, serial, tls-ca, tls-self-signed)");
+            eprintln!("usage: master <transport> (tcp, udp, serial, tls-ca, tls-self-signed)");
             exit(-1);
         }
     };
     match transport {
         "tcp" => create_tcp_channel(),
+        "udp" => create_udp_channel(),
         #[cfg(feature = "serial")]
         "serial" => create_serial_channel(),
         #[cfg(feature = "tls")]
@@ -575,6 +595,18 @@ fn create_tcp_channel() -> Result<MasterChannel, Box<dyn std::error::Error>> {
     Ok(channel)
 }
 
+fn create_udp_channel() -> Result<MasterChannel, Box<dyn std::error::Error>> {
+    // ANCHOR: create_master_udp_channel
+    let channel = spawn_master_udp(
+        "127.0.0.1:20001".parse()?,
+        LinkReadMode::Datagram,
+        Timeout::from_secs(5)?,
+        get_master_channel_config()?,
+    );
+    // ANCHOR_END: create_master_udp_channel
+    Ok(channel)
+}
+
 #[cfg(feature = "serial")]
 fn create_serial_channel() -> Result<MasterChannel, Box<dyn std::error::Error>> {
     // ANCHOR: create_master_serial_channel

dnp3/examples/outstation.rs

@@ -18,6 +18,7 @@ use tokio_util::codec::LinesCodec;
 use dnp3::serial::*;
 #[cfg(feature = "tls")]
 use dnp3::tcp::tls::*;
+use dnp3::udp::{spawn_outstation_udp, OutstationUdpConfig, UdpSocketMode};
 
 /// example of using the outstation API asynchronously from within the Tokio runtime
 ///
@@ -41,6 +42,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     match transport {
         "tcp" => run_tcp_server().await,
         "tcp-client" => run_tcp_client().await,
+        "udp" => run_udp().await,
         #[cfg(feature = "serial")]
         "serial" => run_serial().await,
         #[cfg(feature = "tls")]
@@ -245,6 +247,26 @@ async fn run_tcp_server() -> Result<(), Box<dyn std::error::Error>> {
     run_server(server).await
 }
 
+async fn run_udp() -> Result<(), Box<dyn std::error::Error>> {
+    let udp_config = OutstationUdpConfig {
+        local_endpoint: "127.0.0.1:20000".parse().unwrap(),
+        remote_endpoint: "127.0.0.1:20001".parse().unwrap(),
+        socket_mode: UdpSocketMode::OneToOne,
+        link_read_mode: LinkReadMode::Datagram,
+        retry_delay: Timeout::from_secs(5)?,
+    };
+
+    let outstation = spawn_outstation_udp(
+        udp_config,
+        get_outstation_config(),
+        Box::new(ExampleOutstationApplication),
+        Box::new(ExampleOutstationInformation),
+        Box::new(ExampleControlHandler),
+    );
+
+    run_outstation(outstation).await
+}
+
 async fn run_tcp_client() -> Result<(), Box<dyn std::error::Error>> {
     let outstation = spawn_outstation_tcp_client(
         LinkErrorMode::Close,
@@ -304,7 +326,7 @@ async fn run_server(mut server: Server) -> Result<(), Box<dyn std::error::Error>
     )?;
     // ANCHOR_END: tcp_server_spawn_outstation
 
-    // setup the outstation's database before we spawn it
+    // set up the outstation's database before we spawn it
     // ANCHOR: database_init
     outstation.transaction(|db| {
         // initialize 10 points of each type

dnp3/src/lib.rs

@@ -1,6 +1,6 @@
 #![doc = include_str!("../README.md")]
 #![deny(
-dead_code,
+//dead_code,
 arithmetic_overflow,
 invalid_type_param_default,
 missing_fragment_specifier,
@@ -65,6 +65,8 @@ pub mod outstation;
 pub mod serial;
 /// Entry points and types for TCP
 pub mod tcp;
+/// Entry points and types for UDP
+pub mod udp;
 
 pub(crate) mod transport;
 pub(crate) mod util;

dnp3/src/link/header.rs

@@ -1,4 +1,5 @@
 use crate::link::EndpointAddress;
+use crate::util::phys::PhysAddr;
 
 use super::function::Function;
 
@@ -127,18 +128,21 @@ pub(crate) struct FrameInfo {
     pub(crate) source: EndpointAddress,
     pub(crate) broadcast: Option<BroadcastConfirmMode>,
     pub(crate) frame_type: FrameType,
+    pub(crate) phys_addr: PhysAddr,
 }
 
 impl FrameInfo {
     pub(crate) fn new(
         source: EndpointAddress,
         broadcast: Option<BroadcastConfirmMode>,
         frame_type: FrameType,
+        phys_addr: PhysAddr,
     ) -> Self {
         Self {
             source,
             broadcast,
             frame_type,
+            phys_addr,
         }
     }
 }
@@ -180,13 +184,13 @@ impl Header {
 
     pub(crate) fn request_link_status(
         is_master: bool,
-        destination: AnyAddress,
-        source: AnyAddress,
+        destination: EndpointAddress,
+        source: EndpointAddress,
     ) -> Self {
         Self::new(
             ControlField::new(is_master, Function::PriRequestLinkStatus),
-            destination,
-            source,
+            destination.wrap(),
+            source.wrap(),
         )
     }
 }

dnp3/src/link/layer.rs

@@ -8,9 +8,10 @@ use crate::link::header::{
     AnyAddress, BroadcastConfirmMode, ControlField, FrameInfo, FrameType, Header,
 };
 use crate::link::parser::FramePayload;
-use crate::link::{EndpointAddress, LinkErrorMode};
+use crate::link::reader::LinkModes;
+use crate::link::EndpointAddress;
 use crate::outstation::Feature;
-use crate::util::phys::PhysLayer;
+use crate::util::phys::{PhysAddr, PhysLayer};
 
 enum SecondaryState {
     NotReset,
@@ -39,7 +40,8 @@ impl Reply {
 
 impl Layer {
     pub(crate) fn new(
-        error_mode: LinkErrorMode,
+        link_modes: LinkModes,
+        max_rx_fragment_size: usize,
         endpoint_type: EndpointType,
         self_address: Feature,
         local_address: EndpointAddress,
@@ -49,7 +51,7 @@ impl Layer {
             self_address,
             local_address,
             secondary_state: SecondaryState::NotReset,
-            reader: super::reader::Reader::new(error_mode),
+            reader: super::reader::Reader::new(link_modes, max_rx_fragment_size),
             tx_buffer: [0; super::constant::LINK_HEADER_LENGTH],
         }
     }
@@ -91,19 +93,24 @@ impl Layer {
         level: DecodeLevel,
         payload: &mut FramePayload,
     ) -> Result<Option<FrameInfo>, LinkError> {
-        let header = self.reader.read(io, payload, level).await?;
-        let (info, reply) = self.process_header(&header);
+        let (header, addr) = self.reader.read_frame(io, payload, level).await?;
+        let (info, reply) = self.process_header(header, addr);
         if let Some(reply) = reply {
             let header = self.get_header(reply);
             if level.link.enabled() {
                 tracing::info!("LINK TX - {}", LinkDisplay::new(header, &[], level.link));
             }
-            io.write(self.format_reply(header), level.physical).await?
+            io.write(self.format_reply(header), addr, level.physical)
+                .await?
         }
         Ok(info)
     }
 
-    fn process_header(&mut self, header: &Header) -> (Option<FrameInfo>, Option<Reply>) {
+    fn process_header(
+        &mut self,
+        header: Header,
+        addr: PhysAddr,
+    ) -> (Option<FrameInfo>, Option<Reply>) {
         // ignore frames sent from the same endpoint type
         if header.control.master == self.endpoint_type.dir_bit() {
             //  we don't log this
@@ -176,7 +183,7 @@ impl Layer {
                 }
 
                 (
-                    Some(FrameInfo::new(source, broadcast, FrameType::Data)),
+                    Some(FrameInfo::new(source, broadcast, FrameType::Data, addr)),
                     None,
                 )
             }
@@ -213,7 +220,7 @@ impl Layer {
                         if header.control.fcb == expected {
                             self.secondary_state = SecondaryState::Reset(!expected);
                             (
-                                Some(FrameInfo::new(source, broadcast, FrameType::Data)),
+                                Some(FrameInfo::new(source, broadcast, FrameType::Data, addr)),
                                 response,
                             )
                         } else {
@@ -234,6 +241,7 @@ impl Layer {
                         source,
                         broadcast,
                         FrameType::LinkStatusRequest,
+                        addr,
                     )),
                     Some(Reply::new(source, Function::SecLinkStatus)),
                 )
@@ -243,6 +251,7 @@ impl Layer {
                     source,
                     broadcast,
                     FrameType::LinkStatusResponse,
+                    addr,
                 )),
                 None,
             ),

dnp3/src/link/mod.rs

@@ -41,6 +41,23 @@ pub enum LinkErrorMode {
     Close,
 }
 
+/// Controls how the link-layer parser treats frames that span multiple calls to read of
+/// the physical layer.
+///
+/// UDP is unique in that the specification requires that link layer frames be wholly contained
+/// within datagrams, but this can be relaxed by configuration.
+#[cfg_attr(
+    feature = "serialization",
+    derive(serde::Serialize, serde::Deserialize)
+)]
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+pub enum LinkReadMode {
+    /// Reading from a stream (TCP, serial, etc.) where link-layer frames MAY span separate calls to read
+    Stream,
+    /// Reading datagrams (UDP) where link-layer frames MAY NOT span separate calls to read
+    Datagram,
+}
+
 /// Represents a validated 16-bit endpoint address for a master or an outstation
 /// Certain special addresses are not allowed by the standard to be used
 /// as endpoint addresses.