Refactor to allow configurable network adapters

Signed-off-by: Konrad Gräfe <[email protected]>

Author: kgraefe

src/adapters/framed_tcp.rs

@@ -2,7 +2,7 @@ use crate::network::adapter::{
     Resource, Remote, Local, Adapter, SendStatus, AcceptedType, ReadStatus, ConnectionInfo,
     ListeningInfo, PendingStatus,
 };
-use crate::network::{RemoteAddr, Readiness};
+use crate::network::{RemoteAddr, Readiness, TransportConnect, TransportListen};
 use crate::util::encoding::{self, Decoder, MAX_ENCODED_SIZE};
 
 use mio::net::{TcpListener, TcpStream};
@@ -45,7 +45,10 @@ impl Resource for RemoteResource {
 }
 
 impl Remote for RemoteResource {
-    fn connect(remote_addr: RemoteAddr) -> io::Result<ConnectionInfo<Self>> {
+    fn connect_with(
+        _: TransportConnect,
+        remote_addr: RemoteAddr,
+    ) -> io::Result<ConnectionInfo<Self>> {
         let peer_addr = *remote_addr.socket_addr();
         let stream = TcpStream::connect(peer_addr)?;
         let local_addr = stream.local_addr()?;
@@ -129,7 +132,7 @@ impl Resource for LocalResource {
 impl Local for LocalResource {
     type Remote = RemoteResource;
 
-    fn listen(addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
+    fn listen_with(_: TransportListen, addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
         let listener = TcpListener::bind(addr)?;
         let local_addr = listener.local_addr().unwrap();
         Ok(ListeningInfo { local: { LocalResource { listener } }, local_addr })

src/adapters/tcp.rs

@@ -2,7 +2,7 @@ use crate::network::adapter::{
     Resource, Remote, Local, Adapter, SendStatus, AcceptedType, ReadStatus, ConnectionInfo,
     ListeningInfo, PendingStatus,
 };
-use crate::network::{RemoteAddr, Readiness};
+use crate::network::{RemoteAddr, Readiness, TransportConnect, TransportListen};
 
 use mio::net::{TcpListener, TcpStream};
 use mio::event::{Source};
@@ -40,7 +40,10 @@ impl Resource for RemoteResource {
 }
 
 impl Remote for RemoteResource {
-    fn connect(remote_addr: RemoteAddr) -> io::Result<ConnectionInfo<Self>> {
+    fn connect_with(
+        _: TransportConnect,
+        remote_addr: RemoteAddr,
+    ) -> io::Result<ConnectionInfo<Self>> {
         let peer_addr = *remote_addr.socket_addr();
         let stream = TcpStream::connect(peer_addr)?;
         let local_addr = stream.local_addr()?;
@@ -131,7 +134,7 @@ impl Resource for LocalResource {
 impl Local for LocalResource {
     type Remote = RemoteResource;
 
-    fn listen(addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
+    fn listen_with(_: TransportListen, addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
         let listener = TcpListener::bind(addr)?;
         let local_addr = listener.local_addr().unwrap();
         Ok(ListeningInfo { local: { LocalResource { listener } }, local_addr })

src/adapters/template.rs

@@ -4,7 +4,7 @@ use crate::network::adapter::{
     Resource, Remote, Local, Adapter, SendStatus, AcceptedType, ReadStatus, ConnectionInfo,
     ListeningInfo, PendingStatus,
 };
-use crate::network::{RemoteAddr, Readiness};
+use crate::network::{RemoteAddr, Readiness, TransportConnect, TransportListen};
 
 use mio::event::{Source};
 
@@ -25,7 +25,10 @@ impl Resource for RemoteResource {
 }
 
 impl Remote for RemoteResource {
-    fn connect(remote_addr: RemoteAddr) -> io::Result<ConnectionInfo<Self>> {
+    fn connect_with(
+        config: TransportConnect,
+        remote_addr: RemoteAddr,
+    ) -> io::Result<ConnectionInfo<Self>> {
         todo!()
     }
 
@@ -52,7 +55,7 @@ impl Resource for LocalResource {
 impl Local for LocalResource {
     type Remote = RemoteResource;
 
-    fn listen(addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
+    fn listen_with(config: TransportListen, addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
         todo!()
     }
 

src/adapters/udp.rs

@@ -2,7 +2,7 @@ use crate::network::adapter::{
     Resource, Remote, Local, Adapter, SendStatus, AcceptedType, ReadStatus, ConnectionInfo,
     ListeningInfo, PendingStatus,
 };
-use crate::network::{RemoteAddr, Readiness};
+use crate::network::{RemoteAddr, Readiness, TransportConnect, TransportListen};
 
 use mio::net::{UdpSocket};
 use mio::event::{Source};
@@ -43,7 +43,10 @@ impl Resource for RemoteResource {
 }
 
 impl Remote for RemoteResource {
-    fn connect(remote_addr: RemoteAddr) -> io::Result<ConnectionInfo<Self>> {
+    fn connect_with(
+        _: TransportConnect,
+        remote_addr: RemoteAddr,
+    ) -> io::Result<ConnectionInfo<Self>> {
         let socket = UdpSocket::bind("0.0.0.0:0".parse().unwrap())?;
         let peer_addr = *remote_addr.socket_addr();
         socket.connect(peer_addr)?;
@@ -95,7 +98,7 @@ impl Resource for LocalResource {
 impl Local for LocalResource {
     type Remote = RemoteResource;
 
-    fn listen(addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
+    fn listen_with(_: TransportListen, addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
         let socket = match addr {
             SocketAddr::V4(addr) if addr.ip().is_multicast() => {
                 let listening_addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, addr.port());

src/adapters/ws.rs

@@ -4,6 +4,7 @@ use crate::network::adapter::{
 };
 use crate::network::{RemoteAddr, Readiness};
 use crate::util::thread::{OTHER_THREAD_ERR};
+use crate::network::{TransportConnect, TransportListen};
 
 use mio::event::{Source};
 use mio::net::{TcpStream, TcpListener};
@@ -76,7 +77,10 @@ impl Resource for RemoteResource {
 }
 
 impl Remote for RemoteResource {
-    fn connect(remote_addr: RemoteAddr) -> io::Result<ConnectionInfo<Self>> {
+    fn connect_with(
+        _: TransportConnect,
+        remote_addr: RemoteAddr,
+    ) -> io::Result<ConnectionInfo<Self>> {
         let (peer_addr, url) = match remote_addr {
             RemoteAddr::Socket(addr) => {
                 (addr, Url::parse(&format!("ws://{addr}/message-io-default")).unwrap())
@@ -328,7 +332,7 @@ impl Resource for LocalResource {
 impl Local for LocalResource {
     type Remote = RemoteResource;
 
-    fn listen(addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
+    fn listen_with(_: TransportListen, addr: SocketAddr) -> io::Result<ListeningInfo<Self>> {
         let listener = TcpListener::bind(addr)?;
         let local_addr = listener.local_addr().unwrap();
         Ok(ListeningInfo { local: LocalResource { listener }, local_addr })

src/network.rs

@@ -17,7 +17,7 @@ pub use adapter::{SendStatus};
 pub use resource_id::{ResourceId, ResourceType};
 pub use endpoint::{Endpoint};
 pub use remote_addr::{RemoteAddr, ToRemoteAddr};
-pub use transport::{Transport};
+pub use transport::{Transport, TransportConnect, TransportListen};
 pub use driver::{NetEvent};
 pub use poll::{Readiness};
 
@@ -101,12 +101,33 @@ impl NetworkController {
         &self,
         transport: Transport,
         addr: impl ToRemoteAddr,
+    ) -> io::Result<(Endpoint, SocketAddr)> {
+        self.connect_with(transport.into(), addr)
+    }
+
+    /// Creates a connection to the specified address with custom transport options for transports
+    /// that support it.
+    /// The endpoint, an identifier of the new connection, will be returned.
+    /// This function will generate a [`NetEvent::Connected`] event with the result of the
+    /// connection.  This call will **NOT** block to perform the connection.
+    ///
+    /// Note that this function can return an error in the case the internal socket
+    /// could not be binded or open in the OS, but never will return an error regarding
+    /// the connection itself.
+    /// If you want to check if the connection has been established or not you have to read the
+    /// boolean indicator in the [`NetEvent::Connected`] event.
+    pub fn connect_with(
+        &self,
+        transport_connect: TransportConnect,
+        addr: impl ToRemoteAddr,
     ) -> io::Result<(Endpoint, SocketAddr)> {
         let addr = addr.to_remote_addr().unwrap();
-        self.controllers[transport.id() as usize].connect(addr).map(|(endpoint, addr)| {
-            log::trace!("Connect to {}", endpoint);
-            (endpoint, addr)
-        })
+        self.controllers[transport_connect.id() as usize].connect_with(transport_connect, addr).map(
+            |(endpoint, addr)| {
+                log::trace!("Connect to {}", endpoint);
+                (endpoint, addr)
+            },
+        )
     }
 
     /// Creates a connection to the specified address.
@@ -147,7 +168,28 @@ impl NetworkController {
         transport: Transport,
         addr: impl ToRemoteAddr,
     ) -> io::Result<(Endpoint, SocketAddr)> {
-        let (endpoint, addr) = self.connect(transport, addr)?;
+        self.connect_sync_with(transport.into(), addr)
+    }
+
+    /// Creates a connection to the specified address with custom transport options for transports
+    /// that support it.
+    /// This function is similar to [`NetworkController::connect_with()`] but will block
+    /// until for the connection is ready.
+    /// If the connection can not be established, a `ConnectionRefused` error will be returned.
+    ///
+    /// Note that the `Connect` event will be also generated.
+    ///
+    /// Since this function blocks the current thread, it must NOT be used inside
+    /// the network callback because the internal event could not be processed.
+    ///
+    /// In order to get the best scalability and performance, use the non-blocking
+    /// [`NetworkController::connect_with()`] version.
+    pub fn connect_sync_with(
+        &self,
+        transport_connect: TransportConnect,
+        addr: impl ToRemoteAddr,
+    ) -> io::Result<(Endpoint, SocketAddr)> {
+        let (endpoint, addr) = self.connect_with(transport_connect, addr)?;
         loop {
             std::thread::sleep(Duration::from_millis(1));
             match self.is_ready(endpoint.resource_id()) {
@@ -164,7 +206,7 @@ impl NetworkController {
     }
 
     /// Listen messages from specified transport.
-    /// The giver address will be used as interface and listening port.
+    /// The given address will be used as interface and listening port.
     /// If the port can be opened, a [ResourceId] identifying the listener is returned
     /// along with the local address, or an error if not.
     /// The address is returned despite you passed as parameter because
@@ -173,12 +215,29 @@ impl NetworkController {
         &self,
         transport: Transport,
         addr: impl ToSocketAddrs,
+    ) -> io::Result<(ResourceId, SocketAddr)> {
+        self.listen_with(transport.into(), addr)
+    }
+
+    /// Listen messages from specified transport with custom transport options for transports that
+    /// support it.
+    /// The given address will be used as interface and listening port.
+    /// If the port can be opened, a [ResourceId] identifying the listener is returned
+    /// along with the local address, or an error if not.
+    /// The address is returned despite you passed as parameter because
+    /// when a `0` port is specified, the OS will give choose the value.
+    pub fn listen_with(
+        &self,
+        transport_listen: TransportListen,
+        addr: impl ToSocketAddrs,
     ) -> io::Result<(ResourceId, SocketAddr)> {
         let addr = addr.to_socket_addrs().unwrap().next().unwrap();
-        self.controllers[transport.id() as usize].listen(addr).map(|(resource_id, addr)| {
-            log::trace!("Listening at {} by {}", addr, resource_id);
-            (resource_id, addr)
-        })
+        self.controllers[transport_listen.id() as usize].listen_with(transport_listen, addr).map(
+            |(resource_id, addr)| {
+                log::trace!("Listening at {} by {}", addr, resource_id);
+                (resource_id, addr)
+            },
+        )
     }
 
     /// Send the data message thought the connection represented by the given endpoint.

src/network/adapter.rs

@@ -1,3 +1,5 @@
+use crate::network::transport::{TransportConnect, TransportListen};
+
 use super::remote_addr::{RemoteAddr};
 use super::poll::{Readiness};
 
@@ -31,7 +33,7 @@ pub trait Resource: Send + Sync {
     fn source(&mut self) -> &mut dyn Source;
 }
 
-/// Plain struct used as a returned value of [`Remote::connect()`]
+/// Plain struct used as a returned value of [`Remote::connect_with()`]
 pub struct ConnectionInfo<R: Remote> {
     /// The new created remote resource
     pub remote: R,
@@ -43,7 +45,7 @@ pub struct ConnectionInfo<R: Remote> {
     pub peer_addr: SocketAddr,
 }
 
-/// Plain struct used as a returned value of [`Local::listen()`]
+/// Plain struct used as a returned value of [`Local::listen_with()`]
 pub struct ListeningInfo<L: Local> {
     /// The new created local resource
     pub local: L,
@@ -117,10 +119,16 @@ pub enum PendingStatus {
 pub trait Remote: Resource + Sized {
     /// Called when the user performs a connection request to an specific remote address.
     /// The **implementator** is in change of creating the corresponding remote resource.
+    /// The [`TransportConnect`] wraps custom transport options for transports that support it. It
+    /// is guaranteed by the upper level to be of the variant matching the adapter. Therefore other
+    /// variants can be safely ignored.
     /// The [`RemoteAddr`] contains either a [`SocketAddr`] or a [`url::Url`].
     /// It is in charge of deciding what to do in both cases.
     /// It also must return the extracted address as `SocketAddr`.
-    fn connect(remote_addr: RemoteAddr) -> io::Result<ConnectionInfo<Self>>;
+    fn connect_with(
+        config: TransportConnect,
+        remote_addr: RemoteAddr,
+    ) -> io::Result<ConnectionInfo<Self>>;
 
     /// Called when a remote resource received an event.
     /// The resource must be *ready* to receive this call.
@@ -193,7 +201,10 @@ pub trait Local: Resource + Sized {
     /// The **implementator** is in change of creating the corresponding local resource.
     /// It also must returned the listening address since it could not be the same as param `addr`
     /// (e.g. listening from port `0`).
-    fn listen(addr: SocketAddr) -> io::Result<ListeningInfo<Self>>;
+    /// The [`TransportListen`] wraps custom transport options for transports that support it. It
+    /// is guaranteed by the upper level to be of the variant matching the adapter. Therefore other
+    /// variants can be safely ignored.
+    fn listen_with(config: TransportListen, addr: SocketAddr) -> io::Result<ListeningInfo<Self>>;
 
     /// Called when a local resource received an event.
     /// It means that some resource have tried to connect.

src/network/driver.rs

@@ -4,6 +4,7 @@ use super::poll::{Poll, Readiness};
 use super::registry::{ResourceRegistry, Register};
 use super::remote_addr::{RemoteAddr};
 use super::adapter::{Adapter, Remote, Local, SendStatus, AcceptedType, ReadStatus, PendingStatus};
+use super::transport::{TransportConnect, TransportListen};
 
 use std::net::{SocketAddr};
 use std::sync::{
@@ -68,8 +69,16 @@ impl std::fmt::Debug for NetEvent<'_> {
 }
 
 pub trait ActionController: Send + Sync {
-    fn connect(&self, addr: RemoteAddr) -> io::Result<(Endpoint, SocketAddr)>;
-    fn listen(&self, addr: SocketAddr) -> io::Result<(ResourceId, SocketAddr)>;
+    fn connect_with(
+        &self,
+        config: TransportConnect,
+        addr: RemoteAddr,
+    ) -> io::Result<(Endpoint, SocketAddr)>;
+    fn listen_with(
+        &self,
+        config: TransportListen,
+        addr: SocketAddr,
+    ) -> io::Result<(ResourceId, SocketAddr)>;
     fn send(&self, endpoint: Endpoint, data: &[u8]) -> SendStatus;
     fn remove(&self, id: ResourceId) -> bool;
     fn is_ready(&self, id: ResourceId) -> Option<bool>;
@@ -136,8 +145,12 @@ impl<R: Remote, L: Local> Clone for Driver<R, L> {
 }
 
 impl<R: Remote, L: Local> ActionController for Driver<R, L> {
-    fn connect(&self, addr: RemoteAddr) -> io::Result<(Endpoint, SocketAddr)> {
-        R::connect(addr).map(|info| {
+    fn connect_with(
+        &self,
+        config: TransportConnect,
+        addr: RemoteAddr,
+    ) -> io::Result<(Endpoint, SocketAddr)> {
+        R::connect_with(config, addr).map(|info| {
             let id = self.remote_registry.register(
                 info.remote,
                 RemoteProperties::new(info.peer_addr, None),
@@ -147,8 +160,12 @@ impl<R: Remote, L: Local> ActionController for Driver<R, L> {
         })
     }
 
-    fn listen(&self, addr: SocketAddr) -> io::Result<(ResourceId, SocketAddr)> {
-        L::listen(addr).map(|info| {
+    fn listen_with(
+        &self,
+        config: TransportListen,
+        addr: SocketAddr,
+    ) -> io::Result<(ResourceId, SocketAddr)> {
+        L::listen_with(config, addr).map(|info| {
             let id = self.local_registry.register(info.local, LocalProperties, false);
             (id, info.local_addr)
         })