diff --git a/crates/shadowsocks-service/src/lib.rs b/crates/shadowsocks-service/src/lib.rs index 1cba27dd..a4b53296 100644 --- a/crates/shadowsocks-service/src/lib.rs +++ b/crates/shadowsocks-service/src/lib.rs @@ -51,7 +51,7 @@ use std::time::Duration; #[cfg(feature = "local")] -pub use self::local::{create as create_local, run as run_local}; +pub use self::local::run as run_local; #[cfg(feature = "manager")] pub use self::manager::run as run_manager; diff --git a/crates/shadowsocks-service/src/local/mod.rs b/crates/shadowsocks-service/src/local/mod.rs index 7a9ba4be..ecb0049f 100644 --- a/crates/shadowsocks-service/src/local/mod.rs +++ b/crates/shadowsocks-service/src/local/mod.rs @@ -29,6 +29,18 @@ use self::{ loadbalancing::{PingBalancer, PingBalancerBuilder}, }; +#[cfg(feature = "local-dns")] +use self::dns::{Dns, DnsBuilder}; +#[cfg(feature = "local-http")] +use self::http::{Http, HttpBuilder}; +#[cfg(feature = "local-redir")] +use self::redir::{Redir, RedirBuilder}; +use self::socks::{Socks, SocksBuilder}; +#[cfg(feature = "local-tun")] +use self::tun::{Tun, TunBuilder}; +#[cfg(feature = "local-tunnel")] +use self::tunnel::{Tunnel, TunnelBuilder}; + pub mod context; #[cfg(feature = "local-dns")] pub mod dns; @@ -73,25 +85,396 @@ impl Future for ServerHandle { /// Local Server instance pub struct Server { - vfut: Vec, balancer: PingBalancer, + socks_servers: Vec, + #[cfg(feature = "local-tunnel")] + tunnel_servers: Vec, + #[cfg(feature = "local-http")] + http_servers: Vec, + #[cfg(feature = "local-tun")] + tun_servers: Vec, + #[cfg(feature = "local-dns")] + dns_servers: Vec, + #[cfg(feature = "local-redir")] + redir_servers: Vec, + local_stat_addr: Option, + flow_stat: Arc, } impl Server { /// Create a shadowsocks local server - pub async fn create(config: Config) -> io::Result { - create(config).await + pub async fn new(config: Config) -> io::Result { + assert!(config.config_type == ConfigType::Local && !config.local.is_empty()); + + trace!("{:?}", config); + + // Warning for Stream Ciphers + #[cfg(feature = "stream-cipher")] + for inst in config.server.iter() { + let server = &inst.config; + + if server.method().is_stream() { + log::warn!("stream cipher {} for server {} have inherent weaknesses (see discussion in https://github.com/shadowsocks/shadowsocks-org/issues/36). \ + DO NOT USE. It will be removed in the future.", server.method(), server.addr()); + } + } + + #[cfg(all(unix, not(target_os = "android")))] + if let Some(nofile) = config.nofile { + use crate::sys::set_nofile; + if let Err(err) = set_nofile(nofile) { + log::warn!("set_nofile {} failed, error: {}", nofile, err); + } + } + + // Global ServiceContext template + // Each Local instance will hold a copy of its fields + let mut context = ServiceContext::new(); + + let mut connect_opts = ConnectOpts { + #[cfg(any(target_os = "linux", target_os = "android"))] + fwmark: config.outbound_fwmark, + + #[cfg(target_os = "android")] + vpn_protect_path: config.outbound_vpn_protect_path, + + bind_interface: config.outbound_bind_interface, + bind_local_addr: config.outbound_bind_addr, + + ..Default::default() + }; + connect_opts.tcp.send_buffer_size = config.outbound_send_buffer_size; + connect_opts.tcp.recv_buffer_size = config.outbound_recv_buffer_size; + connect_opts.tcp.nodelay = config.no_delay; + connect_opts.tcp.fastopen = config.fast_open; + connect_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT)); + connect_opts.tcp.mptcp = config.mptcp; + context.set_connect_opts(connect_opts); + + let mut accept_opts = AcceptOpts { + ipv6_only: config.ipv6_only, + ..Default::default() + }; + accept_opts.tcp.send_buffer_size = config.inbound_send_buffer_size; + accept_opts.tcp.recv_buffer_size = config.inbound_recv_buffer_size; + accept_opts.tcp.nodelay = config.no_delay; + accept_opts.tcp.fastopen = config.fast_open; + accept_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT)); + accept_opts.tcp.mptcp = config.mptcp; + context.set_accept_opts(accept_opts); + + if let Some(resolver) = build_dns_resolver(config.dns, config.ipv6_first, context.connect_opts_ref()).await { + context.set_dns_resolver(Arc::new(resolver)); + } + + if config.ipv6_first { + context.set_ipv6_first(config.ipv6_first); + } + + if let Some(acl) = config.acl { + context.set_acl(Arc::new(acl)); + } + + context.set_security_config(&config.security); + + assert!(!config.local.is_empty(), "no valid local server configuration"); + + // Create a service balancer for choosing between multiple servers + let balancer = { + let mut mode = Mode::TcpOnly; + + for local in &config.local { + mode = mode.merge(local.config.mode); + } + + // Load balancer will hold an individual ServiceContext + let mut balancer_builder = PingBalancerBuilder::new(Arc::new(context.clone()), mode); + + // max_server_rtt have to be set before add_server + if let Some(rtt) = config.balancer.max_server_rtt { + balancer_builder.max_server_rtt(rtt); + } + + if let Some(intv) = config.balancer.check_interval { + balancer_builder.check_interval(intv); + } + + if let Some(intv) = config.balancer.check_best_interval { + balancer_builder.check_best_interval(intv); + } + + for server in config.server { + balancer_builder.add_server(server.config); + } + + balancer_builder.build().await? + }; + + let mut local_server = Server { + balancer: balancer.clone(), + socks_servers: Vec::new(), + #[cfg(feature = "local-tunnel")] + tunnel_servers: Vec::new(), + #[cfg(feature = "local-http")] + http_servers: Vec::new(), + #[cfg(feature = "local-tun")] + tun_servers: Vec::new(), + #[cfg(feature = "local-dns")] + dns_servers: Vec::new(), + #[cfg(feature = "local-redir")] + redir_servers: Vec::new(), + local_stat_addr: config.local_stat_addr, + flow_stat: context.flow_stat(), + }; + + for local_instance in config.local { + let local_config = local_instance.config; + + // Clone from global ServiceContext instance + // It will shares Shadowsocks' global context, and FlowStat, DNS reverse cache + let mut context = context.clone(); + + // Private ACL + if let Some(acl) = local_instance.acl { + context.set_acl(Arc::new(acl)) + } + + let context = Arc::new(context); + let balancer = balancer.clone(); + + match local_config.protocol { + ProtocolType::Socks => { + let client_addr = match local_config.addr { + Some(a) => a, + None => return Err(io::Error::new(ErrorKind::Other, "socks requires local address")), + }; + + let mut server_builder = SocksBuilder::with_context(context.clone(), client_addr, balancer); + server_builder.set_mode(local_config.mode); + server_builder.set_socks5_auth(local_config.socks5_auth); + + if let Some(c) = config.udp_max_associations { + server_builder.set_udp_capacity(c); + } + if let Some(d) = config.udp_timeout { + server_builder.set_udp_expiry_duration(d); + } + if let Some(b) = local_config.udp_addr { + server_builder.set_udp_bind_addr(b.clone()); + } + + let server = server_builder.build().await?; + local_server.socks_servers.push(server); + } + #[cfg(feature = "local-tunnel")] + ProtocolType::Tunnel => { + let client_addr = match local_config.addr { + Some(a) => a, + None => return Err(io::Error::new(ErrorKind::Other, "tunnel requires local address")), + }; + + let forward_addr = local_config.forward_addr.expect("tunnel requires forward address"); + + let mut server_builder = + TunnelBuilder::with_context(context.clone(), forward_addr.clone(), client_addr, balancer); + + if let Some(c) = config.udp_max_associations { + server_builder.set_udp_capacity(c); + } + if let Some(d) = config.udp_timeout { + server_builder.set_udp_expiry_duration(d); + } + server_builder.set_mode(local_config.mode); + if let Some(udp_addr) = local_config.udp_addr { + server_builder.set_udp_bind_addr(udp_addr); + } + + let server = server_builder.build().await?; + local_server.tunnel_servers.push(server); + } + #[cfg(feature = "local-http")] + ProtocolType::Http => { + let client_addr = match local_config.addr { + Some(a) => a, + None => return Err(io::Error::new(ErrorKind::Other, "http requires local address")), + }; + + let builder = HttpBuilder::with_context(context.clone(), client_addr, balancer); + let server = builder.build().await?; + local_server.http_servers.push(server); + } + #[cfg(feature = "local-redir")] + ProtocolType::Redir => { + let client_addr = match local_config.addr { + Some(a) => a, + None => return Err(io::Error::new(ErrorKind::Other, "redir requires local address")), + }; + + let mut server_builder = RedirBuilder::with_context(context.clone(), client_addr, balancer); + if let Some(c) = config.udp_max_associations { + server_builder.set_udp_capacity(c); + } + if let Some(d) = config.udp_timeout { + server_builder.set_udp_expiry_duration(d); + } + server_builder.set_mode(local_config.mode); + server_builder.set_tcp_redir(local_config.tcp_redir); + server_builder.set_udp_redir(local_config.udp_redir); + if let Some(udp_addr) = local_config.udp_addr { + server_builder.set_udp_bind_addr(udp_addr); + } + + let server = server_builder.build().await?; + local_server.redir_servers.push(server); + } + #[cfg(feature = "local-dns")] + ProtocolType::Dns => { + let client_addr = match local_config.addr { + Some(a) => a, + None => return Err(io::Error::new(ErrorKind::Other, "dns requires local address")), + }; + + let mut server_builder = { + let local_addr = local_config.local_dns_addr.expect("missing local_dns_addr"); + let remote_addr = local_config.remote_dns_addr.expect("missing remote_dns_addr"); + + DnsBuilder::with_context( + context.clone(), + client_addr, + local_addr.clone(), + remote_addr.clone(), + balancer, + ) + }; + server_builder.set_mode(local_config.mode); + + let server = server_builder.build().await?; + local_server.dns_servers.push(server); + } + #[cfg(feature = "local-tun")] + ProtocolType::Tun => { + use log::info; + use shadowsocks::net::UnixListener; + + let mut builder = TunBuilder::new(context.clone(), balancer); + if let Some(address) = local_config.tun_interface_address { + builder.address(address); + } + if let Some(address) = local_config.tun_interface_destination { + builder.destination(address); + } + if let Some(name) = local_config.tun_interface_name { + builder.name(&name); + } + if let Some(c) = config.udp_max_associations { + builder.udp_capacity(c); + } + if let Some(d) = config.udp_timeout { + builder.udp_expiry_duration(d); + } + builder.mode(local_config.mode); + #[cfg(unix)] + if let Some(fd) = local_config.tun_device_fd { + builder.file_descriptor(fd); + } else if let Some(ref fd_path) = local_config.tun_device_fd_from_path { + use std::fs; + + let _ = fs::remove_file(fd_path); + + let listener = match UnixListener::bind(fd_path) { + Ok(l) => l, + Err(err) => { + log::error!("failed to bind uds path \"{}\", error: {}", fd_path.display(), err); + return Err(err); + } + }; + + info!("waiting tun's file descriptor from {}", fd_path.display()); + + loop { + let (mut stream, peer_addr) = listener.accept().await?; + trace!("accepted {:?} for receiving tun file descriptor", peer_addr); + + let mut buffer = [0u8; 1024]; + let mut fd_buffer = [0]; + + match stream.recv_with_fd(&mut buffer, &mut fd_buffer).await { + Ok((n, fd_size)) => { + if fd_size == 0 { + log::error!( + "client {:?} didn't send file descriptors with buffer.size {} bytes", + peer_addr, + n + ); + continue; + } + + info!("got file descriptor {} for tun from {:?}", fd_buffer[0], peer_addr); + + builder.file_descriptor(fd_buffer[0]); + break; + } + Err(err) => { + log::error!( + "failed to receive file descriptors from {:?}, error: {}", + peer_addr, + err + ); + } + } + } + } + let server = builder.build().await?; + local_server.tun_servers.push(server); + } + } + } + + Ok(local_server) } /// Run local server - #[deprecated] pub async fn run(self) -> io::Result<()> { - self.wait_until_exit().await - } + let mut vfut = Vec::new(); - /// Wait until any of the servers were exited - pub async fn wait_until_exit(self) -> io::Result<()> { - let (res, ..) = future::select_all(self.vfut).await; + for svr in self.socks_servers { + vfut.push(ServerHandle(tokio::spawn(svr.run()))); + } + + #[cfg(feature = "local-tunnel")] + for svr in self.tunnel_servers { + vfut.push(ServerHandle(tokio::spawn(svr.run()))); + } + + #[cfg(feature = "local-http")] + for svr in self.http_servers { + vfut.push(ServerHandle(tokio::spawn(svr.run()))); + } + + #[cfg(feature = "local-tun")] + for svr in self.tun_servers { + vfut.push(ServerHandle(tokio::spawn(svr.run()))); + } + + #[cfg(feature = "local-dns")] + for svr in self.dns_servers { + vfut.push(ServerHandle(tokio::spawn(svr.run()))); + } + + #[cfg(feature = "local-redir")] + for svr in self.redir_servers { + vfut.push(ServerHandle(tokio::spawn(svr.run()))); + } + + #[cfg(feature = "local-flow-stat")] + if let Some(stat_addr) = self.local_stat_addr { + // For Android's flow statistic + + let report_fut = flow_report_task(stat_addr, self.flow_stat); + vfut.push(ServerHandle(tokio::spawn(report_fut))); + } + + let (res, ..) = future::select_all(vfut).await; res } @@ -99,342 +482,41 @@ impl Server { pub fn server_balancer(&self) -> &PingBalancer { &self.balancer } -} -/// Starts a shadowsocks local server -pub async fn create(config: Config) -> io::Result { - assert!(config.config_type == ConfigType::Local && !config.local.is_empty()); - - trace!("{:?}", config); - - // Warning for Stream Ciphers - #[cfg(feature = "stream-cipher")] - for inst in config.server.iter() { - let server = &inst.config; - - if server.method().is_stream() { - log::warn!("stream cipher {} for server {} have inherent weaknesses (see discussion in https://github.com/shadowsocks/shadowsocks-org/issues/36). \ - DO NOT USE. It will be removed in the future.", server.method(), server.addr()); - } + /// Get SOCKS server instances + pub fn socks_servers(&self) -> &[Socks] { + &self.socks_servers } - #[cfg(all(unix, not(target_os = "android")))] - if let Some(nofile) = config.nofile { - use crate::sys::set_nofile; - if let Err(err) = set_nofile(nofile) { - log::warn!("set_nofile {} failed, error: {}", nofile, err); - } + /// Get Tunnel server instances + #[cfg(feature = "local-tunnel")] + pub fn tunnel_servers(&self) -> &[Tunnel] { + &self.tunnel_servers } - // Global ServiceContext template - // Each Local instance will hold a copy of its fields - let mut context = ServiceContext::new(); - - let mut connect_opts = ConnectOpts { - #[cfg(any(target_os = "linux", target_os = "android"))] - fwmark: config.outbound_fwmark, - - #[cfg(target_os = "android")] - vpn_protect_path: config.outbound_vpn_protect_path, - - bind_interface: config.outbound_bind_interface, - bind_local_addr: config.outbound_bind_addr, - - ..Default::default() - }; - connect_opts.tcp.send_buffer_size = config.outbound_send_buffer_size; - connect_opts.tcp.recv_buffer_size = config.outbound_recv_buffer_size; - connect_opts.tcp.nodelay = config.no_delay; - connect_opts.tcp.fastopen = config.fast_open; - connect_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT)); - connect_opts.tcp.mptcp = config.mptcp; - context.set_connect_opts(connect_opts); - - let mut accept_opts = AcceptOpts { - ipv6_only: config.ipv6_only, - ..Default::default() - }; - accept_opts.tcp.send_buffer_size = config.inbound_send_buffer_size; - accept_opts.tcp.recv_buffer_size = config.inbound_recv_buffer_size; - accept_opts.tcp.nodelay = config.no_delay; - accept_opts.tcp.fastopen = config.fast_open; - accept_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT)); - accept_opts.tcp.mptcp = config.mptcp; - context.set_accept_opts(accept_opts); - - if let Some(resolver) = build_dns_resolver(config.dns, config.ipv6_first, context.connect_opts_ref()).await { - context.set_dns_resolver(Arc::new(resolver)); + /// Get HTTP server instances + #[cfg(feature = "local-http")] + pub fn http_servers(&self) -> &[Http] { + &self.http_servers } - if config.ipv6_first { - context.set_ipv6_first(config.ipv6_first); + /// Get Tun server instances + #[cfg(feature = "local-tun")] + pub fn tun_servers(&self) -> &[Tun] { + &self.tun_servers } - if let Some(acl) = config.acl { - context.set_acl(Arc::new(acl)); + /// Get DNS server instances + #[cfg(feature = "local-dns")] + pub fn dns_servers(&self) -> &[Dns] { + &self.dns_servers } - context.set_security_config(&config.security); - - assert!(!config.local.is_empty(), "no valid local server configuration"); - - let mut vfut = Vec::new(); - - // Create a service balancer for choosing between multiple servers - let balancer = { - let mut mode = Mode::TcpOnly; - - for local in &config.local { - mode = mode.merge(local.config.mode); - } - - // Load balancer will hold an individual ServiceContext - let mut balancer_builder = PingBalancerBuilder::new(Arc::new(context.clone()), mode); - - // max_server_rtt have to be set before add_server - if let Some(rtt) = config.balancer.max_server_rtt { - balancer_builder.max_server_rtt(rtt); - } - - if let Some(intv) = config.balancer.check_interval { - balancer_builder.check_interval(intv); - } - - if let Some(intv) = config.balancer.check_best_interval { - balancer_builder.check_best_interval(intv); - } - - for server in config.server { - balancer_builder.add_server(server.config); - } - - balancer_builder.build().await? - }; - - #[cfg(feature = "local-flow-stat")] - if let Some(stat_addr) = config.local_stat_addr { - // For Android's flow statistic - - let report_fut = flow_report_task(stat_addr, context.flow_stat()); - vfut.push(ServerHandle(tokio::spawn(report_fut))); + /// Get Redir server instances + #[cfg(feature = "local-redir")] + pub fn redir_servers(&self) -> &[Redir] { + &self.redir_servers } - - for local_instance in config.local { - let local_config = local_instance.config; - - // Clone from global ServiceContext instance - // It will shares Shadowsocks' global context, and FlowStat, DNS reverse cache - let mut context = context.clone(); - - // Private ACL - if let Some(acl) = local_instance.acl { - context.set_acl(Arc::new(acl)) - } - - let context = Arc::new(context); - let balancer = balancer.clone(); - - match local_config.protocol { - ProtocolType::Socks => { - use self::socks::SocksBuilder; - - let client_addr = match local_config.addr { - Some(a) => a, - None => return Err(io::Error::new(ErrorKind::Other, "socks requires local address")), - }; - - let mut server_builder = SocksBuilder::with_context(context.clone(), client_addr, balancer); - server_builder.set_mode(local_config.mode); - server_builder.set_socks5_auth(local_config.socks5_auth); - - if let Some(c) = config.udp_max_associations { - server_builder.set_udp_capacity(c); - } - if let Some(d) = config.udp_timeout { - server_builder.set_udp_expiry_duration(d); - } - if let Some(b) = local_config.udp_addr { - server_builder.set_udp_bind_addr(b.clone()); - } - - let server = server_builder.build().await?; - vfut.push(ServerHandle(tokio::spawn(async move { server.run().await }))); - } - #[cfg(feature = "local-tunnel")] - ProtocolType::Tunnel => { - use self::tunnel::TunnelBuilder; - - let client_addr = match local_config.addr { - Some(a) => a, - None => return Err(io::Error::new(ErrorKind::Other, "tunnel requires local address")), - }; - - let forward_addr = local_config.forward_addr.expect("tunnel requires forward address"); - - let mut server_builder = - TunnelBuilder::with_context(context.clone(), forward_addr.clone(), client_addr, balancer); - - if let Some(c) = config.udp_max_associations { - server_builder.set_udp_capacity(c); - } - if let Some(d) = config.udp_timeout { - server_builder.set_udp_expiry_duration(d); - } - server_builder.set_mode(local_config.mode); - if let Some(udp_addr) = local_config.udp_addr { - server_builder.set_udp_bind_addr(udp_addr); - } - - let server = server_builder.build().await?; - vfut.push(ServerHandle(tokio::spawn(async move { server.run().await }))); - } - #[cfg(feature = "local-http")] - ProtocolType::Http => { - use self::http::HttpBuilder; - - let client_addr = match local_config.addr { - Some(a) => a, - None => return Err(io::Error::new(ErrorKind::Other, "http requires local address")), - }; - - let builder = HttpBuilder::with_context(context.clone(), client_addr, balancer); - let server = builder.build().await?; - vfut.push(ServerHandle(tokio::spawn(async move { server.run().await }))); - } - #[cfg(feature = "local-redir")] - ProtocolType::Redir => { - use self::redir::RedirBuilder; - - let client_addr = match local_config.addr { - Some(a) => a, - None => return Err(io::Error::new(ErrorKind::Other, "redir requires local address")), - }; - - let mut server_builder = RedirBuilder::with_context(context.clone(), client_addr, balancer); - if let Some(c) = config.udp_max_associations { - server_builder.set_udp_capacity(c); - } - if let Some(d) = config.udp_timeout { - server_builder.set_udp_expiry_duration(d); - } - server_builder.set_mode(local_config.mode); - server_builder.set_tcp_redir(local_config.tcp_redir); - server_builder.set_udp_redir(local_config.udp_redir); - if let Some(udp_addr) = local_config.udp_addr { - server_builder.set_udp_bind_addr(udp_addr); - } - - let server = server_builder.build().await?; - vfut.push(ServerHandle(tokio::spawn(async move { server.run().await }))); - } - #[cfg(feature = "local-dns")] - ProtocolType::Dns => { - use self::dns::DnsBuilder; - - let client_addr = match local_config.addr { - Some(a) => a, - None => return Err(io::Error::new(ErrorKind::Other, "dns requires local address")), - }; - - let mut server_builder = { - let local_addr = local_config.local_dns_addr.expect("missing local_dns_addr"); - let remote_addr = local_config.remote_dns_addr.expect("missing remote_dns_addr"); - - DnsBuilder::with_context( - context.clone(), - client_addr, - local_addr.clone(), - remote_addr.clone(), - balancer, - ) - }; - server_builder.set_mode(local_config.mode); - - let server = server_builder.build().await?; - vfut.push(ServerHandle(tokio::spawn(async move { server.run().await }))); - } - #[cfg(feature = "local-tun")] - ProtocolType::Tun => { - use log::info; - use shadowsocks::net::UnixListener; - - use self::tun::TunBuilder; - - let mut builder = TunBuilder::new(context.clone(), balancer); - if let Some(address) = local_config.tun_interface_address { - builder.address(address); - } - if let Some(address) = local_config.tun_interface_destination { - builder.destination(address); - } - if let Some(name) = local_config.tun_interface_name { - builder.name(&name); - } - if let Some(c) = config.udp_max_associations { - builder.udp_capacity(c); - } - if let Some(d) = config.udp_timeout { - builder.udp_expiry_duration(d); - } - builder.mode(local_config.mode); - #[cfg(unix)] - if let Some(fd) = local_config.tun_device_fd { - builder.file_descriptor(fd); - } else if let Some(ref fd_path) = local_config.tun_device_fd_from_path { - use std::fs; - - let _ = fs::remove_file(fd_path); - - let listener = match UnixListener::bind(fd_path) { - Ok(l) => l, - Err(err) => { - log::error!("failed to bind uds path \"{}\", error: {}", fd_path.display(), err); - return Err(err); - } - }; - - info!("waiting tun's file descriptor from {}", fd_path.display()); - - loop { - let (mut stream, peer_addr) = listener.accept().await?; - trace!("accepted {:?} for receiving tun file descriptor", peer_addr); - - let mut buffer = [0u8; 1024]; - let mut fd_buffer = [0]; - - match stream.recv_with_fd(&mut buffer, &mut fd_buffer).await { - Ok((n, fd_size)) => { - if fd_size == 0 { - log::error!( - "client {:?} didn't send file descriptors with buffer.size {} bytes", - peer_addr, - n - ); - continue; - } - - info!("got file descriptor {} for tun from {:?}", fd_buffer[0], peer_addr); - - builder.file_descriptor(fd_buffer[0]); - break; - } - Err(err) => { - log::error!( - "failed to receive file descriptors from {:?}, error: {}", - peer_addr, - err - ); - } - } - } - } - let server = builder.build().await?; - vfut.push(ServerHandle(tokio::spawn(async move { server.run().await }))); - } - } - } - - Ok(Server { vfut, balancer }) } #[cfg(feature = "local-flow-stat")] @@ -515,5 +597,5 @@ async fn flow_report_task(stat_addr: LocalFlowStatAddress, flow_stat: Arc io::Result<()> { - create(config).await?.wait_until_exit().await + Server::new(config).await?.run().await } diff --git a/src/service/local.rs b/src/service/local.rs index 0aa2ae5a..e1bc6693 100644 --- a/src/service/local.rs +++ b/src/service/local.rs @@ -22,8 +22,7 @@ use shadowsocks_service::{ ProtocolType, ServerInstanceConfig, }, - create_local, - local::loadbalancing::PingBalancer, + local::{loadbalancing::PingBalancer, Server}, shadowsocks::{ config::{Mode, ServerAddr, ServerConfig}, crypto::{available_ciphers, CipherKind}, @@ -869,14 +868,14 @@ pub fn main(matches: &ArgMatches) -> ExitCode { runtime.block_on(async move { let config_path = config.config_path.clone(); - let instance = create_local(config).await.expect("create local"); + let instance = Server::new(config).await.expect("create local"); if let Some(config_path) = config_path { launch_reload_server_task(config_path, instance.server_balancer().clone()); } let abort_signal = monitor::create_signal_monitor(); - let server = instance.wait_until_exit(); + let server = instance.run(); tokio::pin!(abort_signal); tokio::pin!(server);