berkus
/
actix-web
mirror of https://github.com/fafhrd91/actix-web
1
0
Fork 0
Code Issues Releases Wiki Activity

run test servers in io-uring rt friendly way

This commit is contained in:
Rob Ede 2021-11-21 23:11:14 +00:00
parent 452331b745
commit 1f28bea3a5
No known key found for this signature in database
GPG Key ID: 97C636207D3EF933
2 changed files with 169 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
actix-http-test/src/lib.rs
View File

@ -66,25 +66,24 @@ pub async fn test_server_with_addr<F: ServiceFactory<TcpStream>>(
// run server in separate thread // run server in separate thread
thread::spawn(move || { thread::spawn(move || {
let sys = System::new(); System::new().block_on(async move {
let local_addr = tcp.local_addr().unwrap(); let local_addr = tcp.local_addr().unwrap();
let srv = Server::build() let srv = Server::build()
.workers(1) .workers(1)
.disable_signals() .disable_signals()
.listen("test", tcp, factory) .system_exit()
.expect("test server could not be created"); .listen("test", tcp, factory)
.expect("test server could not be created");
let srv = srv.run(); let srv = srv.run();
started_tx started_tx
.send((System::current(), srv.handle(), local_addr)) .send((System::current(), srv.handle(), local_addr))
.unwrap(); .unwrap();
// drive server loop // drive server loop
sys.block_on(srv).unwrap(); srv.await.unwrap();
});
// start system event loop
sys.run().unwrap();
// notify TestServer that server and system have shut down // notify TestServer that server and system have shut down
// all thread managed resources should be dropped at this point // all thread managed resources should be dropped at this point

281
actix-test/src/lib.rs
View File

@ -146,156 +146,183 @@ where
// run server in separate orphaned thread // run server in separate orphaned thread
thread::spawn(move || { thread::spawn(move || {
let sys = rt::System::new(); rt::System::new().block_on(async move {
let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap(); let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap();
let local_addr = tcp.local_addr().unwrap(); let local_addr = tcp.local_addr().unwrap();
let factory = factory.clone(); let factory = factory.clone();
let srv_cfg = cfg.clone(); let srv_cfg = cfg.clone();
let timeout = cfg.client_timeout; let timeout = cfg.client_timeout;
let builder = Server::build().workers(1).disable_signals().system_exit();
let srv = match srv_cfg.stream { let builder = Server::build().workers(1).disable_signals().system_exit();
StreamType::Tcp => match srv_cfg.tp {
HttpVer::Http1 => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() let srv = match srv_cfg.stream {
.into_factory() StreamType::Tcp => match srv_cfg.tp {
.map_err(|err| err.into().error_response()); HttpVer::Http1 => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.h1(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.tcp()
}),
HttpVer::Http2 => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .h1(map_config(fac, move |_| app_cfg.clone()))
.tcp()
}),
HttpVer::Http2 => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.h2(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.tcp()
}),
HttpVer::Both => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .h2(map_config(fac, move |_| app_cfg.clone()))
.tcp()
}),
HttpVer::Both => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.finish(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.tcp()
}),
},
#[cfg(feature = "openssl")]
StreamType::Openssl(acceptor) => match cfg.tp {
HttpVer::Http1 => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .finish(map_config(fac, move |_| app_cfg.clone()))
.tcp()
}),
},
#[cfg(feature = "openssl")]
StreamType::Openssl(acceptor) => match cfg.tp {
HttpVer::Http1 => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.h1(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.openssl(acceptor.clone())
}),
HttpVer::Http2 => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .h1(map_config(fac, move |_| app_cfg.clone()))
.openssl(acceptor.clone())
}),
HttpVer::Http2 => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.h2(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.openssl(acceptor.clone())
}),
HttpVer::Both => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .h2(map_config(fac, move |_| app_cfg.clone()))
.openssl(acceptor.clone())
}),
HttpVer::Both => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.finish(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.openssl(acceptor.clone())
}),
},
#[cfg(feature = "rustls")]
StreamType::Rustls(config) => match cfg.tp {
HttpVer::Http1 => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .finish(map_config(fac, move |_| app_cfg.clone()))
.openssl(acceptor.clone())
}),
},
#[cfg(feature = "rustls")]
StreamType::Rustls(config) => match cfg.tp {
HttpVer::Http1 => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.h1(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.rustls(config.clone())
}),
HttpVer::Http2 => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .h1(map_config(fac, move |_| app_cfg.clone()))
.rustls(config.clone())
}),
HttpVer::Http2 => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.h2(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.rustls(config.clone())
}),
HttpVer::Both => builder.listen("test", tcp, move || {
let app_cfg =
AppConfig::__priv_test_new(false, local_addr.to_string(), local_addr);
let fac = factory() HttpService::build()
.into_factory() .client_timeout(timeout)
.map_err(|err| err.into().error_response()); .h2(map_config(fac, move |_| app_cfg.clone()))
.rustls(config.clone())
}),
HttpVer::Both => builder.listen("test", tcp, move || {
let app_cfg = AppConfig::__priv_test_new(
false,
local_addr.to_string(),
local_addr,
);
HttpService::build() let fac = factory()
.client_timeout(timeout) .into_factory()
.finish(map_config(fac, move |_| app_cfg.clone())) .map_err(|err| err.into().error_response());
.rustls(config.clone())
}),
},
}
.expect("test server could not be created");
let srv = srv.run(); HttpService::build()
started_tx .client_timeout(timeout)
.send((System::current(), srv.handle(), local_addr)) .finish(map_config(fac, move |_| app_cfg.clone()))
.unwrap(); .rustls(config.clone())
}),
},
}
.expect("test server could not be created");
// drive server loop let srv = srv.run();
sys.block_on(srv).unwrap(); started_tx
.send((System::current(), srv.handle(), local_addr))
.unwrap();
// start system event loop // drive server loop
sys.run().unwrap(); srv.await.unwrap();
// notify TestServer that server and system have shut down
// all thread managed resources should be dropped at this point
});
// notify TestServer that server and system have shut down
// all thread managed resources should be dropped at this point
let _ = thread_stop_tx.send(()); let _ = thread_stop_tx.send(());
}); });