actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2025-06-30 20:04:26 +02:00
Code Issues Releases Wiki Activity

clippy warnings

Browse Source
This commit is contained in:
Nikolay Kim
2018-04-29 09:09:08 -07:00
parent d98d723f97
commit c72d1381a6
62 changed files with 1742 additions and 818 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/server/channel.rs
View File

@ -1,6 +1,6 @@
use std::net::{Shutdown, SocketAddr};
use std::rc::Rc;
use std::{io, mem, ptr, time};
use std::{io, ptr, time};
use bytes::{Buf, BufMut, Bytes, BytesMut};
use futures::{Async, Future, Poll};
@ -93,12 +93,12 @@ where
let el = self as *mut _;
self.node = Some(Node::new(el));
let _ = match self.proto {
Some(HttpProtocol::H1(ref mut h1)) => {
self.node.as_ref().map(|n| h1.settings().head().insert(n))
}
Some(HttpProtocol::H2(ref mut h2)) => {
self.node.as_ref().map(|n| h2.settings().head().insert(n))
}
Some(HttpProtocol::H1(ref mut h1)) => self.node
.as_ref()
.map(|n| h1.settings().head().insert(n)),
Some(HttpProtocol::H2(ref mut h2)) => self.node
.as_ref()
.map(|n| h2.settings().head().insert(n)),
Some(HttpProtocol::Unknown(ref mut settings, _, _, _)) => {
self.node.as_ref().map(|n| settings.head().insert(n))
}
@ -168,8 +168,9 @@ where
if let Some(HttpProtocol::Unknown(settings, addr, io, buf)) = self.proto.take() {
match kind {
ProtocolKind::Http1 => {
self.proto =
Some(HttpProtocol::H1(h1::Http1::new(settings, io, addr, buf)));
self.proto = Some(HttpProtocol::H1(h1::Http1::new(
settings, io, addr, buf,
)));
return self.poll();
}
ProtocolKind::Http2 => {
@ -210,10 +211,11 @@ impl<T> Node<T> {
&mut *(next2.as_ref().unwrap() as *const _ as *mut _);
n.prev = Some(next as *const _ as *mut _);
}
let slf: &mut Node<T> = mem::transmute(self);
let slf: &mut Node<T> = &mut *(self as *const _ as *mut _);
slf.next = Some(next as *const _ as *mut _);
let next: &mut Node<T> = mem::transmute(next);
let next: &mut Node<T> = &mut *(next as *const _ as *mut _);
next.prev = Some(slf as *const _ as *mut _);
}
}
@ -249,12 +251,12 @@ impl Node<()> {
loop {
if let Some(n) = next {
unsafe {
let n: &Node<()> = mem::transmute(n.as_ref().unwrap());
let n: &Node<()> = &*(n.as_ref().unwrap() as *const _);
next = n.next.as_ref();
if !n.element.is_null() {
let ch: &mut HttpChannel<T, H> =
mem::transmute(&mut *(n.element as *mut _));
&mut *(&mut *(n.element as *mut _) as *mut () as *mut _);
ch.shutdown();
}
}
@ -278,9 +280,7 @@ where
T: AsyncRead + AsyncWrite + 'static,
{
pub fn new(io: T) -> Self {
WrapperStream {
io,
}
WrapperStream { io }
}
}

13
src/server/encoding.rs
View File

@ -763,7 +763,8 @@ impl TransferEncoding {
return Ok(*remaining == 0);
}
let len = cmp::min(*remaining, msg.len() as u64);
self.buffer.extend(msg.take().split_to(len as usize).into());
self.buffer
.extend(msg.take().split_to(len as usize).into());
*remaining -= len as u64;
Ok(*remaining == 0)
@ -855,8 +856,10 @@ impl AcceptEncoding {
/// Parse a raw Accept-Encoding header value into an ordered list.
pub fn parse(raw: &str) -> ContentEncoding {
let mut encodings: Vec<_> =
raw.replace(' ', "").split(',').map(|l| AcceptEncoding::new(l)).collect();
let mut encodings: Vec<_> = raw.replace(' ', "")
.split(',')
.map(|l| AcceptEncoding::new(l))
.collect();
encodings.sort();
for enc in encodings {
@ -876,7 +879,9 @@ mod tests {
fn test_chunked_te() {
let bytes = SharedBytes::default();
let mut enc = TransferEncoding::chunked(bytes.clone());
assert!(!enc.encode(Binary::from(b"test".as_ref())).ok().unwrap());
assert!(!enc.encode(Binary::from(b"test".as_ref()))
.ok()
.unwrap());
assert!(enc.encode(Binary::from(b"".as_ref())).ok().unwrap());
assert_eq!(
bytes.get_mut().take().freeze(),

130
src/server/h1.rs
View File

@ -1,10 +1,8 @@
#![cfg_attr(feature = "cargo-clippy", allow(redundant_field_names))]
use std::collections::VecDeque;
use std::io;
use std::net::SocketAddr;
use std::rc::Rc;
use std::time::Duration;
use std::{io, mem};
use actix::Arbiter;
use bytes::{BufMut, BytesMut};
@ -153,28 +151,25 @@ where
if !self.flags.intersects(Flags::ERROR)
&& self.tasks.len() < MAX_PIPELINED_MESSAGES && self.can_read()
{
match self.read() {
Ok(true) | Err(_) => {
// notify all tasks
self.stream.disconnected();
for entry in &mut self.tasks {
entry.pipe.disconnected()
}
// kill keepalive
self.flags.remove(Flags::KEEPALIVE);
self.keepalive_timer.take();
// on parse error, stop reading stream but tasks need to be
// completed
self.flags.insert(Flags::ERROR);
if let Some(ref mut payload) = self.payload {
payload.set_error(PayloadError::Incomplete);
}
if self.read() {
// notify all tasks
self.stream.disconnected();
for entry in &mut self.tasks {
entry.pipe.disconnected()
}
Ok(false) => {
self.parse();
// kill keepalive
self.flags.remove(Flags::KEEPALIVE);
self.keepalive_timer.take();
// on parse error, stop reading stream but tasks need to be
// completed
self.flags.insert(Flags::ERROR);
if let Some(ref mut payload) = self.payload {
payload.set_error(PayloadError::Incomplete);
}
} else {
self.parse();
}
}
}
@ -186,7 +181,7 @@ where
let mut io = false;
let mut idx = 0;
while idx < self.tasks.len() {
let item: &mut Entry = unsafe { mem::transmute(&mut self.tasks[idx]) };
let item: &mut Entry = unsafe { &mut *(&mut self.tasks[idx] as *mut _) };
// only one task can do io operation in http/1
if !io && !item.flags.contains(EntryFlags::EOF) {
@ -210,7 +205,8 @@ where
self.stream.reset();
if ready {
item.flags.insert(EntryFlags::EOF | EntryFlags::FINISHED);
item.flags
.insert(EntryFlags::EOF | EntryFlags::FINISHED);
} else {
item.flags.insert(EntryFlags::FINISHED);
}
@ -252,7 +248,10 @@ where
// cleanup finished tasks
let max = self.tasks.len() >= MAX_PIPELINED_MESSAGES;
while !self.tasks.is_empty() {
if self.tasks[0].flags.contains(EntryFlags::EOF | EntryFlags::FINISHED) {
if self.tasks[0]
.flags
.contains(EntryFlags::EOF | EntryFlags::FINISHED)
{
self.tasks.pop_front();
} else {
break;
@ -277,7 +276,7 @@ where
// deal with keep-alive
if self.tasks.is_empty() {
// no keep-alive situations
// no keep-alive
if self.flags.contains(Flags::ERROR)
|| (!self.flags.contains(Flags::KEEPALIVE)
|| !self.settings.keep_alive_enabled())
@ -304,10 +303,7 @@ where
pub fn parse(&mut self) {
'outer: loop {
match self.decoder.decode(&mut self.buf, &self.settings) {
Ok(Some(Message::Message {
msg,
payload,
})) => {
Ok(Some(Message::Message { msg, payload })) => {
self.flags.insert(Flags::STARTED);
if payload {
@ -377,7 +373,7 @@ where
}
#[inline]
fn read(&mut self) -> io::Result<bool> {
fn read(&mut self) -> bool {
loop {
unsafe {
if self.buf.remaining_mut() < LW_BUFFER_SIZE {
@ -386,16 +382,16 @@ where
match self.stream.get_mut().read(self.buf.bytes_mut()) {
Ok(n) => {
if n == 0 {
return Ok(true);
return true;
} else {
self.buf.advance_mut(n);
}
}
Err(e) => {
return if e.kind() == io::ErrorKind::WouldBlock {
Ok(false)
false
} else {
Err(e)
true
};
}
}
@ -420,19 +416,13 @@ mod tests {
impl Message {
fn message(self) -> SharedHttpInnerMessage {
match self {
Message::Message {
msg,
payload: _,
} => msg,
Message::Message { msg, payload: _ } => msg,
_ => panic!("error"),
}
}
fn is_payload(&self) -> bool {
match *self {
Message::Message {
msg: _,
payload,
} => payload,
Message::Message { msg: _, payload } => payload,
_ => panic!("error"),
}
}
@ -628,7 +618,10 @@ mod tests {
assert_eq!(req.version(), Version::HTTP_11);
assert_eq!(*req.method(), Method::GET);
assert_eq!(req.path(), "/test");
assert_eq!(req.headers().get("test").unwrap().as_bytes(), b"value");
assert_eq!(
req.headers().get("test").unwrap().as_bytes(),
b"value"
);
}
Ok(_) | Err(_) => unreachable!("Error during parsing http request"),
}
@ -850,7 +843,12 @@ mod tests {
assert!(!req.keep_alive());
assert!(req.upgrade());
assert_eq!(
reader.decode(&mut buf, &settings).unwrap().unwrap().chunk().as_ref(),
reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.chunk()
.as_ref(),
b"some raw data"
);
}
@ -907,14 +905,30 @@ mod tests {
buf.extend(b"4\r\ndata\r\n4\r\nline\r\n0\r\n\r\n");
assert_eq!(
reader.decode(&mut buf, &settings).unwrap().unwrap().chunk().as_ref(),
reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.chunk()
.as_ref(),
b"data"
);
assert_eq!(
reader.decode(&mut buf, &settings).unwrap().unwrap().chunk().as_ref(),
reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.chunk()
.as_ref(),
b"line"
);
assert!(reader.decode(&mut buf, &settings).unwrap().unwrap().eof());
assert!(
reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.eof()
);
}
#[test]
@ -995,7 +1009,13 @@ mod tests {
assert!(reader.decode(&mut buf, &settings).unwrap().is_none());
buf.extend(b"\r\n");
assert!(reader.decode(&mut buf, &settings).unwrap().unwrap().eof());
assert!(
reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.eof()
);
}
#[test]
@ -1013,9 +1033,17 @@ mod tests {
assert!(req.chunked().unwrap());
buf.extend(b"4;test\r\ndata\r\n4\r\nline\r\n0\r\n\r\n"); // test: test\r\n\r\n")
let chunk = reader.decode(&mut buf, &settings).unwrap().unwrap().chunk();
let chunk = reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.chunk();
assert_eq!(chunk, Bytes::from_static(b"data"));
let chunk = reader.decode(&mut buf, &settings).unwrap().unwrap().chunk();
let chunk = reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.chunk();
assert_eq!(chunk, Bytes::from_static(b"line"));
let msg = reader.decode(&mut buf, &settings).unwrap().unwrap();
assert!(msg.eof());

43
src/server/h1decoder.rs
View File

@ -41,9 +41,7 @@ impl From<io::Error> for DecoderError {
impl H1Decoder {
pub fn new() -> H1Decoder {
H1Decoder {
decoder: None,
}
H1Decoder { decoder: None }
}
pub fn decode<H>(
@ -61,7 +59,9 @@ impl H1Decoder {
}
}
match self.parse_message(src, settings).map_err(DecoderError::Error)? {
match self.parse_message(src, settings)
.map_err(DecoderError::Error)?
{
Async::Ready((msg, decoder)) => {
if let Some(decoder) = decoder {
self.decoder = Some(decoder);
@ -415,9 +415,10 @@ impl ChunkedState {
match byte!(rdr) {
b'\n' if *size > 0 => Ok(Async::Ready(ChunkedState::Body)),
b'\n' if *size == 0 => Ok(Async::Ready(ChunkedState::EndCr)),
_ => {
Err(io::Error::new(io::ErrorKind::InvalidInput, "Invalid chunk size LF"))
}
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Invalid chunk size LF",
)),
}
}
@ -450,33 +451,37 @@ impl ChunkedState {
fn read_body_cr(rdr: &mut BytesMut) -> Poll<ChunkedState, io::Error> {
match byte!(rdr) {
b'\r' => Ok(Async::Ready(ChunkedState::BodyLf)),
_ => {
Err(io::Error::new(io::ErrorKind::InvalidInput, "Invalid chunk body CR"))
}
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Invalid chunk body CR",
)),
}
}
fn read_body_lf(rdr: &mut BytesMut) -> Poll<ChunkedState, io::Error> {
match byte!(rdr) {
b'\n' => Ok(Async::Ready(ChunkedState::Size)),
_ => {
Err(io::Error::new(io::ErrorKind::InvalidInput, "Invalid chunk body LF"))
}
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Invalid chunk body LF",
)),
}
}
fn read_end_cr(rdr: &mut BytesMut) -> Poll<ChunkedState, io::Error> {
match byte!(rdr) {
b'\r' => Ok(Async::Ready(ChunkedState::EndLf)),
_ => {
Err(io::Error::new(io::ErrorKind::InvalidInput, "Invalid chunk end CR"))
}
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Invalid chunk end CR",
)),
}
}
fn read_end_lf(rdr: &mut BytesMut) -> Poll<ChunkedState, io::Error> {
match byte!(rdr) {
b'\n' => Ok(Async::Ready(ChunkedState::End)),
_ => {
Err(io::Error::new(io::ErrorKind::InvalidInput, "Invalid chunk end LF"))
}
_ => Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Invalid chunk end LF",
)),
}
}
}

6
src/server/h1writer.rs
View File

@ -117,7 +117,8 @@ impl<T: AsyncWrite, H: 'static> Writer for H1Writer<T, H> {
let version = msg.version().unwrap_or_else(|| req.version);
if msg.upgrade() {
self.flags.insert(Flags::UPGRADE);
msg.headers_mut().insert(CONNECTION, HeaderValue::from_static("upgrade"));
msg.headers_mut()
.insert(CONNECTION, HeaderValue::from_static("upgrade"));
}
// keep-alive
else if self.flags.contains(Flags::KEEPALIVE) {
@ -126,7 +127,8 @@ impl<T: AsyncWrite, H: 'static> Writer for H1Writer<T, H> {
.insert(CONNECTION, HeaderValue::from_static("keep-alive"));
}
} else if version >= Version::HTTP_11 {
msg.headers_mut().insert(CONNECTION, HeaderValue::from_static("close"));
msg.headers_mut()
.insert(CONNECTION, HeaderValue::from_static("close"));
}
let body = msg.replace_body(Body::Empty);

7
src/server/h2writer.rs
View File

@ -107,7 +107,8 @@ impl<H: 'static> Writer for H2Writer<H> {
);
}
Body::Empty => {
msg.headers_mut().insert(CONTENT_LENGTH, HeaderValue::from_static("0"));
msg.headers_mut()
.insert(CONTENT_LENGTH, HeaderValue::from_static("0"));
}
_ => (),
}
@ -119,7 +120,9 @@ impl<H: 'static> Writer for H2Writer<H> {
resp.headers_mut().insert(key, value.clone());
}
match self.respond.send_response(resp, self.flags.contains(Flags::EOF)) {
match self.respond
.send_response(resp, self.flags.contains(Flags::EOF))
{
Ok(stream) => self.stream = Some(stream),
Err(_) => return Err(io::Error::new(io::ErrorKind::Other, "err")),
}

55
src/server/helpers.rs
View File

@ -96,8 +96,9 @@ const DEC_DIGITS_LUT: &[u8] = b"0001020304050607080910111213141516171819\
8081828384858687888990919293949596979899";
pub(crate) fn write_status_line(version: Version, mut n: u16, bytes: &mut BytesMut) {
let mut buf: [u8; 13] =
[b'H', b'T', b'T', b'P', b'/', b'1', b'.', b'1', b' ', b' ', b' ', b' ', b' '];
let mut buf: [u8; 13] = [
b'H', b'T', b'T', b'P', b'/', b'1', b'.', b'1', b' ', b' ', b' ', b' ', b' ',
];
match version {
Version::HTTP_2 => buf[5] = b'2',
Version::HTTP_10 => buf[7] = b'0',
@ -250,33 +251,63 @@ mod tests {
let mut bytes = BytesMut::new();
bytes.reserve(50);
write_content_length(0, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 0\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 0\r\n"[..]
);
bytes.reserve(50);
write_content_length(9, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 9\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 9\r\n"[..]
);
bytes.reserve(50);
write_content_length(10, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 10\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 10\r\n"[..]
);
bytes.reserve(50);
write_content_length(99, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 99\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 99\r\n"[..]
);
bytes.reserve(50);
write_content_length(100, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 100\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 100\r\n"[..]
);
bytes.reserve(50);
write_content_length(101, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 101\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 101\r\n"[..]
);
bytes.reserve(50);
write_content_length(998, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 998\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 998\r\n"[..]
);
bytes.reserve(50);
write_content_length(1000, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 1000\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 1000\r\n"[..]
);
bytes.reserve(50);
write_content_length(1001, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 1001\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 1001\r\n"[..]
);
bytes.reserve(50);
write_content_length(5909, &mut bytes);
assert_eq!(bytes.take().freeze(), b"\r\ncontent-length: 5909\r\n"[..]);
assert_eq!(
bytes.take().freeze(),
b"\r\ncontent-length: 5909\r\n"[..]
);
}
}

5
src/server/settings.rs
View File

@ -255,7 +255,10 @@ mod tests {
#[test]
fn test_date_len() {
assert_eq!(DATE_VALUE_LENGTH, "Sun, 06 Nov 1994 08:49:37 GMT".len());
assert_eq!(
DATE_VALUE_LENGTH,
"Sun, 06 Nov 1994 08:49:37 GMT".len()
);
}
#[test]

60
src/server/srv.rs
View File

@ -219,7 +219,10 @@ where
if let Some(e) = err.take() {
Err(e)
} else {
Err(io::Error::new(io::ErrorKind::Other, "Can not bind to address."))
Err(io::Error::new(
io::ErrorKind::Other,
"Can not bind to address.",
))
}
} else {
Ok(self)
@ -375,7 +378,10 @@ impl<H: IntoHttpHandler> HttpServer<H> {
/// Start listening for incoming tls connections.
pub fn start_tls(mut self, acceptor: TlsAcceptor) -> io::Result<Addr<Syn, Self>> {
if self.sockets.is_empty() {
Err(io::Error::new(io::ErrorKind::Other, "No socket addresses are bound"))
Err(io::Error::new(
io::ErrorKind::Other,
"No socket addresses are bound",
))
} else {
let (tx, rx) = mpsc::unbounded();
let addrs: Vec<(net::SocketAddr, net::TcpListener)> =
@ -424,7 +430,10 @@ impl<H: IntoHttpHandler> HttpServer<H> {
mut self, mut builder: SslAcceptorBuilder,
) -> io::Result<Addr<Syn, Self>> {
if self.sockets.is_empty() {
Err(io::Error::new(io::ErrorKind::Other, "No socket addresses are bound"))
Err(io::Error::new(
io::ErrorKind::Other,
"No socket addresses are bound",
))
} else {
// alpn support
if !self.no_http2 {
@ -554,35 +563,17 @@ impl<H: IntoHttpHandler> Handler<signal::Signal> for HttpServer<H> {
signal::SignalType::Int => {
info!("SIGINT received, exiting");
self.exit = true;
Handler::<StopServer>::handle(
self,
StopServer {
graceful: false,
},
ctx,
);
Handler::<StopServer>::handle(self, StopServer { graceful: false }, ctx);
}
signal::SignalType::Term => {
info!("SIGTERM received, stopping");
self.exit = true;
Handler::<StopServer>::handle(
self,
StopServer {
graceful: true,
},
ctx,
);
Handler::<StopServer>::handle(self, StopServer { graceful: true }, ctx);
}
signal::SignalType::Quit => {
info!("SIGQUIT received, exiting");
self.exit = true;
Handler::<StopServer>::handle(
self,
StopServer {
graceful: false,
},
ctx,
);
Handler::<StopServer>::handle(self, StopServer { graceful: false }, ctx);
}
_ => (),
}
@ -706,9 +697,7 @@ impl<H: IntoHttpHandler> Handler<StopServer> for HttpServer<H> {
let tx2 = tx.clone();
worker
.1
.send(StopWorker {
graceful: dur,
})
.send(StopWorker { graceful: dur })
.into_actor(self)
.then(move |_, slf, ctx| {
slf.workers.pop();
@ -758,8 +747,9 @@ fn start_accept_thread(
// start accept thread
#[cfg_attr(feature = "cargo-clippy", allow(cyclomatic_complexity))]
let _ = thread::Builder::new().name(format!("Accept on {}", addr)).spawn(
move || {
let _ = thread::Builder::new()
.name(format!("Accept on {}", addr))
.spawn(move || {
const SRV: mio::Token = mio::Token(0);
const CMD: mio::Token = mio::Token(1);
@ -784,9 +774,12 @@ fn start_accept_thread(
}
// Start listening for incoming commands
if let Err(err) =
poll.register(&reg, CMD, mio::Ready::readable(), mio::PollOpt::edge())
{
if let Err(err) = poll.register(
&reg,
CMD,
mio::Ready::readable(),
mio::PollOpt::edge(),
) {
panic!("Can not register Registration: {}", err);
}
@ -917,8 +910,7 @@ fn start_accept_thread(
}
}
}
},
);
});
(readiness, tx)
}

82
src/server/worker.rs
View File

@ -77,7 +77,9 @@ impl<H: HttpHandler + 'static> Worker<H> {
fn update_time(&self, ctx: &mut Context<Self>) {
self.settings.update_date();
ctx.run_later(time::Duration::new(1, 0), |slf, ctx| slf.update_time(ctx));
ctx.run_later(time::Duration::new(1, 0), |slf, ctx| {
slf.update_time(ctx)
});
}
fn shutdown_timeout(
@ -122,7 +124,8 @@ where
if self.tcp_ka.is_some() && msg.io.set_keepalive(self.tcp_ka).is_err() {
error!("Can not set socket keep-alive option");
}
self.handler.handle(Rc::clone(&self.settings), &self.hnd, msg);
self.handler
.handle(Rc::clone(&self.settings), &self.hnd, msg);
}
}
@ -174,57 +177,52 @@ impl StreamHandlerType {
}
#[cfg(feature = "tls")]
StreamHandlerType::Tls(ref acceptor) => {
let Conn {
io,
peer,
http2,
} = msg;
let Conn { io, peer, http2 } = msg;
let _ = io.set_nodelay(true);
let io = TcpStream::from_stream(io, hnd)
.expect("failed to associate TCP stream");
hnd.spawn(TlsAcceptorExt::accept_async(acceptor, io).then(move |res| {
match res {
Ok(io) => {
Arbiter::handle().spawn(HttpChannel::new(h, io, peer, http2))
}
Err(err) => {
trace!("Error during handling tls connection: {}", err)
}
};
future::result(Ok(()))
}));
hnd.spawn(
TlsAcceptorExt::accept_async(acceptor, io).then(move |res| {
match res {
Ok(io) => Arbiter::handle()
.spawn(HttpChannel::new(h, io, peer, http2)),
Err(err) => {
trace!("Error during handling tls connection: {}", err)
}
};
future::result(Ok(()))
}),
);
}
#[cfg(feature = "alpn")]
StreamHandlerType::Alpn(ref acceptor) => {
let Conn {
io,
peer,
..
} = msg;
let Conn { io, peer, .. } = msg;
let _ = io.set_nodelay(true);
let io = TcpStream::from_stream(io, hnd)
.expect("failed to associate TCP stream");
hnd.spawn(SslAcceptorExt::accept_async(acceptor, io).then(move |res| {
match res {
Ok(io) => {
let http2 = if let Some(p) =
io.get_ref().ssl().selected_alpn_protocol()
{
p.len() == 2 && &p == b"h2"
} else {
false
};
Arbiter::handle()
.spawn(HttpChannel::new(h, io, peer, http2));
}
Err(err) => {
trace!("Error during handling tls connection: {}", err)
}
};
future::result(Ok(()))
}));
hnd.spawn(
SslAcceptorExt::accept_async(acceptor, io).then(move |res| {
match res {
Ok(io) => {
let http2 = if let Some(p) =
io.get_ref().ssl().selected_alpn_protocol()
{
p.len() == 2 && &p == b"h2"
} else {
false
};
Arbiter::handle()
.spawn(HttpChannel::new(h, io, peer, http2));
}
Err(err) => {
trace!("Error during handling tls connection: {}", err)
}
};
future::result(Ok(()))
}),
);
}
}
}