Let async_runtime and tokio_runtime reuse socks5 establish

Signed-off-by: Eval EXEC <[email protected]>

tentacle/Cargo.toml

@@ -44,14 +44,13 @@ igd = { version = "0.15", optional = true, package = "igd-next" }
 
 #tls
 tokio-rustls = { version = "0.26.0", optional = true }
-shadowsocks-service = { version = "1.21.2", features = ["local"]}
-shadowsocks = "1.21.0"
 url = "2.5.4"
 
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
 # rand 0.8 not support wasm32
 rand = "0.8"
 socket2 = { version = "0.5.0", features = ["all"] }
+shadowsocks = { version = "1.21.0", default-features = false }
 
 [target.'cfg(target_family = "wasm")'.dependencies]
 js-sys = "0.3"

tentacle/src/runtime/async_runtime/os.rs → tentacle/src/runtime/async_runtime/os/mod.rs

@@ -1,3 +1,5 @@
+mod time;
+
 use crate::{
     runtime::{proxy::socks5_config, CompatStream2},
     service::{
@@ -18,13 +20,11 @@ use futures::{
 use multiaddr::MultiAddr;
 use std::{
     pin::Pin,
+    str::FromStr,
     task::{Context, Poll},
 };
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
-#[cfg(feature = "async-timer")]
-mod time;
-
 #[derive(Debug)]
 pub struct TcpListener {
     /// Why does this need to be handled here?
@@ -158,9 +158,9 @@ pub(crate) fn listen(addr: SocketAddr, tcp_config: TcpSocketConfig) -> io::Resul
     Ok(TcpListener::new(AsyncListener::from(listen), addr))
 }
 
-pub(crate) async fn connect(
+async fn connect_direct(
     addr: SocketAddr,
-    tcp_config: TcpSocketConfig,
+    tcp_socket_transformer: TcpSocketTransformer,
 ) -> io::Result<TcpStream> {
     let domain = Domain::for_address(addr);
     let socket = Socket::new(domain, Type::STREAM, Some(SocketProtocol::TCP))?;
@@ -177,7 +177,7 @@ pub(crate) async fn connect(
         // user can disable it on tcp_config
         #[cfg(not(windows))]
         socket.set_reuse_address(true)?;
-        let t = (tcp_config.tcp_socket_config)(TcpSocket { inner: socket })?;
+        let t = tcp_socket_transformer(TcpSocket { inner: socket })?;
         t.inner
     };
 
@@ -230,9 +230,33 @@ where
         .map_err(io::Error::other)
 }
 
+pub(crate) async fn connect(
+    addr: SocketAddr,
+    tcp_config: TcpSocketConfig,
+) -> io::Result<TcpStream> {
+    let TcpSocketConfig {
+        tcp_socket_config,
+        proxy_config,
+    } = tcp_config;
+    match proxy_config {
+        Some(proxy_config) => connect_by_proxy(addr, tcp_socket_config, proxy_config).await,
+        None => connect_direct(addr, tcp_socket_config).await,
+    }
+}
+
 pub(crate) async fn connect_onion(
     onion_addr: MultiAddr,
     tcp_config: TcpSocketConfig,
 ) -> io::Result<TcpStream> {
-    todo!()
+    let TcpSocketConfig {
+        tcp_socket_config,
+        proxy_config,
+    } = tcp_config;
+    let proxy_config = proxy_config.ok_or(io::Error::other(
+        "need tor proxy server to connect to onion address",
+    ))?;
+    let onion_address = shadowsocks::relay::Address::from_str(onion_addr.to_string().as_str())
+        .map_err(std::io::Error::other)?;
+
+    connect_by_proxy(onion_address, tcp_socket_config, proxy_config).await
 }

tentacle/src/runtime/mod.rs

@@ -16,7 +16,7 @@ mod async_runtime;
     all(target_family = "wasm", feature = "wasm-timer")
 ))]
 mod generic_timer;
-pub(crate) mod socks5_config;
+pub(crate) mod proxy;
 #[cfg(all(not(target_family = "wasm"), feature = "tokio-runtime"))]
 mod tokio_runtime;
 #[cfg(target_family = "wasm")]

tentacle/src/runtime/proxy/mod.rs

@@ -0,0 +1,3 @@
+#[cfg(not(target_family = "wasm"))]
+pub(crate) mod socks5;
+pub(crate) mod socks5_config;

tentacle/src/runtime/tokio_runtime/socks5.rs → tentacle/src/runtime/proxy/socks5.rs

@@ -5,16 +5,17 @@ use shadowsocks::relay::socks5::{
     self, Address, Command, Error as Socks5Error, HandshakeRequest, HandshakeResponse,
     PasswdAuthRequest, PasswdAuthResponse, Reply, TcpRequestHeader, TcpResponseHeader,
 };
-use tokio::net::TcpStream;
+use tokio::io::{AsyncRead, AsyncWrite};
 
-use super::super::socks5_config::Socks5Config;
+use super::socks5_config::Socks5Config;
 
-pub async fn establish_connection<A>(
-    mut s: TcpStream,
+pub async fn establish_connection<S, A>(
+    mut s: S,
     target_addr: A,
     socks5_config: Socks5Config,
-) -> Result<TcpStream, Socks5Error>
+) -> Result<S, Socks5Error>
 where
+    S: AsyncRead + AsyncWrite + Unpin,
     A: Into<Address>,
 {
     debug!(

tentacle/src/runtime/tokio_runtime/mod.rs

@@ -1,5 +1,4 @@
-use super::socks5_config;
-mod socks5;
+use super::proxy::socks5_config;
 use multiaddr::MultiAddr;
 pub use tokio::{
     net::{TcpListener, TcpStream},
@@ -155,7 +154,7 @@ where
     let dial_addr: SocketAddr = socks5_config.proxy_url.parse().map_err(io::Error::other)?;
     let stream = connect_direct(dial_addr, tcp_socket_transformer).await?;
 
-    super::tokio_runtime::socks5::establish_connection(stream, target_addr, socks5_config)
+    super::proxy::socks5::establish_connection(stream, target_addr, socks5_config)
         .await
         .map_err(io::Error::other)
 }