Adjust organization of compat layer

Signed-off-by: Yilin Chen <[email protected]>

src/io/compat.rs

@@ -0,0 +1,43 @@
+#[cfg(feature = "futures-io")]
+mod futures;
+
+/// A compatibility layer for using non-tokio types with this crate.
+///
+/// Example:
+/// ```no_run
+/// use async_std::os::unix::net::UnixStream;
+/// use tokio_socks::{io::Compat, tcp::Socks5Stream};
+/// let socket = Compat::new(UnixStream::connect(proxy_addr)
+///     .await
+///     .map_err(Error::Io)?); // Compat<UnixStream>
+/// let conn =
+///     Socks5Stream::connect_with_password_and_socket(socket, target, username, password).await?;
+/// // Socks5Stream has implemented futures-io AsyncRead + AsyncWrite.
+/// ```
+pub struct Compat<S>(S);
+
+#[cfg(feature = "futures-io")]
+impl<S> Compat<S> {
+    pub fn new(inner: S) -> Self {
+        Compat(inner)
+    }
+
+    /// Consumes the `Compat``, returning the inner value.
+    pub fn into_inner(self) -> S {
+        self.0
+    }
+}
+
+#[cfg(feature = "futures-io")]
+impl<S> AsRef<S> for Compat<S> {
+    fn as_ref(&self) -> &S {
+        &self.0
+    }
+}
+
+#[cfg(feature = "futures-io")]
+impl<S> AsMut<S> for Compat<S> {
+    fn as_mut(&mut self) -> &mut S {
+        &mut self.0
+    }
+}

src/io/compat/futures.rs

