actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-06-25 22:49:21 +02:00
Code Issues Releases Wiki Activity

upgrade to tokio 0.2

Browse Source
This commit is contained in:
Nikolay Kim
2019-12-05 23:35:43 +06:00
parent b45c6cd66b
commit 205a964d8f
72 changed files with 764 additions and 555 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
actix-http/src/h1/decoder.rs
View File

@ -1,3 +1,4 @@
use std::convert::TryFrom;
use std::io;
use std::marker::PhantomData;
use std::mem::MaybeUninit;
@ -6,7 +7,7 @@ use std::task::Poll;
use actix_codec::Decoder;
use bytes::{Bytes, BytesMut};
use http::header::{HeaderName, HeaderValue};
use http::{header, HttpTryFrom, Method, StatusCode, Uri, Version};
use http::{header, Method, StatusCode, Uri, Version};
use httparse;
use log::{debug, error, trace};
@ -79,8 +80,8 @@ pub(crate) trait MessageType: Sized {
// Unsafe: httparse check header value for valid utf-8
let value = unsafe {
HeaderValue::from_shared_unchecked(
slice.slice(idx.value.0, idx.value.1),
HeaderValue::from_maybe_shared_unchecked(
slice.slice(idx.value.0..idx.value.1),
)
};
match name {
@ -428,7 +429,7 @@ impl Decoder for PayloadDecoder {
let len = src.len() as u64;
let buf;
if *remaining > len {
buf = src.take().freeze();
buf = src.split().freeze();
*remaining -= len;
} else {
buf = src.split_to(*remaining as usize).freeze();
@ -463,7 +464,7 @@ impl Decoder for PayloadDecoder {
if src.is_empty() {
Ok(None)
} else {
Ok(Some(PayloadItem::Chunk(src.take().freeze())))
Ok(Some(PayloadItem::Chunk(src.split().freeze())))
}
}
}
@ -581,7 +582,7 @@ impl ChunkedState {
} else {
let slice;
if *rem > len {
slice = rdr.take().freeze();
slice = rdr.split().freeze();
*rem -= len;
} else {
slice = rdr.split_to(*rem as usize).freeze();

5
actix-http/src/h1/dispatcher.rs
View File

@ -2,11 +2,10 @@ use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Instant;
use std::{fmt, io, net};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed, FramedParts};
use actix_rt::time::{delay, Delay};
use actix_rt::time::{delay_until, Delay, Instant};
use actix_service::Service;
use bitflags::bitflags;
use bytes::{BufMut, BytesMut};
@ -610,7 +609,7 @@ where
// shutdown timeout
if self.flags.contains(Flags::SHUTDOWN) {
if let Some(interval) = self.codec.config().client_disconnect_timer() {
self.ka_timer = Some(delay(interval));
self.ka_timer = Some(delay_until(interval));
} else {
self.flags.insert(Flags::READ_DISCONNECT);
if let Some(mut payload) = self.payload.take() {

109
actix-http/src/h1/encoder.rs
View File

@ -2,11 +2,13 @@
use std::fmt::Write as FmtWrite;
use std::io::Write;
use std::marker::PhantomData;
use std::ptr::copy_nonoverlapping;
use std::rc::Rc;
use std::slice::from_raw_parts_mut;
use std::str::FromStr;
use std::{cmp, fmt, io, mem};
use bytes::{BufMut, Bytes, BytesMut};
use bytes::{buf::BufMutExt, BufMut, Bytes, BytesMut};
use crate::body::BodySize;
use crate::config::ServiceConfig;
@ -144,8 +146,8 @@ pub(crate) trait MessageType: Sized {
// write headers
let mut pos = 0;
let mut has_date = false;
let mut remaining = dst.remaining_mut();
let mut buf = unsafe { &mut *(dst.bytes_mut() as *mut [u8]) };
let mut remaining = dst.capacity() - dst.len();
let mut buf = dst.bytes_mut().as_mut_ptr() as *mut u8;
for (key, value) in headers {
match *key {
CONNECTION => continue,
@ -159,61 +161,67 @@ pub(crate) trait MessageType: Sized {
match value {
map::Value::One(ref val) => {
let v = val.as_ref();
let len = k.len() + v.len() + 4;
let v_len = v.len();
let k_len = k.len();
let len = k_len + v_len + 4;
if len > remaining {
unsafe {
dst.advance_mut(pos);
}
pos = 0;
dst.reserve(len * 2);
remaining = dst.remaining_mut();
unsafe {
buf = &mut *(dst.bytes_mut() as *mut _);
}
remaining = dst.capacity() - dst.len();
buf = dst.bytes_mut().as_mut_ptr() as *mut u8;
}
// use upper Camel-Case
if camel_case {
write_camel_case(k, &mut buf[pos..pos + k.len()]);
} else {
buf[pos..pos + k.len()].copy_from_slice(k);
unsafe {
if camel_case {
write_camel_case(k, from_raw_parts_mut(buf, k_len))
} else {
write_data(k, buf, k_len)
}
buf = buf.add(k_len);
write_data(b": ", buf, 2);
buf = buf.add(2);
write_data(v, buf, v_len);
buf = buf.add(v_len);
write_data(b"\r\n", buf, 2);
buf = buf.add(2);
pos += len;
remaining -= len;
}
pos += k.len();
buf[pos..pos + 2].copy_from_slice(b": ");
pos += 2;
buf[pos..pos + v.len()].copy_from_slice(v);
pos += v.len();
buf[pos..pos + 2].copy_from_slice(b"\r\n");
pos += 2;
remaining -= len;
}
map::Value::Multi(ref vec) => {
for val in vec {
let v = val.as_ref();
let len = k.len() + v.len() + 4;
let v_len = v.len();
let k_len = k.len();
let len = k_len + v_len + 4;
if len > remaining {
unsafe {
dst.advance_mut(pos);
}
pos = 0;
dst.reserve(len * 2);
remaining = dst.remaining_mut();
unsafe {
buf = &mut *(dst.bytes_mut() as *mut _);
}
remaining = dst.capacity() - dst.len();
buf = dst.bytes_mut().as_mut_ptr() as *mut u8;
}
// use upper Camel-Case
if camel_case {
write_camel_case(k, &mut buf[pos..pos + k.len()]);
} else {
buf[pos..pos + k.len()].copy_from_slice(k);
}
pos += k.len();
buf[pos..pos + 2].copy_from_slice(b": ");
pos += 2;
buf[pos..pos + v.len()].copy_from_slice(v);
pos += v.len();
buf[pos..pos + 2].copy_from_slice(b"\r\n");
pos += 2;
unsafe {
if camel_case {
write_camel_case(k, from_raw_parts_mut(buf, k_len));
} else {
write_data(k, buf, k_len);
}
buf = buf.add(k_len);
write_data(b": ", buf, 2);
buf = buf.add(2);
write_data(v, buf, v_len);
buf = buf.add(v_len);
write_data(b"\r\n", buf, 2);
buf = buf.add(2);
};
pos += len;
remaining -= len;
}
}
@ -298,6 +306,12 @@ impl MessageType for RequestHeadType {
Version::HTTP_10 => "HTTP/1.0",
Version::HTTP_11 => "HTTP/1.1",
Version::HTTP_2 => "HTTP/2.0",
Version::HTTP_3 => "HTTP/3.0",
_ =>
return Err(io::Error::new(
io::ErrorKind::Other,
"unsupported version"
)),
}
)
.map_err(|e| io::Error::new(io::ErrorKind::Other, e))
@ -479,6 +493,10 @@ impl<'a> io::Write for Writer<'a> {
}
}
unsafe fn write_data(value: &[u8], buf: *mut u8, len: usize) {
copy_nonoverlapping(value.as_ptr(), buf, len);
}
fn write_camel_case(value: &[u8], buffer: &mut [u8]) {
let mut index = 0;
let key = value;
@ -525,7 +543,7 @@ mod tests {
assert!(enc.encode(b"", &mut bytes).ok().unwrap());
}
assert_eq!(
bytes.take().freeze(),
bytes.split().freeze(),
Bytes::from_static(b"4\r\ntest\r\n0\r\n\r\n")
);
}
@ -548,7 +566,8 @@ mod tests {
ConnectionType::Close,
&ServiceConfig::default(),
);
let data = String::from_utf8(Vec::from(bytes.take().freeze().as_ref())).unwrap();
let data =
String::from_utf8(Vec::from(bytes.split().freeze().as_ref())).unwrap();
assert!(data.contains("Content-Length: 0\r\n"));
assert!(data.contains("Connection: close\r\n"));
assert!(data.contains("Content-Type: plain/text\r\n"));
@ -561,7 +580,8 @@ mod tests {
ConnectionType::KeepAlive,
&ServiceConfig::default(),
);
let data = String::from_utf8(Vec::from(bytes.take().freeze().as_ref())).unwrap();
let data =
String::from_utf8(Vec::from(bytes.split().freeze().as_ref())).unwrap();
assert!(data.contains("Transfer-Encoding: chunked\r\n"));
assert!(data.contains("Content-Type: plain/text\r\n"));
assert!(data.contains("Date: date\r\n"));
@ -573,7 +593,8 @@ mod tests {
ConnectionType::KeepAlive,
&ServiceConfig::default(),
);
let data = String::from_utf8(Vec::from(bytes.take().freeze().as_ref())).unwrap();
let data =
String::from_utf8(Vec::from(bytes.split().freeze().as_ref())).unwrap();
assert!(data.contains("Content-Length: 100\r\n"));
assert!(data.contains("Content-Type: plain/text\r\n"));
assert!(data.contains("Date: date\r\n"));
@ -594,7 +615,8 @@ mod tests {
ConnectionType::KeepAlive,
&ServiceConfig::default(),
);
let data = String::from_utf8(Vec::from(bytes.take().freeze().as_ref())).unwrap();
let data =
String::from_utf8(Vec::from(bytes.split().freeze().as_ref())).unwrap();
assert!(data.contains("transfer-encoding: chunked\r\n"));
assert!(data.contains("content-type: xml\r\n"));
assert!(data.contains("content-type: plain/text\r\n"));
@ -627,7 +649,8 @@ mod tests {
ConnectionType::Close,
&ServiceConfig::default(),
);
let data = String::from_utf8(Vec::from(bytes.take().freeze().as_ref())).unwrap();
let data =
String::from_utf8(Vec::from(bytes.split().freeze().as_ref())).unwrap();
assert!(data.contains("content-length: 0\r\n"));
assert!(data.contains("connection: close\r\n"));
assert!(data.contains("authorization: another authorization\r\n"));

53
actix-http/src/h1/service.rs
View File

@ -97,9 +97,8 @@ where
mod openssl {
use super::*;
use actix_tls::openssl::{Acceptor, SslStream};
use actix_tls::openssl::{Acceptor, SslAcceptor, SslStream};
use actix_tls::{openssl::HandshakeError, SslError};
use open_ssl::ssl::SslAcceptor;
impl<S, B, X, U> H1Service<SslStream<TcpStream>, S, B, X, U>
where
@ -144,6 +143,56 @@ mod openssl {
}
}
#[cfg(feature = "rustls")]
mod rustls {
use super::*;
use actix_tls::rustls::{Acceptor, ServerConfig, TlsStream};
use actix_tls::SslError;
use std::{fmt, io};
impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
S::Response: Into<Response<B>>,
B: MessageBody,
X: ServiceFactory<Config = (), Request = Request, Response = Request>,
X::Error: Into<Error>,
X::InitError: fmt::Debug,
U: ServiceFactory<
Config = (),
Request = (Request, Framed<TlsStream<TcpStream>, Codec>),
Response = (),
>,
U::Error: fmt::Display,
U::InitError: fmt::Debug,
{
/// Create rustls based service
pub fn rustls(
self,
config: ServerConfig,
) -> impl ServiceFactory<
Config = (),
Request = TcpStream,
Response = (),
Error = SslError<io::Error, DispatchError>,
InitError = (),
> {
pipeline_factory(
Acceptor::new(config)
.map_err(SslError::Ssl)
.map_init_err(|_| panic!()),
)
.and_then(|io: TlsStream<TcpStream>| {
let peer_addr = io.get_ref().0.peer_addr().ok();
ok((io, peer_addr))
})
.and_then(self.map_err(SslError::Service))
}
}
}
impl<T, S, B, X, U> H1Service<T, S, B, X, U>
where
S: ServiceFactory<Config = (), Request = Request>,