net: Initial transport abstraction concept.

So far, only outbound TCP connections.

Cargo.lock

@@ -1575,6 +1575,7 @@ dependencies = [
  "indexmap",
  "itertools",
  "lazy_static",
+ "libc",
  "libsqlite3-sys",
  "log",
  "native-tls",
@@ -1590,6 +1591,7 @@ dependencies = [
  "simplelog",
  "sled",
  "smol",
+ "socket2",
  "sqlx",
  "subtle",
  "termion",

Cargo.toml

@@ -33,6 +33,7 @@ members = [
 
 [dependencies]
 # Hard dependencies
+libc = "0.2.121"
 log = "0.4.16"
 thiserror = "1.0.30"
 
@@ -48,6 +49,9 @@ async-executor = {version = "1.4.1", optional = true}
 async-native-tls = {version = "0.4.0", optional = true}
 native-tls = {version = "0.2.8", optional = true}
 
+# Networking
+socket2 = {version = "0.4.4", optional = true}
+
 # Encoding
 hex = {version = "0.4.3", optional = true}
 bs58 = {version = "0.4.0", optional = true}
@@ -196,6 +200,7 @@ system = [
 
 net = [
     "fxhash",
+    "socket2",
 
     "util",
     "system",
@@ -262,6 +267,11 @@ name = "net"
 path = "example/net.rs"
 required-features = ["async-runtime", "net"]
 
+[[example]]
+name = "net2"
+path = "example/net2.rs"
+required-features = ["async-runtime", "net"]
+
 [[example]]
 name = "tx"
 path = "example/tx.rs"

example/net2.rs

@@ -0,0 +1,13 @@
+use async_std::io::WriteExt;
+use darkfi::net::transport::{TcpTransport, Transport};
+use url::Url;
+
+#[async_std::main]
+async fn main() {
+    let tcp = TcpTransport { ttl: None };
+    let url = Url::parse("tcp://127.0.0.1:5432").unwrap();
+
+    let mut socket = tcp.dial(url).unwrap().await.unwrap();
+    socket.write_all(b"ohai\n").await.unwrap();
+    // socket.flush().await;
+}

src/net/mod.rs

@@ -85,6 +85,9 @@ pub mod session;
 /// Network configuration settings.
 pub mod settings;
 
+/// Network transport implementations.
+pub mod transport;
+
 pub use acceptor::{Acceptor, AcceptorPtr};
 pub use channel::{Channel, ChannelPtr};
 pub use connector::Connector;

src/net/transport.rs

@@ -0,0 +1,29 @@
+use std::error::Error;
+
+use futures::prelude::*;
+use url::Url;
+
+pub mod tcp;
+pub use tcp::TcpTransport;
+
+pub trait Transport {
+    type Output;
+    type Error: Error;
+    type Dial: Future<Output = Result<Self::Output, Self::Error>>;
+
+    fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>>
+    where
+        Self: Sized;
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TransportError<TErr> {
+    #[error("Address not supported: {0}")]
+    AddrNotSupported(Url),
+
+    #[error("Transport IO Error: {0}")]
+    IoError(#[from] std::io::Error),
+
+    #[error("{0}")]
+    Other(TErr),
+}

src/net/transport/tcp.rs

@@ -0,0 +1,61 @@
+use std::{io, net::SocketAddr, pin::Pin};
+
+use async_std::net::TcpStream;
+use futures::prelude::*;
+use log::debug;
+use socket2::{Domain, Socket, Type};
+use url::Url;
+
+use super::{Transport, TransportError};
+
+pub struct TcpTransport {
+    pub ttl: Option<u32>,
+}
+
+impl Transport for TcpTransport {
+    type Output = TcpStream;
+    type Error = io::Error;
+    type Dial = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
+
+    fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>> {
+        if url.scheme() != "tcp" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        debug!(target: "tcptransport", "dialing {}", socket_addr);
+        Ok(Box::pin(self.do_dial(socket_addr)))
+    }
+}
+
+impl TcpTransport {
+    fn create_socket(&self, socket_addr: SocketAddr) -> io::Result<Socket> {
+        let domain = if socket_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::STREAM, Some(socket2::Protocol::TCP))?;
+
+        if socket_addr.is_ipv6() {
+            socket.set_only_v6(true)?;
+        }
+
+        if let Some(ttl) = self.ttl {
+            socket.set_ttl(ttl)?;
+        }
+
+        Ok(socket)
+    }
+
+    async fn do_dial(self, socket_addr: SocketAddr) -> Result<TcpStream, io::Error> {
+        let socket = self.create_socket(socket_addr)?;
+        socket.set_nonblocking(true)?;
+
+        match socket.connect(&socket_addr.into()) {
+            Ok(()) => {}
+            Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) => {}
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            Err(err) => return Err(err),
+        };
+
+        let stream = TcpStream::from(std::net::TcpStream::from(socket));
+        Ok(stream)
+    }
+}