Refactor master modules

dnp3/examples/master.rs

@@ -243,37 +243,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // ANCHOR_END: logging
 
     // spawn the master channel based on the command line argument
-    let mut channel = create_channel()?;
-
-    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
-        }
-    };
+    let (mut channel, mut association) = create_channel_and_association().await?;
 
     // create an event poll
     // ANCHOR: add_poll
@@ -497,7 +467,8 @@ impl CliHandler {
 }
 
 // create the specified channel based on the command line argument
-fn create_channel() -> Result<MasterChannel, Box<dyn std::error::Error>> {
+async fn create_channel_and_association(
+) -> Result<(MasterChannel, AssociationHandle), Box<dyn std::error::Error>> {
     let args: Vec<String> = std::env::args().collect();
     let transport: &str = match args.as_slice() {
         [_, x] => x,
@@ -508,14 +479,34 @@ fn create_channel() -> Result<MasterChannel, Box<dyn std::error::Error>> {
         }
     };
     match transport {
-        "tcp" => create_tcp_channel(),
-        "udp" => create_udp_channel(),
+        "tcp" => {
+            let mut channel = create_tcp_channel()?;
+            let assoc = add_association(&mut channel).await?;
+            Ok((channel, assoc))
+        }
+        "udp" => {
+            let mut channel = create_udp_channel()?;
+            let assoc = add_udp_association(&mut channel).await?;
+            Ok((channel, assoc))
+        }
         #[cfg(feature = "serial")]
-        "serial" => create_serial_channel(),
+        "serial" => {
+            let mut channel = create_serial_channel()?;
+            let assoc = add_association(&mut channel).await?;
+            Ok((channel, assoc))
+        }
         #[cfg(feature = "tls")]
-        "tls-ca" => create_tls_channel(get_tls_authority_config()?),
+        "tls-ca" => {
+            let mut channel = create_tls_channel(get_tls_authority_config()?)?;
+            let assoc = add_association(&mut channel).await?;
+            Ok((channel, assoc))
+        }
         #[cfg(feature = "tls")]
-        "tls-self-signed" => create_tls_channel(get_tls_self_signed_config()?),
+        "tls-self-signed" => {
+            let mut channel = create_tls_channel(get_tls_self_signed_config()?)?;
+            let assoc = add_association(&mut channel).await?;
+            Ok((channel, assoc))
+        }
         _ => {
             eprintln!(
                 "unknown transport '{}', options are (tcp, serial, tls-ca, tls-self-signed)",
@@ -526,6 +517,41 @@ fn create_channel() -> Result<MasterChannel, Box<dyn std::error::Error>> {
     }
 }
 
+async fn add_association(
+    channel: &mut MasterChannel,
+) -> Result<AssociationHandle, Box<dyn std::error::Error>> {
+    // ANCHOR: association_create
+    let association = channel
+        .add_association(
+            EndpointAddress::try_new(1024)?,
+            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,
+) -> Result<AssociationHandle, Box<dyn std::error::Error>> {
+    // ANCHOR: association_create_udp
+    let association = 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
+    Ok(association)
+}
+
 // ANCHOR: master_channel_config
 fn get_master_channel_config() -> Result<MasterChannelConfig, Box<dyn std::error::Error>> {
     let mut config = MasterChannelConfig::new(EndpointAddress::try_new(1)?);

dnp3/src/master/association.rs

@@ -11,7 +11,7 @@ use crate::app::{Sequence, Timeout};
 use crate::link::EndpointAddress;
 use crate::master::error::{AssociationError, TaskError, TimeSyncError};
 use crate::master::extract::extract_measurements;
-use crate::master::handler::{AssociationHandler, Promise};
+use crate::master::handler::AssociationHandler;
 use crate::master::messages::AssociationMsgType;
 use crate::master::poll::{PollHandle, PollMap, PollMsg};
 use crate::master::request::{Classes, EventClasses, TimeSyncProcedure};
@@ -22,6 +22,7 @@ use crate::master::tasks::{AppTask, AssociationTask, ReadTask, Task};
 use crate::master::{AssociationInformation, ReadHandler, ReadType, TaskType};
 use crate::util::Smallest;
 
+use crate::master::promise::Promise;
 use crate::transport::FragmentAddr;
 use crate::util::session::RunError;
 use tokio::time::Instant;

dnp3/src/master/extract.rs

@@ -3,7 +3,7 @@ use crate::app::measurement::*;
 use crate::app::parse::parser::{HeaderCollection, HeaderDetails, ObjectHeader};
 use crate::app::variations::*;
 use crate::app::ResponseHeader;
-use crate::master::handler::ReadHandler;
+use crate::master::ReadHandler;
 use crate::master::ReadType;
 
 /// Extract measurements from a HeaderCollection, sinking them into
@@ -91,7 +91,7 @@ mod test {
     use crate::app::control::CommandStatus;
     use crate::app::parse::parser::HeaderCollection;
     use crate::app::*;
-    use crate::master::handler::{HeaderInfo, ReadHandler};
+    use crate::master::{HeaderInfo, ReadHandler};
 
     use super::*;