chat replaced with ConnectionHandler object. connct function

Cargo.toml

 [package]
 name = "roboplc"
-version = "0.3.2"
+version = "0.4.0"
 edition = "2021"
 authors = ["Serhij S. <div@altertech.com>"]
 license = "Apache-2.0"

src/comm/mod.rs

@@ -21,6 +21,11 @@ impl Client {
     pub fn lock(&self) -> MutexGuard<()> {
         self.0.lock()
     }
+    /// Connect the client. Does not need to be called for request/response protocols as the client
+    /// is automatically connected when the first request is made.
+    pub fn connect(&self) -> Result<()> {
+        self.0.connect()
+    }
     /// Reconnect the client in case of read/write problems
     pub fn reconnect(&self) {
         self.0.reconnect();
@@ -81,6 +86,7 @@ pub trait Stream: Read + Write + Send {}
 
 trait Communicator {
     fn lock(&self) -> MutexGuard<()>;
+    fn connect(&self) -> Result<()>;
     fn reconnect(&self);
     fn write(&self, buf: &[u8]) -> Result<()>;
     fn read_exact(&self, buf: &mut [u8]) -> Result<()>;
@@ -130,14 +136,18 @@ impl Timeouts {
     }
 }
 
-pub type ChatFn = dyn Fn(&mut dyn Stream) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
-    + Send
-    + Sync;
+pub trait ConnectionHandler {
+    /// called right after the connection is established
+    fn on_connect(
+        &self,
+        stream: &mut dyn Stream,
+    ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>;
+}
 
 /// Connection Options
 pub struct ConnectionOptions {
     with_reader: bool,
-    chat: Option<Box<ChatFn>>,
+    connection_handler: Option<Box<dyn ConnectionHandler + Send + Sync>>,
     timeouts: Timeouts,
 }
 
@@ -146,7 +156,7 @@ impl ConnectionOptions {
     pub fn new(timeout: Duration) -> Self {
         Self {
             with_reader: false,
-            chat: None,
+            connection_handler: None,
             timeouts: Timeouts {
                 connect: timeout,
                 read: timeout,
@@ -161,16 +171,13 @@ impl ConnectionOptions {
         self.with_reader = true;
         self
     }
-    /// Set the chat function. The chat function is called after the connection is established. The
-    /// chat function can be used to implement custom protocols that require additional setup.
-    pub fn chat<F>(mut self, chat: F) -> Self
+    /// Set the connection handler. The connection handler is used to implement custom protocols
+    /// that require additional setup/handling. Replaces "chat" function.
+    pub fn connection_handler<T>(mut self, connection_handler: T) -> Self
     where
-        F: Fn(&mut dyn Stream) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
-            + Send
-            + Sync
-            + 'static,
+        T: ConnectionHandler + Send + Sync + 'static,
     {
-        self.chat = Some(Box::new(chat));
+        self.connection_handler = Some(Box::new(connection_handler));
         self
     }
     /// Set timeouts

src/comm/serial.rs

@@ -153,6 +153,9 @@ impl Communicator for Serial {
     fn session_id(&self) -> usize {
         self.session_id.load(Ordering::Acquire)
     }
+    fn connect(&self) -> Result<()> {
+        self.get_port().map(|_| ())
+    }
     fn reconnect(&self) {
         let mut port = self.port.lock();
         port.system_port.take();

src/comm/tcp.rs

@@ -2,7 +2,8 @@ use crate::pchannel;
 use crate::{Error, Result};
 
 use super::{
-    ChatFn, Client, CommReader, Communicator, ConnectionOptions, Protocol, Stream, Timeouts,
+    Client, CommReader, Communicator, ConnectionHandler, ConnectionOptions, Protocol, Stream,
+    Timeouts,
 };
 use core::fmt;
 use parking_lot_rt::{Mutex, MutexGuard};
@@ -46,7 +47,7 @@ pub struct Tcp {
     session_id: AtomicUsize,
     allow_reconnect: AtomicBool,
     reader_tx: Option<pchannel::Sender<CommReader>>,
-    chat: Option<Box<ChatFn>>,
+    connection_handler: Option<Box<dyn ConnectionHandler + Send + Sync>>,
 }
 
 #[allow(clippy::module_name_repetitions)]
@@ -68,6 +69,9 @@ impl Communicator for Tcp {
     fn session_id(&self) -> usize {
         self.session_id.load(Ordering::Acquire)
     }
+    fn connect(&self) -> Result<()> {
+        self.get_stream().map(|_| ())
+    }
     fn reconnect(&self) {
         self.stream
             .lock()
@@ -136,7 +140,7 @@ impl Tcp {
             session_id: <_>::default(),
             allow_reconnect: AtomicBool::new(true),
             reader_tx: tx,
-            chat: options.chat,
+            connection_handler: options.connection_handler,
         };
         Ok((client.into(), rx))
     }
@@ -160,9 +164,11 @@ impl Tcp {
                 stream.set_write_timeout(Some(self.timeouts.write))?;
             }
             stream.set_nodelay(true)?;
-            if let Some(ref chat) = self.chat {
-                trace!("chatting with the server");
-                chat(&mut stream).map_err(Error::io)?;
+            if let Some(ref connection_handler) = self.connection_handler {
+                trace!("starting connection handler");
+                connection_handler
+                    .on_connect(&mut stream)
+                    .map_err(Error::io)?;
             }
             self.session_id.fetch_add(1, Ordering::Release);
             trace!(addr=%self.addr, session_id=self.session_id(), "TCP session started");