From 7a698e36e7a084bf9490f51fff0f64a92e9b4b3d Mon Sep 17 00:00:00 2001 From: zonyitoo Date: Sun, 10 May 2020 21:43:44 +0800 Subject: [PATCH] Removed unnecessary Runtime Handle in servers entry function BREAKING: run_local, run_server, run_manager have removed Handle parameter --- src/bin/local.rs | 3 +- src/bin/manager.rs | 3 +- src/bin/server.rs | 3 +- src/context.rs | 24 ++-- src/lib.rs | 14 +-- src/relay/dns_resolver/trust_dns_resolver.rs | 119 ++++++++----------- src/relay/local.rs | 5 +- src/relay/manager.rs | 6 +- src/relay/server.rs | 6 +- tests/socks5.rs | 81 ++++++------- tests/tunnel.rs | 79 ++++++------ tests/udp.rs | 74 ++++++------ 12 files changed, 183 insertions(+), 234 deletions(-) diff --git a/src/bin/local.rs b/src/bin/local.rs index 5c77bdcc..589ff9a3 100644 --- a/src/bin/local.rs +++ b/src/bin/local.rs @@ -294,10 +294,9 @@ fn main() { builder.threaded_scheduler(); } let mut runtime = builder.enable_all().build().expect("create tokio Runtime"); - let rt_handle = runtime.handle().clone(); runtime.block_on(async move { let abort_signal = monitor::create_signal_monitor(); - let server = run_local(config, rt_handle); + let server = run_local(config); tokio::pin!(abort_signal); tokio::pin!(server); diff --git a/src/bin/manager.rs b/src/bin/manager.rs index 07931d48..84a486f9 100644 --- a/src/bin/manager.rs +++ b/src/bin/manager.rs @@ -150,10 +150,9 @@ fn main() { builder.threaded_scheduler(); } let mut runtime = builder.enable_all().build().expect("create tokio Runtime"); - let rt_handle = runtime.handle().clone(); runtime.block_on(async move { let abort_signal = monitor::create_signal_monitor(); - let server = run_manager(config, rt_handle); + let server = run_manager(config); tokio::pin!(abort_signal); tokio::pin!(server); diff --git a/src/bin/server.rs b/src/bin/server.rs index 7e196f66..15f88357 100644 --- a/src/bin/server.rs +++ b/src/bin/server.rs @@ -174,10 +174,9 @@ fn main() { builder.threaded_scheduler(); } let mut runtime = builder.enable_all().build().expect("create tokio Runtime"); - let rt_handle = runtime.handle().clone(); runtime.block_on(async move { let abort_signal = monitor::create_signal_monitor(); - let server = run_server(config, rt_handle); + let server = run_server(config); tokio::pin!(abort_signal); tokio::pin!(server); diff --git a/src/context.rs b/src/context.rs index 4ca5ef38..71f5992b 100644 --- a/src/context.rs +++ b/src/context.rs @@ -17,18 +17,17 @@ use log::warn; #[cfg(feature = "local-dns-relay")] use lru_time_cache::LruCache; use spin::Mutex; -use tokio::runtime::Handle; #[cfg(feature = "trust-dns")] use trust_dns_resolver::TokioAsyncResolver; #[cfg(feature = "trust-dns")] use crate::relay::dns_resolver::create_resolver; -#[cfg(feature = "local-flow-stat")] -use crate::relay::flow::ServerFlowStatistic; -#[cfg(feature = "local-dns-relay")] -use crate::relay::dnsrelay::upstream::LocalUpstream; #[cfg(not(feature = "local-dns-relay"))] use crate::relay::dns_resolver::resolve; +#[cfg(feature = "local-dns-relay")] +use crate::relay::dnsrelay::upstream::LocalUpstream; +#[cfg(feature = "local-flow-stat")] +use crate::relay::flow::ServerFlowStatistic; use crate::{ acl::AccessControl, config::{Config, ConfigType, ServerConfig}, @@ -130,9 +129,9 @@ pub struct ServerState { #[cfg(feature = "trust-dns")] impl ServerState { /// Create a global shared server state - pub async fn new_shared(config: &Config, rt: Handle) -> SharedServerState { + pub async fn new_shared(config: &Config) -> SharedServerState { let state = ServerState { - dns_resolver: match create_resolver(config.get_dns_config(), config.timeout, config.ipv6_first, rt).await { + dns_resolver: match create_resolver(config.get_dns_config(), config.timeout, config.ipv6_first).await { Ok(resolver) => Some(resolver), Err(..) => None, }, @@ -335,11 +334,12 @@ impl Context { /// Add a record to the reverse lookup cache #[cfg(feature = "local-dns-relay")] pub fn add_to_reverse_lookup_cache(&self, addr: &IpAddr, forward: bool) { - let is_exception = forward != match self.acl() { - // Proxy everything by default - None => true, - Some(ref a) => a.check_ip_in_proxy_list(addr) - }; + let is_exception = forward + != match self.acl() { + // Proxy everything by default + None => true, + Some(ref a) => a.check_ip_in_proxy_list(addr), + }; let mut reverse_lookup_cache = self.reverse_lookup_cache.lock(); match reverse_lookup_cache.get_mut(addr) { Some(value) => { diff --git a/src/lib.rs b/src/lib.rs index 391ed936..64324411 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,10 +55,9 @@ //! use shadowsocks::{run_local, Config, ConfigType}; //! //! let mut rt = Runtime::new().expect("Failed to create runtime"); -//! let rt_handle = rt.handle().clone(); //! //! let config = Config::load_from_file("shadowsocks.json", ConfigType::Socks5Local).unwrap(); -//! rt.block_on(run_local(config, rt_handle)); +//! rt.block_on(run_local(config)); //! ``` //! //! That's all! And let me show you how to run a proxy server @@ -68,10 +67,9 @@ //! use shadowsocks::{run_server, Config, ConfigType}; //! //! let mut rt = Runtime::new().expect("Failed to create runtime"); -//! let rt_handle = rt.handle().clone(); //! //! let config = Config::load_from_file("shadowsocks.json", ConfigType::Server).unwrap(); -//! rt.block_on(run_server(config, rt_handle)); +//! rt.block_on(run_server(config)); //! ``` #![crate_type = "lib"] @@ -80,8 +78,6 @@ use std::io; -use tokio::runtime::Handle; - /// ShadowSocks version pub const VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -105,10 +101,10 @@ pub mod relay; /// Start a ShadowSocks' server /// /// For `config.config_type` in `Socks5Local`, `HttpLocal` and `TunnelLocal`, server will run in Local mode. -pub async fn run(config: Config, rt: Handle) -> io::Result<()> { +pub async fn run(config: Config) -> io::Result<()> { if config.config_type.is_local() { - run_local(config, rt).await + run_local(config).await } else { - run_server(config, rt).await + run_server(config).await } } diff --git a/src/relay/dns_resolver/trust_dns_resolver.rs b/src/relay/dns_resolver/trust_dns_resolver.rs index 715d7798..96301d3b 100644 --- a/src/relay/dns_resolver/trust_dns_resolver.rs +++ b/src/relay/dns_resolver/trust_dns_resolver.rs @@ -7,7 +7,6 @@ use std::{ }; use log::{error, trace}; -use tokio::{self, runtime::Handle}; use trust_dns_resolver::{ config::{LookupIpStrategy, ResolverConfig, ResolverOpts}, TokioAsyncResolver, @@ -21,7 +20,6 @@ pub async fn create_resolver( dns: Option, timeout: Option, ipv6_first: bool, - rt: Handle, ) -> io::Result { let mut resolver_opts = ResolverOpts::default(); if let Some(d) = timeout { @@ -32,76 +30,63 @@ pub async fn create_resolver( resolver_opts.ip_strategy = LookupIpStrategy::Ipv6thenIpv4; } - { - // To make this independent, if targeting macOS, BSD, Linux, or Windows, we can use the system's configuration: - #[cfg(any(unix, windows))] - { - if let Some(conf) = dns { - trace!( - "initializing DNS resolver with config {:?} opts {:?}", - conf, - resolver_opts - ); - TokioAsyncResolver::new(conf, resolver_opts, rt) - } else { - use trust_dns_resolver::system_conf::read_system_conf; - // use the system resolver configuration - let (config, mut opts) = match read_system_conf() { - Ok(o) => o, - Err(err) => { - error!("failed to initialize DNS resolver with system-config, error: {}", err); - - // From::from is required because on error type is different on Windows - #[allow(clippy::identity_conversion)] - return Err(From::from(err)); - } - }; - - // NOTE: timeout will be set by config (for example, /etc/resolv.conf on UNIX-like system) - // - // Only ip_strategy should be changed - if ipv6_first { - opts.ip_strategy = LookupIpStrategy::Ipv6thenIpv4; - } - - trace!( - "initializing DNS resolver with system-config {:?} opts {:?}", - config, - opts - ); - - TokioAsyncResolver::new(config, opts, rt) - } + // Customized dns resolution + match dns { + Some(conf) => { + trace!( + "initializing DNS resolver with config {:?} opts {:?}", + conf, + resolver_opts + ); + TokioAsyncResolver::tokio(conf, resolver_opts).await } - // For other operating systems, we can use one of the preconfigured definitions - #[cfg(not(any(unix, windows)))] - { - if let Some(conf) = dns { - trace!( - "initializing DNS resolver with config {:?} opts {:?}", - conf, - resolver_opts - ); - TokioAsyncResolver::new(conf, resolver_opts, rt) - } else { - // Get a new resolver with the google nameservers as the upstream recursive resolvers - trace!( - "initializing DNS resolver with google-config {:?} opts {:?}", - ResolverConfig::google(), - resolver_opts - ); - TokioAsyncResolver::new(ResolverConfig::google(), resolver_opts, rt) + // To make this independent, if targeting macOS, BSD, Linux, or Windows, we can use the system's configuration + #[cfg(any(unix, windows))] + None => { + use tokio::runtime::Handle; + use trust_dns_resolver::system_conf::read_system_conf; + + // use the system resolver configuration + let (config, mut opts) = match read_system_conf() { + Ok(o) => o, + Err(err) => { + error!("failed to initialize DNS resolver with system-config, error: {}", err); + + // From::from is required because on error type is different on Windows + #[allow(clippy::identity_conversion)] + return Err(From::from(err)); + } + }; + + // NOTE: timeout will be set by config (for example, /etc/resolv.conf on UNIX-like system) + // + // Only ip_strategy should be changed + if ipv6_first { + opts.ip_strategy = LookupIpStrategy::Ipv6thenIpv4; } + + trace!( + "initializing DNS resolver with system-config {:?} opts {:?}", + config, + opts + ); + + TokioAsyncResolver::new(config, opts, Handle::current()).await + } + + #[cfg(not(any(unix, windows)))] + None => { + // Get a new resolver with the google nameservers as the upstream recursive resolvers + trace!( + "initializing DNS resolver with google-config {:?} opts {:?}", + ResolverConfig::google(), + resolver_opts + ); + TokioAsyncResolver::tokio(ResolverConfig::google(), resolver_opts).await } } - .await - .map_err(|err| { - Error::new( - ErrorKind::Other, - format!("failed to create trust-dns DNS resolver, error: {}", err), - ) - }) + .map_err(From::from) } /// Perform a DNS resolution diff --git a/src/relay/local.rs b/src/relay/local.rs index 8b47fe8e..c52da43a 100644 --- a/src/relay/local.rs +++ b/src/relay/local.rs @@ -4,7 +4,6 @@ use std::io::{self, ErrorKind}; use futures::{future::select_all, FutureExt}; use log::{debug, error, trace, warn}; -use tokio::runtime::Handle; #[cfg(feature = "local-flow-stat")] use crate::context::SharedContext; @@ -16,7 +15,7 @@ use crate::{ }; /// Relay server running under local environment. -pub async fn run(mut config: Config, rt: Handle) -> io::Result<()> { +pub async fn run(mut config: Config) -> io::Result<()> { trace!("initializing local server with {:?}", config); assert!(config.config_type.is_local()); @@ -43,7 +42,7 @@ pub async fn run(mut config: Config, rt: Handle) -> io::Result<()> { let mode = config.mode; // Create a context containing a DNS resolver and server running state flag. - let state = ServerState::new_shared(&config, rt).await; + let state = ServerState::new_shared(&config).await; let mut vf = Vec::new(); diff --git a/src/relay/manager.rs b/src/relay/manager.rs index 97fe6a5b..d64dad17 100644 --- a/src/relay/manager.rs +++ b/src/relay/manager.rs @@ -16,7 +16,7 @@ use futures::future; use log::{debug, error, trace, warn}; #[cfg(unix)] use tokio::net::UnixDatagram; -use tokio::{self, net::UdpSocket, runtime::Handle, sync::oneshot}; +use tokio::{self, net::UdpSocket, sync::oneshot}; use crate::{ config::{Config, ConfigType, ManagerAddr, Mode, ServerAddr, ServerConfig}, @@ -533,7 +533,7 @@ impl ManagerService { } /// Server manager for supporting [Manage Multiple Users](https://github.com/shadowsocks/shadowsocks/wiki/Manage-Multiple-Users) APIs -pub async fn run(config: Config, rt: Handle) -> io::Result<()> { +pub async fn run(config: Config) -> io::Result<()> { assert!(config.config_type.is_manager()); if let Some(nofile) = config.nofile { @@ -555,7 +555,7 @@ pub async fn run(config: Config, rt: Handle) -> io::Result<()> { } // Create a context containing a DNS resolver and server running state flag. - let state = ServerState::new_shared(&config, rt.clone()).await; + let state = ServerState::new_shared(&config).await; let context = Context::new_shared(config, state.clone()); let bind_addr = match context.config().manager_addr { diff --git a/src/relay/server.rs b/src/relay/server.rs index 8537b7c4..c896fe80 100644 --- a/src/relay/server.rs +++ b/src/relay/server.rs @@ -7,7 +7,7 @@ use std::{ use futures::future::{select_all, FutureExt}; use log::{debug, error, trace, warn}; -use tokio::{runtime::Handle, time}; +use tokio::time; use crate::{ config::Config, @@ -24,9 +24,9 @@ use crate::{ /// Runs Relay server on server side. #[inline] -pub async fn run(config: Config, rt: Handle) -> io::Result<()> { +pub async fn run(config: Config) -> io::Result<()> { // Create a context containing a DNS resolver and server running state flag. - let server_state = ServerState::new_shared(&config, rt).await; + let server_state = ServerState::new_shared(&config).await; // Create statistics for multiple servers // diff --git a/tests/socks5.rs b/tests/socks5.rs index 7b4a749c..eaea2a4d 100644 --- a/tests/socks5.rs +++ b/tests/socks5.rs @@ -2,7 +2,6 @@ use std::net::{SocketAddr, ToSocketAddrs}; use tokio::{ prelude::*, - runtime::{Builder, Handle}, time::{self, Duration}, }; @@ -51,19 +50,19 @@ impl Socks5TestServer { &self.local_addr } - pub async fn run(&self, rt_handle: Handle) { + pub async fn run(&self) { let svr_cfg = self.svr_config.clone(); - tokio::spawn(run_server(svr_cfg, rt_handle.clone())); + tokio::spawn(run_server(svr_cfg)); let client_cfg = self.cli_config.clone(); - tokio::spawn(run_local(client_cfg, rt_handle)); + tokio::spawn(run_local(client_cfg)); time::delay_for(Duration::from_secs(1)).await; } } -#[test] -fn socks5_relay_stream() { +#[tokio::test] +async fn socks5_relay_stream() { let _ = env_logger::try_init(); const SERVER_ADDR: &str = "127.0.0.1:8100"; @@ -72,33 +71,28 @@ fn socks5_relay_stream() { const PASSWORD: &str = "test-password"; const METHOD: CipherType = CipherType::Aes128Gcm; - let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap(); - let rt_handle = rt.handle().clone(); + let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false); + svr.run().await; - rt.block_on(async move { - let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false); - svr.run(rt_handle).await; + let mut c = Socks5Client::connect( + Address::DomainNameAddress("www.example.com".to_owned(), 80), + svr.client_addr(), + ) + .await + .unwrap(); - let mut c = Socks5Client::connect( - Address::DomainNameAddress("www.example.com".to_owned(), 80), - svr.client_addr(), - ) - .await - .unwrap(); + let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n"; + c.write_all(req).await.unwrap(); + c.flush().await.unwrap(); - let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n"; - c.write_all(req).await.unwrap(); - c.flush().await.unwrap(); + let mut buf = Vec::new(); + c.read_to_end(&mut buf).await.unwrap(); - let mut buf = Vec::new(); - c.read_to_end(&mut buf).await.unwrap(); - - println!("Got reply from server: {}", String::from_utf8(buf).unwrap()); - }); + println!("Got reply from server: {}", String::from_utf8(buf).unwrap()); } -#[test] -fn socks5_relay_aead() { +#[tokio::test] +async fn socks5_relay_aead() { let _ = env_logger::try_init(); const SERVER_ADDR: &str = "127.0.0.1:8110"; @@ -107,27 +101,22 @@ fn socks5_relay_aead() { const PASSWORD: &str = "test-password"; const METHOD: CipherType = CipherType::Aes256Gcm; - let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap(); - let rt_handle = rt.handle().clone(); + let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false); + svr.run().await; - rt.block_on(async move { - let svr = Socks5TestServer::new(SERVER_ADDR, LOCAL_ADDR, PASSWORD, METHOD, false); - svr.run(rt_handle).await; + let mut c = Socks5Client::connect( + Address::DomainNameAddress("www.example.com".to_owned(), 80), + svr.client_addr(), + ) + .await + .unwrap(); - let mut c = Socks5Client::connect( - Address::DomainNameAddress("www.example.com".to_owned(), 80), - svr.client_addr(), - ) - .await - .unwrap(); + let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n"; + c.write_all(req).await.unwrap(); + c.flush().await.unwrap(); - let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n"; - c.write_all(req).await.unwrap(); - c.flush().await.unwrap(); + let mut buf = Vec::new(); + c.read_to_end(&mut buf).await.unwrap(); - let mut buf = Vec::new(); - c.read_to_end(&mut buf).await.unwrap(); - - println!("Got reply from server: {}", String::from_utf8(buf).unwrap()); - }); + println!("Got reply from server: {}", String::from_utf8(buf).unwrap()); } diff --git a/tests/tunnel.rs b/tests/tunnel.rs index 01e94d52..604d845d 100644 --- a/tests/tunnel.rs +++ b/tests/tunnel.rs @@ -2,7 +2,6 @@ use tokio::{ self, net::{TcpStream, UdpSocket}, prelude::*, - runtime::Builder, time::{self, Duration}, }; @@ -13,8 +12,8 @@ use shadowsocks::{ run_server, }; -#[test] -fn tcp_tunnel() { +#[tokio::test] +async fn tcp_tunnel() { let _ = env_logger::try_init(); let mut local_config = Config::load_from_str( @@ -43,31 +42,26 @@ fn tcp_tunnel() { ) .unwrap(); - let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap(); - let rt_handle = rt.handle().clone(); + tokio::spawn(run_local(local_config)); + tokio::spawn(run_server(server_config)); - rt.block_on(async move { - tokio::spawn(run_local(local_config, rt_handle.clone())); - tokio::spawn(run_server(server_config, rt_handle)); + time::delay_for(Duration::from_secs(1)).await; - time::delay_for(Duration::from_secs(1)).await; + // Connect it directly, because it is now established a TCP tunnel with www.example.com + let mut stream = TcpStream::connect("127.0.0.1:9110").await.unwrap(); - // Connect it directly, because it is now established a TCP tunnel with www.example.com - let mut stream = TcpStream::connect("127.0.0.1:9110").await.unwrap(); + let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n"; + stream.write_all(req).await.unwrap(); + stream.flush().await.unwrap(); - let req = b"GET / HTTP/1.0\r\nHost: www.example.com\r\nAccept: */*\r\n\r\n"; - stream.write_all(req).await.unwrap(); - stream.flush().await.unwrap(); + let mut buf = Vec::new(); + stream.read_to_end(&mut buf).await.unwrap(); - let mut buf = Vec::new(); - stream.read_to_end(&mut buf).await.unwrap(); - - println!("Got reply from server: {}", String::from_utf8(buf).unwrap()); - }); + println!("Got reply from server: {}", String::from_utf8(buf).unwrap()); } -#[test] -fn udp_tunnel() { +#[tokio::test] +async fn udp_tunnel() { let _ = env_logger::try_init(); let mut local_config = Config::load_from_str( @@ -98,33 +92,28 @@ fn udp_tunnel() { ) .unwrap(); - let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap(); - let rt_handle = rt.handle().clone(); + tokio::spawn(run_local(local_config)); + tokio::spawn(run_server(server_config)); - rt.block_on(async move { - tokio::spawn(run_local(local_config, rt_handle.clone())); - tokio::spawn(run_server(server_config, rt_handle)); - - // Start a UDP echo server - tokio::spawn(async { - let mut socket = UdpSocket::bind("127.0.0.1:9230").await.unwrap(); - - let mut buf = vec![0u8; 65536]; - let (n, src) = socket.recv_from(&mut buf).await.unwrap(); - - println!("UDP Echo server received packet, size: {}, src: {}", n, src); - - socket.send_to(&buf[..n], src).await.unwrap(); - }); - - time::delay_for(Duration::from_secs(1)).await; - - let mut socket = UdpSocket::bind("0.0.0.0:0").await.unwrap(); - socket.send_to(b"HELLO WORLD", "127.0.0.1:9210").await.unwrap(); + // Start a UDP echo server + tokio::spawn(async { + let mut socket = UdpSocket::bind("127.0.0.1:9230").await.unwrap(); let mut buf = vec![0u8; 65536]; - let n = socket.recv(&mut buf).await.unwrap(); + let (n, src) = socket.recv_from(&mut buf).await.unwrap(); - println!("Got reply from server: {}", ::std::str::from_utf8(&buf[..n]).unwrap()); + println!("UDP Echo server received packet, size: {}, src: {}", n, src); + + socket.send_to(&buf[..n], src).await.unwrap(); }); + + time::delay_for(Duration::from_secs(1)).await; + + let mut socket = UdpSocket::bind("0.0.0.0:0").await.unwrap(); + socket.send_to(b"HELLO WORLD", "127.0.0.1:9210").await.unwrap(); + + let mut buf = vec![0u8; 65536]; + let n = socket.recv(&mut buf).await.unwrap(); + + println!("Got reply from server: {}", ::std::str::from_utf8(&buf[..n]).unwrap()); } diff --git a/tests/udp.rs b/tests/udp.rs index 061c31bc..aaa0d12f 100644 --- a/tests/udp.rs +++ b/tests/udp.rs @@ -9,7 +9,6 @@ use bytes::{BufMut, BytesMut}; use log::debug; use tokio::{ prelude::*, - runtime::{Builder, Handle}, time::{self, Duration}, }; @@ -60,12 +59,12 @@ fn get_client_addr() -> SocketAddr { LOCAL_ADDR.parse().unwrap() } -fn start_server(handle: Handle) { - tokio::spawn(run_server(get_svr_config(), handle)); +fn start_server() { + tokio::spawn(run_server(get_svr_config())); } -fn start_local(handle: Handle) { - tokio::spawn(run_local(get_cli_config(), handle)); +fn start_local() { + tokio::spawn(run_local(get_cli_config())); } fn start_udp_echo_server() { @@ -102,56 +101,51 @@ fn start_udp_request_holder(addr: Address) { }); } -#[test] -fn udp_relay() { +#[tokio::test] +async fn udp_relay() { use tokio::net::UdpSocket; let _ = env_logger::try_init(); let remote_addr = Address::SocketAddress(UDP_ECHO_SERVER_ADDR.parse().unwrap()); - let mut rt = Builder::new().basic_scheduler().enable_all().build().unwrap(); - let rt_handle = rt.handle().clone(); + start_server(); + start_local(); - rt.block_on(async move { - start_server(rt_handle.clone()); - start_local(rt_handle); + start_udp_echo_server(); - start_udp_echo_server(); + // Wait until all server starts + time::delay_for(Duration::from_secs(1)).await; - // Wait until all server starts - time::delay_for(Duration::from_secs(1)).await; + start_udp_request_holder(remote_addr.clone()); - start_udp_request_holder(remote_addr.clone()); + let mut l = UdpSocket::bind(UDP_LOCAL_ADDR).await.unwrap(); - let mut l = UdpSocket::bind(UDP_LOCAL_ADDR).await.unwrap(); + let header = UdpAssociateHeader::new(0, remote_addr); + let mut buf = BytesMut::with_capacity(header.serialized_len()); + header.write_to_buf(&mut buf); - let header = UdpAssociateHeader::new(0, remote_addr); - let mut buf = BytesMut::with_capacity(header.serialized_len()); - header.write_to_buf(&mut buf); + let payload = b"HEllo WORld"; - let payload = b"HEllo WORld"; + buf.reserve(payload.len()); + buf.put_slice(payload); - buf.reserve(payload.len()); - buf.put_slice(payload); + let local_addr = LOCAL_ADDR.parse::().unwrap(); + l.send_to(&buf[..], &local_addr).await.unwrap(); - let local_addr = LOCAL_ADDR.parse::().unwrap(); - l.send_to(&buf[..], &local_addr).await.unwrap(); + let mut buf = vec![0u8; 65536]; + let (amt, _) = time::timeout(Duration::from_secs(5), l.recv_from(&mut buf)) + .await + .unwrap() + .unwrap(); + println!("Received buf size={} {:?}", amt, &buf[..amt]); - let mut buf = vec![0u8; 65536]; - let (amt, _) = time::timeout(Duration::from_secs(5), l.recv_from(&mut buf)) - .await - .unwrap() - .unwrap(); - println!("Received buf size={} {:?}", amt, &buf[..amt]); + let mut cur = Cursor::new(buf[..amt].to_vec()); + let header = UdpAssociateHeader::read_from(&mut cur).await.unwrap(); + println!("{:?}", header); + let header_len = cur.position() as usize; + let buf = cur.into_inner(); + let buf = &buf[header_len..]; - let mut cur = Cursor::new(buf[..amt].to_vec()); - let header = UdpAssociateHeader::read_from(&mut cur).await.unwrap(); - println!("{:?}", header); - let header_len = cur.position() as usize; - let buf = cur.into_inner(); - let buf = &buf[header_len..]; - - assert_eq!(buf, payload); - }); + assert_eq!(buf, payload); }