Configurable network adapters

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};
@@ -26,6 +26,20 @@ pub const MAX_LOCAL_PAYLOAD_LEN: usize = 65535 - 20 - 8;
 #[cfg(target_os = "macos")]
 pub const MAX_LOCAL_PAYLOAD_LEN: usize = 9216 - 20 - 8;
 
+#[derive(Clone, PartialEq, Eq, Hash, Debug, Default)]
+pub struct UdpConnectConfig {
+    /// Enables the socket capabilities to send broadcast messages.
+    pub broadcast: bool,
+}
+
+#[derive(Clone, PartialEq, Eq, Hash, Debug, Default)]
+pub struct UdpListenConfig {
+    /// Enables the socket capabilities to send broadcast messages when the listening socket is
+    /// also used for sending with
+    /// [`Endpoint::from_listener`](crate::network::Endpoint::from_listener).
+    pub broadcast: bool,
+}
+
 pub(crate) struct UdpAdapter;
 impl Adapter for UdpAdapter {
     type Remote = RemoteResource;
@@ -43,9 +57,22 @@ 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>> {
+        let config = match config {
+            TransportConnect::Udp(config) => config,
+            _ => panic!("Internal error: Got wrong config"),
+        };
+
         let socket = UdpSocket::bind("0.0.0.0:0".parse().unwrap())?;
         let peer_addr = *remote_addr.socket_addr();
+
+        if config.broadcast {
+            socket.set_broadcast(true)?;
+        }
+
         socket.connect(peer_addr)?;
         let local_addr = socket.local_addr()?;
         Ok(ConnectionInfo { remote: RemoteResource { socket }, local_addr, peer_addr })
@@ -95,7 +122,12 @@ 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>> {
+        let config = match config {
+            TransportListen::Udp(config) => config,
+            _ => panic!("Internal error: Got wrong config"),
+        };
+
         let socket = match addr {
             SocketAddr::V4(addr) if addr.ip().is_multicast() => {
                 let listening_addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, addr.port());
@@ -112,6 +144,10 @@ impl Local for LocalResource {
             _ => UdpSocket::bind(addr)?,
         };
 
+        if config.broadcast {
+            socket.set_broadcast(true)?;
+        }
+
         let local_addr = socket.local_addr().unwrap();
         Ok(ListeningInfo { local: { LocalResource { socket } }, local_addr })
     }

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,65 @@ 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.
+    ///
+    /// Example
+    /// ```
+    /// use message_io::node::{self, NodeEvent};
+    /// use message_io::network::{TransportConnect, NetEvent};
+    /// use message_io::adapters::udp::{UdpConnectConfig};
+    ///
+    /// let (handler, listener) = node::split();
+    /// handler.signals().send_with_timer((), std::time::Duration::from_secs(1));
+    ///
+    /// let config = UdpConnectConfig { broadcast: true };
+    /// let addr = "255.255.255.255:7777";
+    /// let (conn_endpoint, _) = handler.network().connect_with(TransportConnect::Udp(config), addr).unwrap();
+    /// // The socket could not be able to send yet.
+    ///
+    /// listener.for_each(move |event| match event {
+    ///     NodeEvent::Network(net_event) => match net_event {
+    ///         NetEvent::Connected(endpoint, established) => {
+    ///             assert_eq!(conn_endpoint, endpoint);
+    ///             if established {
+    ///                 println!("Connected!");
+    ///                 handler.network().send(endpoint, &[42]);
+    ///             }
+    ///             else {
+    ///                 println!("Could not connect");
+    ///             }
+    ///         },
+    ///         _ => (),
+    ///     }
+    ///     NodeEvent::Signal(_) => handler.stop(),
+    /// });
+    /// ```
+    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 +200,51 @@ 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.
+    ///
+    /// Example
+    /// ```
+    /// use message_io::node::{self, NodeEvent};
+    /// use message_io::network::{TransportConnect, NetEvent};
+    /// use message_io::adapters::udp::{UdpConnectConfig};
+    ///
+    /// let (handler, listener) = node::split();
+    /// handler.signals().send_with_timer((), std::time::Duration::from_secs(1));
+    ///
+    /// let config = UdpConnectConfig { broadcast: true };
+    /// let addr = "255.255.255.255:7777";
+    /// match handler.network().connect_sync_with(TransportConnect::Udp(config), addr) {
+    ///     Ok((endpoint, _)) => {
+    ///         println!("Connected!");
+    ///         handler.network().send(endpoint, &[42]);
+    ///     }
+    ///     Err(err) if err.kind() == std::io::ErrorKind::ConnectionRefused => {
+    ///         println!("Could not connect");
+    ///     }
+    ///     Err(err) => println!("An OS error creating the socket"),
+    /// }
+    /// ```
+    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 +261,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 +270,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.