mirrors/actix-web
1
0
Fork 0
mirror of https://github.com/actix/actix-web.git synced 2024-07-03 20:45:46 +00:00
Code Issues Projects Releases Wiki Activity

use one thread for accept loop; refactor rust-tls support

Browse source
This commit is contained in:
Nikolay Kim 2018-07-31 19:51:26 -07:00
parent 2071ea0532
commit 58230b15b9
7 changed files with 406 additions and 232 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.travis.yml
View file

@ -32,12 +32,12 @@ script:
- |
if [[ "$TRAVIS_RUST_VERSION" != "stable" ]]; then
cargo clean
cargo test --features="alpn,tls" -- --nocapture
cargo test --features="alpn,tls,rust-tls" -- --nocapture
fi
- |
if [[ "$TRAVIS_RUST_VERSION" == "stable" ]]; then
RUSTFLAGS="--cfg procmacro2_semver_exempt" cargo install -f cargo-tarpaulin
cargo tarpaulin --features="alpn,tls" --out Xml --no-count
cargo tarpaulin --features="alpn,tls,rust-tls" --out Xml --no-count
bash <(curl -s https://codecov.io/bash)
echo "Uploaded code coverage"
fi
@ -46,7 +46,7 @@ script:
after_success:
- |
if [[ "$TRAVIS_OS_NAME" == "linux" && "$TRAVIS_PULL_REQUEST" = "false" && "$TRAVIS_BRANCH" == "master" && "$TRAVIS_RUST_VERSION" == "beta" ]]; then
cargo doc --features "alpn, tls, session" --no-deps &&
cargo doc --features "alpn, tls, rust-tls, session" --no-deps &&
echo "<meta http-equiv=refresh content=0;url=os_balloon/index.html>" > target/doc/index.html &&
git clone https://github.com/davisp/ghp-import.git &&
./ghp-import/ghp_import.py -n -p -f -m "Documentation upload" -r https://"$GH_TOKEN"@github.com/"$TRAVIS_REPO_SLUG.git" target/doc &&

439
src/server/accept.rs
View file

@ -1,22 +1,16 @@
use std::sync::mpsc as sync_mpsc;
use std::time::Duration;
use std::time::{Duration, Instant};
use std::{io, net, thread};
use futures::sync::mpsc;
use futures::{sync::mpsc, Future};
use mio;
use slab::Slab;
use tokio_timer::Delay;
#[cfg(feature = "tls")]
use native_tls::TlsAcceptor;
#[cfg(feature = "alpn")]
use openssl::ssl::{AlpnError, SslAcceptorBuilder};
#[cfg(feature = "rust-tls")]
use rustls::ServerConfig;
use actix::{msgs::Execute, Arbiter, System};
use super::srv::{ServerCommand, Socket};
use super::worker::{Conn, SocketInfo};
use super::worker::Conn;
pub(crate) enum Command {
Pause,
@ -25,169 +19,43 @@ pub(crate) enum Command {
Worker(usize, mpsc::UnboundedSender<Conn<net::TcpStream>>),
}
struct ServerSocketInfo {
addr: net::SocketAddr,
token: usize,
sock: mio::net::TcpListener,
timeout: Option<Instant>,
}
struct Accept {
poll: mio::Poll,
rx: sync_mpsc::Receiver<Command>,
sockets: Slab<ServerSocketInfo>,
workers: Vec<(usize, mpsc::UnboundedSender<Conn<net::TcpStream>>)>,
_reg: mio::Registration,
next: usize,
srv: mpsc::UnboundedSender<ServerCommand>,
timer: (mio::Registration, mio::SetReadiness),
}
const CMD: mio::Token = mio::Token(0);
const TIMER: mio::Token = mio::Token(1);
pub(crate) fn start_accept_thread(
token: usize, sock: Socket, srv: mpsc::UnboundedSender<ServerCommand>,
socks: Slab<SocketInfo>,
mut workers: Vec<(usize, mpsc::UnboundedSender<Conn<net::TcpStream>>)>,
socks: Vec<(usize, Socket)>, srv: mpsc::UnboundedSender<ServerCommand>,
workers: Vec<(usize, mpsc::UnboundedSender<Conn<net::TcpStream>>)>,
) -> (mio::SetReadiness, sync_mpsc::Sender<Command>) {
let (tx, rx) = sync_mpsc::channel();
let (reg, readiness) = mio::Registration::new2();
let sys = System::current();
// start accept thread
#[cfg_attr(feature = "cargo-clippy", allow(cyclomatic_complexity))]
let _ = thread::Builder::new()
.name(format!("Accept on {}", sock.addr))
.name("actix-web accept loop".to_owned())
.spawn(move || {
const SRV: mio::Token = mio::Token(0);
const CMD: mio::Token = mio::Token(1);
let addr = sock.addr;
let mut server = Some(
mio::net::TcpListener::from_std(sock.lst)
.expect("Can not create mio::net::TcpListener"),
);
// Create a poll instance
let poll = match mio::Poll::new() {
Ok(poll) => poll,
Err(err) => panic!("Can not create mio::Poll: {}", err),
};
// Start listening for incoming connections
if let Some(ref srv) = server {
if let Err(err) =
poll.register(srv, SRV, mio::Ready::readable(), mio::PollOpt::edge())
{
panic!("Can not register io: {}", err);
}
}
// Start listening for incoming commands
if let Err(err) =
poll.register(&reg, CMD, mio::Ready::readable(), mio::PollOpt::edge())
{
panic!("Can not register Registration: {}", err);
}
// Create storage for events
let mut events = mio::Events::with_capacity(128);
// Sleep on error
let sleep = Duration::from_millis(100);
let mut next = 0;
loop {
if let Err(err) = poll.poll(&mut events, None) {
panic!("Poll error: {}", err);
}
for event in events.iter() {
match event.token() {
SRV => if let Some(ref server) = server {
loop {
match server.accept_std() {
Ok((io, addr)) => {
let mut msg = Conn {
io,
token,
peer: Some(addr),
http2: false,
};
while !workers.is_empty() {
match workers[next].1.unbounded_send(msg) {
Ok(_) => (),
Err(err) => {
let _ = srv.unbounded_send(
ServerCommand::WorkerDied(
workers[next].0,
socks.clone(),
),
);
msg = err.into_inner();
workers.swap_remove(next);
if workers.is_empty() {
error!("No workers");
thread::sleep(sleep);
break;
} else if workers.len() <= next {
next = 0;
}
continue;
}
}
next = (next + 1) % workers.len();
break;
}
}
Err(ref e)
if e.kind() == io::ErrorKind::WouldBlock =>
{
break
}
Err(ref e) if connection_error(e) => continue,
Err(e) => {
error!("Error accepting connection: {}", e);
// sleep after error
thread::sleep(sleep);
break;
}
}
}
},
CMD => match rx.try_recv() {
Ok(cmd) => match cmd {
Command::Pause => if let Some(ref server) = server {
if let Err(err) = poll.deregister(server) {
error!(
"Can not deregister server socket {}",
err
);
} else {
info!(
"Paused accepting connections on {}",
addr
);
}
},
Command::Resume => {
if let Some(ref server) = server {
if let Err(err) = poll.register(
server,
SRV,
mio::Ready::readable(),
mio::PollOpt::edge(),
) {
error!("Can not resume socket accept process: {}", err);
} else {
info!("Accepting connections on {} has been resumed",
addr);
}
}
}
Command::Stop => {
if let Some(server) = server.take() {
let _ = poll.deregister(&server);
}
return;
}
Command::Worker(idx, addr) => {
workers.push((idx, addr));
}
},
Err(err) => match err {
sync_mpsc::TryRecvError::Empty => (),
sync_mpsc::TryRecvError::Disconnected => {
if let Some(server) = server.take() {
let _ = poll.deregister(&server);
}
return;
}
},
},
_ => unreachable!(),
}
}
}
System::set_current(sys);
Accept::new(reg, rx, socks, workers, srv).poll();
});
(readiness, tx)
@ -205,3 +73,244 @@ fn connection_error(e: &io::Error) -> bool {
|| e.kind() == io::ErrorKind::ConnectionAborted
|| e.kind() == io::ErrorKind::ConnectionReset
}
impl Accept {
fn new(
_reg: mio::Registration, rx: sync_mpsc::Receiver<Command>,
socks: Vec<(usize, Socket)>,
workers: Vec<(usize, mpsc::UnboundedSender<Conn<net::TcpStream>>)>,
srv: mpsc::UnboundedSender<ServerCommand>,
) -> Accept {
// Create a poll instance
let poll = match mio::Poll::new() {
Ok(poll) => poll,
Err(err) => panic!("Can not create mio::Poll: {}", err),
};
// Start listening for incoming commands
if let Err(err) =
poll.register(&_reg, CMD, mio::Ready::readable(), mio::PollOpt::edge())
{
panic!("Can not register Registration: {}", err);
}
// Start accept
let mut sockets = Slab::new();
for (stoken, sock) in socks {
let server = mio::net::TcpListener::from_std(sock.lst)
.expect("Can not create mio::net::TcpListener");
let entry = sockets.vacant_entry();
let token = entry.key();
// Start listening for incoming connections
if let Err(err) = poll.register(
&server,
mio::Token(token + 1000),
mio::Ready::readable(),
mio::PollOpt::edge(),
) {
panic!("Can not register io: {}", err);
}
entry.insert(ServerSocketInfo {
token: stoken,
addr: sock.addr,
sock: server,
timeout: None,
});
}
// Timer
let (tm, tmr) = mio::Registration::new2();
if let Err(err) =
poll.register(&tm, TIMER, mio::Ready::readable(), mio::PollOpt::edge())
{
panic!("Can not register Registration: {}", err);
}
Accept {
poll,
rx,
_reg,
sockets,
workers,
srv,
next: 0,
timer: (tm, tmr),
}
}
fn poll(&mut self) {
// Create storage for events
let mut events = mio::Events::with_capacity(128);
loop {
if let Err(err) = self.poll.poll(&mut events, None) {
panic!("Poll error: {}", err);
}
for event in events.iter() {
let token = event.token();
match token {
CMD => if !self.process_cmd() {
return;
},
TIMER => self.process_timer(),
_ => self.accept(token),
}
}
}
}
fn process_timer(&mut self) {
let now = Instant::now();
for (token, info) in self.sockets.iter_mut() {
if let Some(inst) = info.timeout.take() {
if now > inst {
if let Err(err) = self.poll.register(
&info.sock,
mio::Token(token + 1000),
mio::Ready::readable(),
mio::PollOpt::edge(),
) {
error!("Can not register server socket {}", err);
} else {
info!("Resume accepting connections on {}", info.addr);
}
} else {
info.timeout = Some(inst);
}
}
}
}
fn process_cmd(&mut self) -> bool {
loop {
match self.rx.try_recv() {
Ok(cmd) => match cmd {
Command::Pause => {
for (_, info) in self.sockets.iter_mut() {
if let Err(err) = self.poll.deregister(&info.sock) {
error!("Can not deregister server socket {}", err);
} else {
info!("Paused accepting connections on {}", info.addr);
}
}
}
Command::Resume => {
for (token, info) in self.sockets.iter() {
if let Err(err) = self.poll.register(
&info.sock,
mio::Token(token + 1000),
mio::Ready::readable(),
mio::PollOpt::edge(),
) {
error!("Can not resume socket accept process: {}", err);
} else {
info!(
"Accepting connections on {} has been resumed",
info.addr
);
}
}
}
Command::Stop => {
for (_, info) in self.sockets.iter() {
let _ = self.poll.deregister(&info.sock);
}
return false;
}
Command::Worker(idx, addr) => {
self.workers.push((idx, addr));
}
},
Err(err) => match err {
sync_mpsc::TryRecvError::Empty => break,
sync_mpsc::TryRecvError::Disconnected => {
for (_, info) in self.sockets.iter() {
let _ = self.poll.deregister(&info.sock);
}
return false;
}
},
}
}
true
}
fn accept(&mut self, token: mio::Token) {
let token = usize::from(token);
if token < 1000 {
return;
}
if let Some(info) = self.sockets.get_mut(token - 1000) {
loop {
match info.sock.accept_std() {
Ok((io, addr)) => {
let mut msg = Conn {
io,
token: info.token,
peer: Some(addr),
http2: false,
};
while !self.workers.is_empty() {
match self.workers[self.next].1.unbounded_send(msg) {
Ok(_) => (),
Err(err) => {
let _ = self.srv.unbounded_send(
ServerCommand::WorkerDied(
self.workers[self.next].0,
),
);
msg = err.into_inner();
self.workers.swap_remove(self.next);
if self.workers.is_empty() {
error!("No workers");
thread::sleep(Duration::from_millis(100));
break;
} else if self.workers.len() <= self.next {
self.next = 0;
}
continue;
}
}
self.next = (self.next + 1) % self.workers.len();
break;
}
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
Err(ref e) if connection_error(e) => continue,
Err(e) => {
error!("Error accepting connection: {}", e);
if let Err(err) = self.poll.deregister(&info.sock) {
error!("Can not deregister server socket {}", err);
}
// sleep after error
info.timeout = Some(Instant::now() + Duration::from_millis(500));
let r = self.timer.1.clone();
System::current().arbiter().do_send(Execute::new(
move || -> Result<(), ()> {
Arbiter::spawn(
Delay::new(
Instant::now() + Duration::from_millis(510),
).map_err(|_| ())
.and_then(move |_| {
let _ =
r.set_readiness(mio::Ready::readable());
Ok(())
}),
);
Ok(())
},
));
break;
}
}
}
}
}
}

6
src/server/mod.rs
View file

@ -315,10 +315,10 @@ impl IoStream for TlsStream<TcpStream> {
#[cfg(feature = "rust-tls")]
use rustls::{ClientSession, ServerSession};
#[cfg(feature = "rust-tls")]
use tokio_rustls::TlsStream;
use tokio_rustls::TlsStream as RustlsStream;
#[cfg(feature = "rust-tls")]
impl IoStream for TlsStream<TcpStream, ClientSession> {
impl IoStream for RustlsStream<TcpStream, ClientSession> {
#[inline]
fn shutdown(&mut self, _how: Shutdown) -> io::Result<()> {
let _ = <Self as AsyncWrite>::shutdown(self);
@ -337,7 +337,7 @@ impl IoStream for TlsStream<TcpStream, ClientSession> {
}
#[cfg(feature = "rust-tls")]
impl IoStream for TlsStream<TcpStream, ServerSession> {
impl IoStream for RustlsStream<TcpStream, ServerSession> {
#[inline]
fn shutdown(&mut self, _how: Shutdown) -> io::Result<()> {
let _ = <Self as AsyncWrite>::shutdown(self);

57
src/server/srv.rs
View file

@ -46,14 +46,6 @@ fn configure_alpn(builder: &mut SslAcceptorBuilder) -> io::Result<()> {
Ok(())
}
#[cfg(all(feature = "rust-tls", not(feature = "alpn")))]
fn configure_alpn(builder: &mut Arc<ServerConfig>) -> io::Result<()> {
Arc::<ServerConfig>::get_mut(builder)
.unwrap()
.set_protocols(&vec!["h2".to_string(), "http/1.1".to_string()]);
Ok(())
}
/// An HTTP Server
pub struct HttpServer<H>
where
@ -68,7 +60,11 @@ where
#[cfg_attr(feature = "cargo-clippy", allow(type_complexity))]
workers: Vec<(usize, Addr<Worker<H::Handler>>)>,
sockets: Vec<Socket>,
accept: Vec<(mio::SetReadiness, sync_mpsc::Sender<Command>)>,
accept: Option<(
mio::SetReadiness,
sync_mpsc::Sender<Command>,
Slab<SocketInfo>,
)>,
exit: bool,
shutdown_timeout: u16,
signals: Option<Addr<signal::ProcessSignals>>,
@ -77,7 +73,7 @@ where
}
pub(crate) enum ServerCommand {
WorkerDied(usize, Slab<SocketInfo>),
WorkerDied(usize),
}
impl<H> Actor for HttpServer<H>
@ -114,7 +110,7 @@ where
factory: Arc::new(f),
workers: Vec::new(),
sockets: Vec::new(),
accept: Vec::new(),
accept: None,
exit: false,
shutdown_timeout: 30,
signals: None,
@ -280,22 +276,22 @@ where
Ok(self)
}
#[cfg(all(feature = "rust-tls", not(feature = "alpn")))]
#[cfg(feature = "rust-tls")]
/// Use listener for accepting incoming tls connection requests
///
/// This method sets alpn protocols to "h2" and "http/1.1"
pub fn listen_ssl(
mut self, lst: net::TcpListener, mut builder: Arc<ServerConfig>,
pub fn listen_rustls(
mut self, lst: net::TcpListener, mut builder: ServerConfig,
) -> io::Result<Self> {
// alpn support
if !self.no_http2 {
configure_alpn(&mut builder)?;
builder.set_protocols(&vec!["h2".to_string(), "http/1.1".to_string()]);
}
let addr = lst.local_addr().unwrap();
self.sockets.push(Socket {
addr,
lst,
tp: StreamHandlerType::Rustls(builder.clone()),
tp: StreamHandlerType::Rustls(Arc::new(builder)),
});
Ok(self)
}
@ -378,20 +374,21 @@ where
Ok(self)
}
#[cfg(all(feature = "rust-tls", not(feature = "alpn")))]
#[cfg(feature = "rust-tls")]
/// Start listening for incoming tls connections.
///
/// This method sets alpn protocols to "h2" and "http/1.1"
pub fn bind_ssl<S: net::ToSocketAddrs>(
mut self, addr: S, mut builder: Arc<ServerConfig>,
pub fn bind_rustls<S: net::ToSocketAddrs>(
mut self, addr: S, mut builder: ServerConfig,
) -> io::Result<Self> {
// alpn support
if !self.no_http2 {
configure_alpn(&mut builder)?;
builder.set_protocols(&vec!["h2".to_string(), "http/1.1".to_string()]);
}
let builder = Arc::new(builder);
let sockets = self.bind2(addr)?;
self.sockets.extend(sockets.into_iter().map(|mut s| {
self.sockets.extend(sockets.into_iter().map(move |mut s| {
s.tp = StreamHandlerType::Rustls(builder.clone());
s
}));
@ -487,17 +484,12 @@ impl<H: IntoHttpHandler> HttpServer<H> {
let settings = ServerSettings::new(Some(addrs[0].1.addr), &self.host, false);
let workers = self.start_workers(&settings, &socks);
// start acceptors threads
for (token, sock) in addrs {
// start accept thread
for (_, sock) in &addrs {
info!("Starting server on http://{}", sock.addr);
self.accept.push(start_accept_thread(
token,
sock,
tx.clone(),
socks.clone(),
workers.clone(),
));
}
let (r, cmd) = start_accept_thread(addrs, tx.clone(), workers.clone());
self.accept = Some((r, cmd, socks));
// start http server actor
let signals = self.subscribe_to_signals();
@ -672,7 +664,7 @@ impl<H: IntoHttpHandler> StreamHandler<ServerCommand, ()> for HttpServer<H> {
fn handle(&mut self, msg: ServerCommand, _: &mut Context<Self>) {
match msg {
ServerCommand::WorkerDied(idx, socks) => {
ServerCommand::WorkerDied(idx) => {
let mut found = false;
for i in 0..self.workers.len() {
if self.workers[i].0 == idx {
@ -700,6 +692,7 @@ impl<H: IntoHttpHandler> StreamHandler<ServerCommand, ()> for HttpServer<H> {
let ka = self.keep_alive;
let factory = Arc::clone(&self.factory);
let host = self.host.clone();
let socks = self.accept.as_ref().unwrap().2.clone();
let addr = socks[0].addr;
let addr = Arbiter::start(move |ctx: &mut Context<_>| {
@ -709,7 +702,7 @@ impl<H: IntoHttpHandler> StreamHandler<ServerCommand, ()> for HttpServer<H> {
ctx.add_message_stream(rx);
Worker::new(apps, socks, ka, settings)
});
for item in &self.accept {
if let Some(ref item) = &self.accept {
let _ = item.1.send(Command::Worker(new_idx, tx.clone()));
let _ = item.0.set_readiness(mio::Ready::readable());
}

65
src/test.rs
View file

@ -15,10 +15,10 @@ use tokio::runtime::current_thread::Runtime;
#[cfg(feature = "alpn")]
use openssl::ssl::SslAcceptorBuilder;
#[cfg(feature = "rust-tls")]
#[cfg(all(feature = "rust-tls"))]
use rustls::ServerConfig;
#[cfg(feature = "rust-tls")]
use std::sync::Arc;
//#[cfg(all(feature = "rust-tls"))]
//use std::sync::Arc;
use application::{App, HttpApplication};
use body::Binary;
@ -144,7 +144,7 @@ impl TestServer {
builder.set_verify(SslVerifyMode::NONE);
ClientConnector::with_connector(builder.build()).start()
}
#[cfg(feature = "rust-tls")]
#[cfg(all(feature = "rust-tls", not(feature = "alpn")))]
{
use rustls::ClientConfig;
use std::fs::File;
@ -256,7 +256,7 @@ pub struct TestServerBuilder<S> {
#[cfg(feature = "alpn")]
ssl: Option<SslAcceptorBuilder>,
#[cfg(feature = "rust-tls")]
ssl: Option<Arc<ServerConfig>>,
rust_ssl: Option<ServerConfig>,
}
impl<S: 'static> TestServerBuilder<S> {
@ -267,8 +267,10 @@ impl<S: 'static> TestServerBuilder<S> {
{
TestServerBuilder {
state: Box::new(state),
#[cfg(any(feature = "alpn", feature = "rust-tls"))]
#[cfg(feature = "alpn")]
ssl: None,
#[cfg(feature = "rust-tls")]
rust_ssl: None,
}
}
@ -280,9 +282,9 @@ impl<S: 'static> TestServerBuilder<S> {
}
#[cfg(feature = "rust-tls")]
/// Create ssl server
pub fn ssl(mut self, ssl: Arc<ServerConfig>) -> Self {
self.ssl = Some(ssl);
/// Create rust tls server
pub fn rustls(mut self, ssl: ServerConfig) -> Self {
self.rust_ssl = Some(ssl);
self
}
@ -294,41 +296,56 @@ impl<S: 'static> TestServerBuilder<S> {
{
let (tx, rx) = mpsc::channel();
#[cfg(any(feature = "alpn", feature = "rust-tls"))]
let ssl = self.ssl.is_some();
#[cfg(not(any(feature = "alpn", feature = "rust-tls")))]
let ssl = false;
let mut has_ssl = false;
#[cfg(feature = "alpn")]
{
has_ssl = has_ssl || self.ssl.is_some();
}
#[cfg(feature = "rust-tls")]
{
has_ssl = has_ssl || self.rust_ssl.is_some();
}
// run server in separate thread
thread::spawn(move || {
let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap();
let local_addr = tcp.local_addr().unwrap();
let addr = TestServer::unused_addr();
let sys = System::new("actix-test-server");
let state = self.state;
let srv = HttpServer::new(move || {
let mut srv = HttpServer::new(move || {
let mut app = TestApp::new(state());
config(&mut app);
vec![app]
}).workers(1)
.disable_signals();
tx.send((System::current(), local_addr, TestServer::get_conn()))
tx.send((System::current(), addr, TestServer::get_conn()))
.unwrap();
#[cfg(any(feature = "alpn", feature = "rust-tls"))]
#[cfg(feature = "alpn")]
{
let ssl = self.ssl.take();
if let Some(ssl) = ssl {
srv.listen_ssl(tcp, ssl).unwrap().start();
} else {
srv.listen(tcp).start();
let tcp = net::TcpListener::bind(addr).unwrap();
srv = srv.listen_ssl(tcp, ssl).unwrap();
}
}
#[cfg(not(any(feature = "alpn", feature = "rust-tls")))]
#[cfg(feature = "rust-tls")]
{
srv.listen(tcp).start();
let ssl = self.rust_ssl.take();
if let Some(ssl) = ssl {
let tcp = net::TcpListener::bind(addr).unwrap();
srv = srv.listen_rustls(tcp, ssl).unwrap();
}
}
if !has_ssl {
let tcp = net::TcpListener::bind(addr).unwrap();
srv = srv.listen(tcp);
}
srv.start();
sys.run();
});
@ -336,8 +353,8 @@ impl<S: 'static> TestServerBuilder<S> {
System::set_current(system);
TestServer {
addr,
ssl,
conn,
ssl: has_ssl,
rt: Runtime::new().unwrap(),
}
}

56
tests/test_server.rs
View file

@ -153,6 +153,62 @@ fn test_shutdown() {
let _ = sys.stop();
}
#[test]
#[cfg(unix)]
fn test_panic() {
let _ = test::TestServer::unused_addr();
let (tx, rx) = mpsc::channel();
thread::spawn(|| {
System::run(move || {
let srv = server::new(|| {
App::new()
.resource("/panic", |r| {
r.method(http::Method::GET).f(|_| -> &'static str {
panic!("error");
});
})
.resource("/", |r| {
r.method(http::Method::GET).f(|_| HttpResponse::Ok())
})
}).workers(1);
let srv = srv.bind("127.0.0.1:0").unwrap();
let addr = srv.addrs()[0];
srv.start();
let _ = tx.send((addr, System::current()));
});
});
let (addr, sys) = rx.recv().unwrap();
System::set_current(sys.clone());
let mut rt = Runtime::new().unwrap();
{
let req = client::ClientRequest::get(format!("http://{}/panic", addr).as_str())
.finish()
.unwrap();
let response = rt.block_on(req.send());
assert!(response.is_err());
}
{
let req = client::ClientRequest::get(format!("http://{}/", addr).as_str())
.finish()
.unwrap();
let response = rt.block_on(req.send());
assert!(response.is_err());
}
{
let req = client::ClientRequest::get(format!("http://{}/", addr).as_str())
.finish()
.unwrap();
let response = rt.block_on(req.send()).unwrap();
assert!(response.status().is_success());
}
let _ = sys.stop();
}
#[test]
fn test_simple() {
let mut srv = test::TestServer::new(|app| app.handler(|_| HttpResponse::Ok()));

9
tests/test_ws.rs
View file

@ -277,13 +277,12 @@ fn test_ws_server_ssl() {
#[test]
#[cfg(feature = "rust-tls")]
fn test_ws_server_ssl() {
fn test_ws_server_rust_tls() {
extern crate rustls;
use rustls::{ServerConfig, NoClientAuth};
use rustls::internal::pemfile::{certs, rsa_private_keys};
use std::io::BufReader;
use std::sync::Arc;
use rustls::{NoClientAuth, ServerConfig};
use std::fs::File;
use std::io::BufReader;
// load ssl keys
let mut config = ServerConfig::new(NoClientAuth::new());
@ -293,7 +292,7 @@ fn test_ws_server_ssl() {
let mut keys = rsa_private_keys(key_file).unwrap();
config.set_single_cert(cert_chain, keys.remove(0)).unwrap();
let mut srv = test::TestServer::build().ssl(Arc::new(config)).start(|app| {
let mut srv = test::TestServer::build().rustls(config).start(|app| {
app.handler(|req| {
ws::start(
req,