@@ -0,0 +1,44 @@
+use super::Compat;
+use crate::io::AsyncSocket;
+use futures_io::{AsyncRead, AsyncWrite};
+use std::{
+    io::Result as IoResult,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+impl<S> AsyncSocket for Compat<S>
+where S: AsyncRead + AsyncWrite + Unpin
+{
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<IoResult<usize>> {
+        AsyncRead::poll_read(Pin::new(&mut self.0), cx, buf)
+    }
+
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<IoResult<usize>> {
+        AsyncWrite::poll_write(Pin::new(&mut self.0), cx, buf)
+    }
+}
+
+impl<S> AsyncRead for Compat<S>
+where S: AsyncRead + Unpin
+{
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<IoResult<usize>> {
+        AsyncRead::poll_read(Pin::new(&mut self.0), cx, buf)
+    }
+}
+
+impl<S> AsyncWrite for Compat<S>
+where S: AsyncWrite + Unpin
+{
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<IoResult<usize>> {
+        AsyncWrite::poll_write(Pin::new(&mut self.0), cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> {
+        AsyncWrite::poll_flush(Pin::new(&mut self.0), cx)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> {
+        AsyncWrite::poll_close(Pin::new(&mut self.0), cx)
+    }
+}

src/io/mod.rs

@@ -1,6 +1,5 @@
 //! Asynchronous I/O abstractions for sockets.
-#[cfg(feature = "futures-io")]
-mod futures;
+
 #[cfg(feature = "tokio")]
 mod tokio;
 
@@ -14,7 +13,9 @@ use std::{
 };
 
 #[cfg(feature = "futures-io")]
-pub use futures::{Compat, FuturesIoCompatExt};
+mod compat;
+#[cfg(feature = "futures-io")]
+pub use compat::Compat;
 
 /// A trait for asynchronous socket I/O.
 ///

src/lib.rs

@@ -14,9 +14,6 @@ use std::{
 
 pub use error::Error;
 
-#[cfg(not(any(feature = "tokio", feature = "futures-io")))]
-compile_error!("At least one of the features `tokio` or `futures-io` must be enabled.");
-
 pub type Result<T> = std::result::Result<T, Error>;
 
 /// A trait for objects which can be converted or resolved to one or more

tests/common/mod.rs

@@ -4,8 +4,6 @@ pub mod futures_utils;
 #[cfg(feature = "tokio")]
 pub mod tokio_utils;
 
-#[cfg(feature = "futures-io")]
-pub use tokio_socks::io::FuturesIoCompatExt as _;
 #[cfg(feature = "tokio")]
 pub use tokio_utils::*;
 

tests/long_username_password_auth.rs

@@ -1,6 +1,8 @@
 mod common;
 
 use common::*;
+#[cfg(feature = "futures-io")]
+use tokio_socks::io::Compat;
 use tokio_socks::{
     tcp::socks5::{Socks5Listener, Socks5Stream},
     Result,
@@ -62,7 +64,7 @@ fn bind_with_socket_long_username_password() -> Result<()> {
 #[test]
 fn connect_with_socket_long_username_password_futures_io() -> Result<()> {
     let runtime = futures_utils::runtime().lock().unwrap();
-    let socket = runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?.compat();
+    let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?);
     let conn = runtime.block_on(Socks5Stream::connect_with_password_and_socket(
         socket, ECHO_SERVER_ADDR, "mylonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglogin",
                                                                     "longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglongpassword"))?;
@@ -74,7 +76,7 @@ fn connect_with_socket_long_username_password_futures_io() -> Result<()> {
 fn bind_with_socket_long_username_password_futures_io() -> Result<()> {
     let bind = {
         let runtime = futures_utils::runtime().lock().unwrap();
-        let socket = runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?.compat();
+        let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?);
         runtime.block_on(Socks5Listener::bind_with_password_and_socket(
             socket,
             ECHO_SERVER_ADDR,

tests/no_auth.rs

@@ -1,6 +1,8 @@
 mod common;
 
 use common::*;
+#[cfg(feature = "futures-io")]
+use tokio_socks::io::Compat;
 use tokio_socks::{
     tcp::socks5::{Socks5Listener, Socks5Stream},
     Result,
@@ -48,7 +50,7 @@ fn bind_with_socket_no_auth() -> Result<()> {
 #[test]
 fn connect_with_socket_no_auth_futures_io() -> Result<()> {
     let runtime = futures_utils::runtime().lock().unwrap();
-    let socket = runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?.compat();
+    let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?);
     let conn = runtime.block_on(Socks5Stream::connect_with_socket(socket, ECHO_SERVER_ADDR))?;
     runtime.block_on(futures_utils::test_connect(conn))
 }
@@ -58,7 +60,7 @@ fn connect_with_socket_no_auth_futures_io() -> Result<()> {
 fn bind_with_socket_no_auth_futures_io() -> Result<()> {
     let bind = {
         let runtime = futures_utils::runtime().lock().unwrap();
-        let socket = runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?.compat();
+        let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?);
         runtime.block_on(Socks5Listener::bind_with_socket(socket, ECHO_SERVER_ADDR))
     }?;
     futures_utils::test_bind(bind)

tests/socks4_no_auth.rs

@@ -1,6 +1,8 @@
 mod common;
 
 use common::*;
+#[cfg(feature = "futures-io")]
+use tokio_socks::io::Compat;
 use tokio_socks::{
     tcp::socks4::{Socks4Listener, Socks4Stream},
     Result,
@@ -49,9 +51,7 @@ fn bind_with_socket_no_auth() -> Result<()> {
 #[test]
 fn connect_with_socket_no_auth_futures_io() -> Result<()> {
     let runtime = futures_utils::runtime().lock().unwrap();
-    let socket = runtime
-        .block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?
-        .compat();
+    let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?);
     println!("socket connected");
     let conn = runtime.block_on(Socks4Stream::connect_with_socket(socket, ECHO_SERVER_ADDR))?;
     runtime.block_on(futures_utils::test_connect(conn))
@@ -62,9 +62,7 @@ fn connect_with_socket_no_auth_futures_io() -> Result<()> {
 fn bind_with_socket_no_auth_futures_io() -> Result<()> {
     let bind = {
         let runtime = futures_utils::runtime().lock().unwrap();
-        let socket = runtime
-            .block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?
-            .compat();
+        let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?);
         runtime.block_on(Socks4Listener::bind_with_socket(socket, ECHO_SERVER_ADDR))
     }?;
     futures_utils::test_bind_socks4(bind)

tests/socks4_userid.rs

@@ -1,6 +1,8 @@
 mod common;
 
 use common::*;
+#[cfg(feature = "futures-io")]
+use tokio_socks::io::Compat;
 use tokio_socks::{tcp::socks4::*, Result};
 
 #[cfg(feature = "tokio")]
@@ -61,9 +63,7 @@ fn bind_with_socket_userid() -> Result<()> {
 #[test]
 fn connect_with_socket_userid_futures_io() -> Result<()> {
     let runtime = futures_utils::runtime().lock().unwrap();
-    let socket = runtime
-        .block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?
-        .compat();
+    let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?);
     let conn = runtime.block_on(Socks4Stream::connect_with_userid_and_socket(
         socket,
         ECHO_SERVER_ADDR,
@@ -77,9 +77,7 @@ fn connect_with_socket_userid_futures_io() -> Result<()> {
 fn bind_with_socket_userid_futures_io() -> Result<()> {
     let bind = {
         let runtime = futures_utils::runtime().lock().unwrap();
-        let socket = runtime
-            .block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?
-            .compat();
+        let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_SOCKS4_PROXY_ADDR))?);
         runtime.block_on(Socks4Listener::bind_with_user_and_socket(
             socket,
             ECHO_SERVER_ADDR,

tests/username_auth.rs

@@ -1,6 +1,8 @@
 mod common;
 
 use common::*;
+#[cfg(feature = "futures-io")]
+use tokio_socks::io::Compat;
 use tokio_socks::{
     tcp::socks5::{Socks5Listener, Socks5Stream},
     Result,
@@ -68,7 +70,7 @@ fn bind_with_socket_username_auth() -> Result<()> {
 #[test]
 fn connect_with_socket_username_auth_futures_io() -> Result<()> {
     let runtime = futures_utils::runtime().lock().unwrap();
-    let socket = runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?.compat();
+    let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?);
     let conn = runtime.block_on(Socks5Stream::connect_with_password_and_socket(
         socket,
         ECHO_SERVER_ADDR,
@@ -83,7 +85,7 @@ fn connect_with_socket_username_auth_futures_io() -> Result<()> {
 fn bind_with_socket_username_auth_futures_io() -> Result<()> {
     let bind = {
         let runtime = futures_utils::runtime().lock().unwrap();
-        let socket = runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?.compat();
+        let socket = Compat::new(runtime.block_on(futures_utils::connect_unix(UNIX_PROXY_ADDR))?);
         runtime.block_on(Socks5Listener::bind_with_password_and_socket(
             socket,
             ECHO_SERVER_ADDR,