extern crate core;
use core::time;
use std::thread;
use time::Duration;
use bytes::Buf;
use futures::future::{FutureExt, LocalBoxFuture};
use env_logger;
use pi_async_rt::rt::{serial::AsyncRuntimeBuilder};
use pi_tcp::{AsyncService, Socket, SocketHandle, SocketConfig, SocketStatus,
connect::TcpSocket,
tls_connect::TlsSocket,
server::{PortsAdapterFactory, SocketListener},
utils::TlsConfig};
#[test]
fn test_accept_connect() {
use std::net::SocketAddr;
use mio::{Events, Poll, Interest, Token};
use mio::net::TcpListener;
let addr: SocketAddr = "127.0.0.1:38880".parse().unwrap();
let mut server = TcpListener::bind(addr).unwrap();
let mut poll = Poll::new().unwrap();
poll.registry().register(&mut server, Token(0), Interest::READABLE).unwrap();
let mut events = Events::with_capacity(1024);
loop {
poll.poll(&mut events, None).unwrap();
for event in &events {
println!("!!!!!!event: {:?}", event);
let (_, addr) = server.accept().unwrap();
println!("!!!!!!connected, addr: {:?}", addr);
poll.registry().reregister(&mut server, Token(0), Interest::READABLE).unwrap();
}
}
}
struct TestService;
impl<S: Socket> AsyncService<S> for TestService {
fn handle_connected(&self,
handle: SocketHandle<S>,
status: SocketStatus) -> LocalBoxFuture<'static, ()> {
async move {
let token = handle.get_token().clone();
match status {
SocketStatus::Connected(Err(e)) => {
println!("!!!> Connect Error, token: {:?}, remote: {:?}, local: {:?}, reason: {:?}", token, handle.get_remote(), handle.get_local(), e);
},
SocketStatus::Connected(Ok(_)) => {
println!("===> Connect Ok, token: {:?}, remote: {:?}, local: {:?}", token, handle.get_remote(), handle.get_local());
},
_ => unimplemented!(),
}
}.boxed_local()
}
fn handle_readed(&self,
handle: SocketHandle<S>,
status: SocketStatus) -> LocalBoxFuture<'static, ()> {
async move {
println!("!!!!!!callback readed");
let token = handle.get_token().clone();
match status {
SocketStatus::Readed(Err(e)) => {
println!("!!!> Socket Receive Error, token: {:?}, remote: {:?}, local: {:?}, reason: {:?}", token, handle.get_remote(), handle.get_local(), e);
},
SocketStatus::Readed(Ok(_)) => {
println!("===> Socket Receive Ok, token: {:?}, remote: {:?}, local: {:?}", token, handle.get_remote(), handle.get_local());
let mut ready_len = 0;
if let Some(buf) = unsafe { (&mut *handle.get_read_buffer().get()) } {
if buf.remaining() == 0 {
println!("!!!!!!readed, read ready start, len: 0");
ready_len = match handle.read_ready(0) {
Err(len) => len,
Ok(value) => {
println!("!!!!!!wait read_ready");
let r = value.await;
println!("!!!!!!wakeup read_ready, len: {}", r);
r
},
};
if ready_len == 0 {
return;
}
}
println!("===> Socket Read Ok, token: {:?}, data: {:?}", token, String::from_utf8_lossy(buf.as_ref()));
let mut bin = b"HTTP/1.0 200 OK\r\nContent-Length: 35\r\nConnection: close\r\n\r\nHello world from rust web server!\r\n";
if let Ok(_) = handle.write_ready(bin) {
println!("===> Socket Write Ok, token: {:?}", token);
}
} else {
return;
}
},
_ => unimplemented!(),
}
}.boxed_local()
}
fn handle_writed(&self,
handle: SocketHandle<S>,
status: SocketStatus) -> LocalBoxFuture<'static, ()> {
async move {
let token = handle.get_token().clone();
match status {
SocketStatus::Writed(Err(e)) => {
println!("!!!> Socket Send Error, token: {:?}, remote: {:?}, local: {:?}, reason: {:?}", token, handle.get_remote(), handle.get_local(), e);
},
SocketStatus::Writed(Ok(_)) => {
println!("===> Socket Send Ok, token: {:?}, remote: {:?}, local: {:?}", token, handle.get_remote(), handle.get_local());
},
_ => unimplemented!(),
}
}.boxed_local()
}
fn handle_closed(&self,
handle: SocketHandle<S>,
status: SocketStatus) -> LocalBoxFuture<'static, ()> {
async move {
let token = handle.get_token().clone();
match status {
SocketStatus::Closed(Err(e)) => {
println!("!!!> Socket Close Error, token: {:?}, remote: {:?}, local: {:?}, reason: {:?}", token, handle.get_remote(), handle.get_local(), e);
},
SocketStatus::Closed(Ok(_)) => {
println!("===> Socket Close Ok, token: {:?}, remote: {:?}, local: {:?}", token, handle.get_remote(), handle.get_local());
},
_ => unimplemented!(),
}
}.boxed_local()
}
fn handle_timeouted(&self,
handle: SocketHandle<S>,
_status: SocketStatus) -> LocalBoxFuture<'static, ()> {
async move {
let token = handle.get_token().clone();
println!("!!!> Socket Timeout, token: {:?}, remote: {:?}, local: {:?}", token, handle.get_remote(), handle.get_local());
}.boxed_local()
}
}
#[test]
fn test_tcp_connect() {
env_logger::builder().format_timestamp_millis().init();
let rt = AsyncRuntimeBuilder::default_local_thread(None, None);
let mut factory = PortsAdapterFactory::<TcpSocket>::new();
factory.bind(38080, Box::new(TestService));
let mut config = SocketConfig::new("0.0.0.0", factory.ports().as_slice());
config.set_option(16384, 16384, 16384, 16);
match SocketListener::bind(vec![rt],
factory,
config,
1024,
1024 * 1024,
1024,
16,
4096,
4096,
Some(100)) {
Err(e) => {
println!("!!!> Socket Listener Bind Ipv4 & Ipv6 Address Error, reason: {:?}", e);
},
Ok(driver) => {
println!("===> Socket Listener Bind Ipv4 & Ipv6 Address Ok");
}
}
thread::sleep(Duration::from_millis(10000000));
}
#[test]
fn test_tls_connect() {
env_logger::builder().format_timestamp_millis().init();
let rt = AsyncRuntimeBuilder::default_local_thread(None, None);
let mut factory = PortsAdapterFactory::<TlsSocket>::new();
factory.bind(38080, Box::new(TestService));
let tls_config = TlsConfig::new_server("",
false,
"./tests/17youx.cn.pem",
"./tests/17youx.cn.key",
"",
"",
"",
512,
false,
"").unwrap();
let mut config = SocketConfig::with_tls("0.0.0.0", &[(38080, tls_config)]);
config.set_option(16384, 16384, 16384, 16);
match SocketListener::bind(vec![rt],
factory,
config,
1024,
1024 * 1024,
1024,
16,
4096,
4096,
Some(100)) {
Err(e) => {
println!("!!!> Socket Listener Bind Ipv4 & Ipv6 Address Error, reason: {:?}", e);
},
Ok(driver) => {
println!("===> Socket Listener Bind Ipv4 & Ipv6 Address Ok");
}
}
thread::sleep(Duration::from_millis(10000000));
}
#[test]
fn test_close_tcp_listener() {
env_logger::builder().format_timestamp_millis().init();
let rt = AsyncRuntimeBuilder::default_local_thread(None, None);
let mut factory = PortsAdapterFactory::<TcpSocket>::new();
factory.bind(38080, Box::new(TestService));
let mut config = SocketConfig::new("0.0.0.0", factory.ports().as_slice());
config.set_option(16384, 16384, 16384, 16);
match SocketListener::bind(vec![rt],
factory,
config,
1024,
1024 * 1024,
1024,
16,
4096,
4096,
Some(100)) {
Err(e) => {
println!("!!!> Socket Listener Bind Ipv4 & Ipv6 Address Error, reason: {:?}", e);
},
Ok(driver) => {
println!("===> Socket Listener Bind Ipv4 & Ipv6 Address Ok");
thread::sleep(Duration::from_millis(10000));
driver.close(Ok(()));
}
}
thread::sleep(Duration::from_millis(10000000));
}