Replace the macro in libp2p-tcp with a generic, inner implementation

To avoid code duplication, a generic inner implementation is used
that does all of the heavy lifting. To abstract over the individual
types, we create traits that are private to the module.

Consumers of the crate shouldn't be concerned with this because
a) none of the generic types are exported
b) there are no type parameters in public interfaces

Author: thomaseizinger

src/lib.rs

@@ -141,7 +141,7 @@
 //! [`Keypair`]: identity::Keypair
 //! [`PublicKey`]: identity::PublicKey
 //! [`Future`]: futures::Future
-//! [`TcpConfig`]: tcp::TcpConfig
+//! [`TcpConfig`]: tcp::async_std::TcpConfig
 //! [`NetworkBehaviour`]: swarm::NetworkBehaviour
 //! [`StreamMuxer`]: core::muxing::StreamMuxer
 //! [`Yamux`]: yamux::Yamux
@@ -282,7 +282,7 @@ struct CommonTransport {
 }
 
 #[cfg(all(not(any(target_os = "emscripten", target_os = "unknown")), feature = "libp2p-websocket"))]
-type InnerImplementation = core::transport::OrTransport<dns::DnsConfig<tcp::TcpConfig>, websocket::WsConfig<dns::DnsConfig<tcp::TcpConfig>>>;
+type InnerImplementation = core::transport::OrTransport<dns::DnsConfig<tcp::async_std::TcpConfig>, websocket::WsConfig<dns::DnsConfig<tcp::async_std::TcpConfig>>>;
 #[cfg(all(not(any(target_os = "emscripten", target_os = "unknown")), not(feature = "libp2p-websocket")))]
 type InnerImplementation = dns::DnsConfig<tcp::TcpConfig>;
 #[cfg(any(target_os = "emscripten", target_os = "unknown"))]
@@ -297,7 +297,7 @@ impl CommonTransport {
     /// Initializes the `CommonTransport`.
     #[cfg(not(any(target_os = "emscripten", target_os = "unknown")))]
     pub fn new() -> io::Result<CommonTransport> {
-        let tcp = tcp::TcpConfig::new().nodelay(true);
+        let tcp = tcp::async_std::TcpConfig::new().nodelay(true);
         let transport = dns::DnsConfig::new(tcp)?;
         #[cfg(feature = "libp2p-websocket")]
         let transport = {

transports/tcp/Cargo.toml

@@ -18,6 +18,7 @@ ipnet = "2.0.0"
 libp2p-core = { version = "0.16.0", path = "../../core" }
 log = "0.4.1"
 tokio = { version = "0.2", default-features = false, features = ["tcp"], optional = true }
+async-trait = "0.1"
 
 [features]
 default = ["async-std"]

transports/tcp/src/async_std.rs

@@ -0,0 +1,120 @@
+use crate::internal::{self, TcpListener, TcpStream, TcpTransStream};
+use futures::{AsyncRead, AsyncWrite};
+use libp2p_core::{transport::TransportError, Multiaddr, Transport};
+use std::{
+    io,
+    net::SocketAddr,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+type AsyncStdConfig = internal::TcpConfig<async_std::net::TcpListener>;
+
+/// Represents the configuration for a TCP/IP transport capability for libp2p.
+///
+/// This implementation uses the network futures from async_std.
+#[derive(Debug, Clone, Default)]
+pub struct TcpConfig {
+    inner: AsyncStdConfig,
+}
+
+impl TcpConfig {
+    /// Creates a new configuration object for TCP/IP.
+    pub fn new() -> Self {
+        Self {
+            inner: AsyncStdConfig::new(),
+        }
+    }
+
+    /// Sets the TTL to set for opened sockets.
+    pub fn ttl(mut self, value: u32) -> Self {
+        self.inner.ttl = Some(value);
+        self
+    }
+
+    /// Sets the `TCP_NODELAY` to set for opened sockets.
+    pub fn nodelay(mut self, value: bool) -> Self {
+        self.inner.nodelay = Some(value);
+        self
+    }
+}
+
+impl Transport for TcpConfig {
+    type Output = <AsyncStdConfig as Transport>::Output;
+    type Error = <AsyncStdConfig as Transport>::Error;
+    type Listener = <AsyncStdConfig as Transport>::Listener;
+    type ListenerUpgrade = <AsyncStdConfig as Transport>::ListenerUpgrade;
+    type Dial = <AsyncStdConfig as Transport>::Dial;
+
+    fn listen_on(self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>> {
+        self.inner.listen_on(addr)
+    }
+
+    fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
+        self.inner.dial(addr)
+    }
+}
+
+#[async_trait::async_trait]
+impl TcpListener for async_std::net::TcpListener {
+    type TcpStream = async_std::net::TcpStream;
+
+    async fn bind(addrs: &SocketAddr) -> io::Result<async_std::net::TcpListener> {
+        async_std::net::TcpListener::bind(addrs).await
+    }
+
+    async fn accept(&mut self) -> io::Result<(async_std::net::TcpStream, SocketAddr)> {
+        async_std::net::TcpListener::accept(self).await
+    }
+
+    fn local_addr(&self) -> io::Result<SocketAddr> {
+        async_std::net::TcpListener::local_addr(self)
+    }
+}
+
+impl AsyncRead for TcpTransStream<async_std::net::TcpStream> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        AsyncRead::poll_read(Pin::new(&mut self.inner), cx, buf)
+    }
+}
+
+impl AsyncWrite for TcpTransStream<async_std::net::TcpStream> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        AsyncWrite::poll_write(Pin::new(&mut self.inner), cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
+        AsyncWrite::poll_flush(Pin::new(&mut self.inner), cx)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
+        AsyncWrite::poll_close(Pin::new(&mut self.inner), cx)
+    }
+}
+
+#[async_trait::async_trait]
+impl TcpStream for async_std::net::TcpStream {
+    async fn connect(addrs: &SocketAddr) -> io::Result<Self> {
+        async_std::net::TcpStream::connect(addrs).await
+    }
+    fn set_ttl(&self, ttl: u32) -> io::Result<()> {
+        async_std::net::TcpStream::set_ttl(self, ttl)
+    }
+    fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
+        async_std::net::TcpStream::set_nodelay(self, nodelay)
+    }
+    fn peer_addr(&self) -> io::Result<SocketAddr> {
+        async_std::net::TcpStream::peer_addr(self)
+    }
+    fn local_addr(&self) -> io::Result<SocketAddr> {
+        async_std::net::TcpStream::local_addr(self)
+    }
+}