mirrors/actix-web
1
0
Fork 0
mirror of https://github.com/actix/actix-web.git synced 2024-06-11 09:49:29 +00:00
Code Issues Projects Releases Wiki Activity

refactor decoder

Browse source
This commit is contained in:
Nikolay Kim 2018-11-16 19:28:07 -08:00
parent 3b7bc41418
commit 625469f0f4
13 changed files with 493 additions and 495 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/client/pipeline.rs
View file

@ -13,7 +13,7 @@ use super::{Connect, Connection};
use body::{BodyType, MessageBody, PayloadStream};
use error::PayloadError;
use h1;
use request::RequestHead;
use message::RequestHead;
pub(crate) fn send_request<T, I, B>(
head: RequestHead,

2
src/client/request.rs
View file

@ -16,7 +16,7 @@ use http::{
uri, Error as HttpError, HeaderMap, HeaderName, HeaderValue, HttpTryFrom, Method,
Uri, Version,
};
use request::RequestHead;
use message::RequestHead;
use super::response::ClientResponse;
use super::{pipeline, Connect, Connection, ConnectorError, SendRequestError};

64
src/client/response.rs
View file

@ -1,6 +1,5 @@
use std::cell::{Cell, Ref, RefCell, RefMut};
use std::cell::RefCell;
use std::fmt;
use std::rc::Rc;
use bytes::Bytes;
use futures::{Async, Poll, Stream};
@ -8,16 +7,14 @@ use http::{HeaderMap, StatusCode, Version};
use body::PayloadStream;
use error::PayloadError;
use extensions::Extensions;
use httpmessage::HttpMessage;
use request::{Message, MessageFlags, MessagePool, RequestHead};
use uri::Url;
use message::{MessageFlags, ResponseHead};
use super::pipeline::Payload;
/// Client Response
pub struct ClientResponse {
pub(crate) inner: Rc<Message>,
pub(crate) head: ResponseHead,
pub(crate) payload: RefCell<Option<PayloadStream>>,
}
@ -25,7 +22,7 @@ impl HttpMessage for ClientResponse {
type Stream = PayloadStream;
fn headers(&self) -> &HeaderMap {
&self.inner.head.headers
&self.head.headers
}
#[inline]
@ -41,75 +38,50 @@ impl HttpMessage for ClientResponse {
impl ClientResponse {
/// Create new Request instance
pub fn new() -> ClientResponse {
ClientResponse::with_pool(MessagePool::pool())
}
/// Create new Request instance with pool
pub(crate) fn with_pool(pool: &'static MessagePool) -> ClientResponse {
ClientResponse {
inner: Rc::new(Message {
pool,
head: RequestHead::default(),
status: StatusCode::OK,
url: Url::default(),
flags: Cell::new(MessageFlags::empty()),
payload: RefCell::new(None),
extensions: RefCell::new(Extensions::new()),
}),
head: ResponseHead::default(),
payload: RefCell::new(None),
}
}
#[inline]
pub(crate) fn inner(&self) -> &Message {
self.inner.as_ref()
pub(crate) fn head(&self) -> &ResponseHead {
&self.head
}
#[inline]
pub(crate) fn inner_mut(&mut self) -> &mut Message {
Rc::get_mut(&mut self.inner).expect("Multiple copies exist")
pub(crate) fn head_mut(&mut self) -> &mut ResponseHead {
&mut self.head
}
/// Read the Request Version.
#[inline]
pub fn version(&self) -> Version {
self.inner().head.version
self.head().version.clone().unwrap()
}
/// Get the status from the server.
#[inline]
pub fn status(&self) -> StatusCode {
self.inner().status
self.head().status
}
#[inline]
/// Returns Request's headers.
pub fn headers(&self) -> &HeaderMap {
&self.inner().head.headers
&self.head().headers
}
#[inline]
/// Returns mutable Request's headers.
pub fn headers_mut(&mut self) -> &mut HeaderMap {
&mut self.inner_mut().head.headers
&mut self.head_mut().headers
}
/// Checks if a connection should be kept alive.
#[inline]
pub fn keep_alive(&self) -> bool {
self.inner().flags.get().contains(MessageFlags::KEEPALIVE)
}
/// Request extensions
#[inline]
pub fn extensions(&self) -> Ref<Extensions> {
self.inner().extensions.borrow()
}
/// Mutable reference to a the request's extensions
#[inline]
pub fn extensions_mut(&self) -> RefMut<Extensions> {
self.inner().extensions.borrow_mut()
self.head().flags.contains(MessageFlags::KEEPALIVE)
}
}
@ -126,14 +98,6 @@ impl Stream for ClientResponse {
}
}
impl Drop for ClientResponse {
fn drop(&mut self) {
if Rc::strong_count(&self.inner) == 1 {
self.inner.pool.release(self.inner.clone());
}
}
}
impl fmt::Debug for ClientResponse {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
writeln!(f, "\nClientResponse {:?} {}", self.version(), self.status(),)?;

21
src/h1/client.rs
View file

@ -4,7 +4,7 @@ use std::io::{self, Write};
use bytes::{BufMut, Bytes, BytesMut};
use tokio_codec::{Decoder, Encoder};
use super::decoder::{PayloadDecoder, PayloadItem, PayloadType, ResponseDecoder};
use super::decoder::{MessageDecoder, PayloadDecoder, PayloadItem, PayloadType};
use super::encoder::{RequestEncoder, ResponseLength};
use super::{Message, MessageType};
use body::{Binary, Body, BodyType};
@ -16,7 +16,7 @@ use http::header::{
HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING, UPGRADE,
};
use http::{Method, Version};
use request::{MessagePool, RequestHead};
use message::{MessagePool, RequestHead};
bitflags! {
struct Flags: u8 {
@ -42,7 +42,7 @@ pub struct ClientPayloadCodec {
struct ClientCodecInner {
config: ServiceConfig,
decoder: ResponseDecoder,
decoder: MessageDecoder<ClientResponse>,
payload: Option<PayloadDecoder>,
version: Version,
@ -63,11 +63,6 @@ impl ClientCodec {
///
/// `keepalive_enabled` how response `connection` header get generated.
pub fn new(config: ServiceConfig) -> Self {
ClientCodec::with_pool(MessagePool::pool(), config)
}
/// Create HTTP/1 codec with request's pool
pub(crate) fn with_pool(pool: &'static MessagePool, config: ServiceConfig) -> Self {
let flags = if config.keep_alive_enabled() {
Flags::KEEPALIVE_ENABLED
} else {
@ -76,7 +71,7 @@ impl ClientCodec {
ClientCodec {
inner: ClientCodecInner {
config,
decoder: ResponseDecoder::with_pool(pool),
decoder: MessageDecoder::default(),
payload: None,
version: Version::HTTP_11,
@ -185,10 +180,10 @@ impl Decoder for ClientCodec {
debug_assert!(!self.inner.payload.is_some(), "Payload decoder is set");
if let Some((req, payload)) = self.inner.decoder.decode(src)? {
self.inner
.flags
.set(Flags::HEAD, req.inner.head.method == Method::HEAD);
self.inner.version = req.inner.head.version;
// self.inner
// .flags
// .set(Flags::HEAD, req.head.method == Method::HEAD);
// self.inner.version = req.head.version;
if self.inner.flags.contains(Flags::KEEPALIVE_ENABLED) {
self.inner.flags.set(Flags::KEEPALIVE, req.keep_alive());
}

13
src/h1/codec.rs
View file

@ -5,7 +5,7 @@ use std::io::{self, Write};
use bytes::{BufMut, Bytes, BytesMut};
use tokio_codec::{Decoder, Encoder};
use super::decoder::{PayloadDecoder, PayloadItem, PayloadType, RequestDecoder};
use super::decoder::{MessageDecoder, PayloadDecoder, PayloadItem, PayloadType};
use super::encoder::{ResponseEncoder, ResponseLength};
use super::{Message, MessageType};
use body::{Binary, Body};
@ -14,7 +14,7 @@ use error::ParseError;
use helpers;
use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING};
use http::{Method, Version};
use request::{MessagePool, Request};
use request::Request;
use response::Response;
bitflags! {
@ -32,7 +32,7 @@ const AVERAGE_HEADER_SIZE: usize = 30;
/// HTTP/1 Codec
pub struct Codec {
config: ServiceConfig,
decoder: RequestDecoder,
decoder: MessageDecoder<Request>,
payload: Option<PayloadDecoder>,
version: Version,
@ -59,11 +59,6 @@ impl Codec {
///
/// `keepalive_enabled` how response `connection` header get generated.
pub fn new(config: ServiceConfig) -> Self {
Codec::with_pool(MessagePool::pool(), config)
}
/// Create HTTP/1 codec with request's pool
pub(crate) fn with_pool(pool: &'static MessagePool, config: ServiceConfig) -> Self {
let flags = if config.keep_alive_enabled() {
Flags::KEEPALIVE_ENABLED
} else {
@ -71,7 +66,7 @@ impl Codec {
};
Codec {
config,
decoder: RequestDecoder::with_pool(pool),
decoder: MessageDecoder::default(),
payload: None,
version: Version::HTTP_11,

551
src/h1/decoder.rs
View file

@ -1,3 +1,4 @@
use std::marker::PhantomData;
use std::{io, mem};
use bytes::{Bytes, BytesMut};
@ -8,327 +9,305 @@ use tokio_codec::Decoder;
use client::ClientResponse;
use error::ParseError;
use http::header::{HeaderName, HeaderValue};
use http::{header, HttpTryFrom, Method, StatusCode, Uri, Version};
use request::{MessageFlags, MessagePool, Request};
use uri::Url;
use http::{header, HeaderMap, HttpTryFrom, Method, StatusCode, Uri, Version};
use message::MessageFlags;
use request::Request;
const MAX_BUFFER_SIZE: usize = 131_072;
const MAX_HEADERS: usize = 96;
/// Client request decoder
pub struct RequestDecoder(&'static MessagePool);
/// Server response decoder
pub struct ResponseDecoder(&'static MessagePool);
/// Incoming messagd decoder
pub(crate) struct MessageDecoder<T: MessageTypeDecoder>(PhantomData<T>);
/// Incoming request type
pub enum PayloadType {
pub(crate) enum PayloadType {
None,
Payload(PayloadDecoder),
Stream(PayloadDecoder),
}
impl RequestDecoder {
pub(crate) fn with_pool(pool: &'static MessagePool) -> RequestDecoder {
RequestDecoder(pool)
impl<T: MessageTypeDecoder> Default for MessageDecoder<T> {
fn default() -> Self {
MessageDecoder(PhantomData)
}
}
impl Default for RequestDecoder {
fn default() -> RequestDecoder {
RequestDecoder::with_pool(MessagePool::pool())
}
}
impl Decoder for RequestDecoder {
type Item = (Request, PayloadType);
impl<T: MessageTypeDecoder> Decoder for MessageDecoder<T> {
type Item = (T, PayloadType);
type Error = ParseError;
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
// Parse http message
T::decode(src)
}
}
pub(crate) enum PayloadLength {
None,
Chunked,
Upgrade,
Length(u64),
}
pub(crate) trait MessageTypeDecoder: Sized {
fn keep_alive(&mut self);
fn headers_mut(&mut self) -> &mut HeaderMap;
fn decode(src: &mut BytesMut) -> Result<Option<(Self, PayloadType)>, ParseError>;
fn process_headers(
&mut self,
slice: &Bytes,
version: Version,
raw_headers: &[HeaderIndex],
) -> Result<PayloadLength, ParseError> {
let mut ka = version != Version::HTTP_10;
let mut has_upgrade = false;
let mut chunked = false;
let mut content_length = None;
let msg = {
// Unsafe: we read only this data only after httparse parses headers into.
// performance bump for pipeline benchmarks.
let mut headers: [HeaderIndex; MAX_HEADERS] =
{
let headers = self.headers_mut();
for idx in raw_headers.iter() {
if let Ok(name) = HeaderName::from_bytes(&slice[idx.name.0..idx.name.1])
{
// Unsafe: httparse check header value for valid utf-8
let value = unsafe {
HeaderValue::from_shared_unchecked(
slice.slice(idx.value.0, idx.value.1),
)
};
match name {
header::CONTENT_LENGTH => {
if let Ok(s) = value.to_str() {
if let Ok(len) = s.parse::<u64>() {
content_length = Some(len);
} else {
debug!("illegal Content-Length: {:?}", s);
return Err(ParseError::Header);
}
} else {
debug!("illegal Content-Length: {:?}", value);
return Err(ParseError::Header);
}
}
// transfer-encoding
header::TRANSFER_ENCODING => {
if let Ok(s) = value.to_str() {
chunked = s.to_lowercase().contains("chunked");
} else {
return Err(ParseError::Header);
}
}
// connection keep-alive state
header::CONNECTION => {
ka = if let Ok(conn) = value.to_str() {
if version == Version::HTTP_10
&& conn.contains("keep-alive")
{
true
} else {
version == Version::HTTP_11 && !(conn
.contains("close")
|| conn.contains("upgrade"))
}
} else {
false
}
}
header::UPGRADE => {
has_upgrade = true;
// check content-length, some clients (dart)
// sends "content-length: 0" with websocket upgrade
if let Ok(val) = value.to_str() {
if val == "websocket" {
content_length = None;
}
}
}
_ => (),
}
headers.append(name, value);
} else {
return Err(ParseError::Header);
}
}
}
if ka {
self.keep_alive();
}
if chunked {
Ok(PayloadLength::Chunked)
} else if let Some(len) = content_length {
Ok(PayloadLength::Length(len))
} else if has_upgrade {
Ok(PayloadLength::Upgrade)
} else {
Ok(PayloadLength::None)
}
}
}
impl MessageTypeDecoder for Request {
fn keep_alive(&mut self) {
self.inner_mut().flags.set(MessageFlags::KEEPALIVE);
}
fn headers_mut(&mut self) -> &mut HeaderMap {
self.headers_mut()
}
fn decode(src: &mut BytesMut) -> Result<Option<(Self, PayloadType)>, ParseError> {
// Unsafe: we read only this data only after httparse parses headers into.
// performance bump for pipeline benchmarks.
let mut headers: [HeaderIndex; MAX_HEADERS] = unsafe { mem::uninitialized() };
let (len, method, uri, version, headers_len) = {
let mut parsed: [httparse::Header; MAX_HEADERS] =
unsafe { mem::uninitialized() };
let (len, method, path, version, headers_len) = {
let mut parsed: [httparse::Header; MAX_HEADERS] =
unsafe { mem::uninitialized() };
let mut req = httparse::Request::new(&mut parsed);
match req.parse(src)? {
httparse::Status::Complete(len) => {
let method = Method::from_bytes(req.method.unwrap().as_bytes())
.map_err(|_| ParseError::Method)?;
let path = Url::new(Uri::try_from(req.path.unwrap())?);
let version = if req.version.unwrap() == 1 {
Version::HTTP_11
} else {
Version::HTTP_10
};
HeaderIndex::record(src, req.headers, &mut headers);
(len, method, path, version, req.headers.len())
}
httparse::Status::Partial => return Ok(None),
}
};
let slice = src.split_to(len).freeze();
// convert headers
let mut msg = MessagePool::get_request(self.0);
{
let inner = msg.inner_mut();
inner
.flags
.get_mut()
.set(MessageFlags::KEEPALIVE, version != Version::HTTP_10);
for idx in headers[..headers_len].iter() {
if let Ok(name) =
HeaderName::from_bytes(&slice[idx.name.0..idx.name.1])
{
// Unsafe: httparse check header value for valid utf-8
let value = unsafe {
HeaderValue::from_shared_unchecked(
slice.slice(idx.value.0, idx.value.1),
)
};
match name {
header::CONTENT_LENGTH => {
if let Ok(s) = value.to_str() {
if let Ok(len) = s.parse::<u64>() {
content_length = Some(len);
} else {
debug!("illegal Content-Length: {:?}", len);
return Err(ParseError::Header);
}
} else {
debug!("illegal Content-Length: {:?}", len);
return Err(ParseError::Header);
}
}
// transfer-encoding
header::TRANSFER_ENCODING => {
if let Ok(s) = value.to_str() {
chunked = s.to_lowercase().contains("chunked");
} else {
return Err(ParseError::Header);
}
}
// connection keep-alive state
header::CONNECTION => {
let ka = if let Ok(conn) = value.to_str() {
if version == Version::HTTP_10
&& conn.contains("keep-alive")
{
true
} else {
version == Version::HTTP_11 && !(conn
.contains("close")
|| conn.contains("upgrade"))
}
} else {
false
};
inner.flags.get_mut().set(MessageFlags::KEEPALIVE, ka);
}
header::UPGRADE => {
has_upgrade = true;
// check content-length, some clients (dart)
// sends "content-length: 0" with websocket upgrade
if let Ok(val) = value.to_str() {
if val == "websocket" {
content_length = None;
}
}
}
_ => (),
}
inner.head.headers.append(name, value);
let mut req = httparse::Request::new(&mut parsed);
match req.parse(src)? {
httparse::Status::Complete(len) => {
let method = Method::from_bytes(req.method.unwrap().as_bytes())
.map_err(|_| ParseError::Method)?;
let uri = Uri::try_from(req.path.unwrap())?;
let version = if req.version.unwrap() == 1 {
Version::HTTP_11
} else {
return Err(ParseError::Header);
}
}
Version::HTTP_10
};
HeaderIndex::record(src, req.headers, &mut headers);
inner.url = path;
inner.head.method = method;
inner.head.version = version;
(len, method, uri, version, req.headers.len())
}
httparse::Status::Partial => return Ok(None),
}
msg
};
// convert headers
let mut msg = Request::new();
let len = msg.process_headers(
&src.split_to(len).freeze(),
version,
&headers[..headers_len],
)?;
// https://tools.ietf.org/html/rfc7230#section-3.3.3
let decoder = if chunked {
// Chunked encoding
PayloadType::Payload(PayloadDecoder::chunked())
} else if let Some(len) = content_length {
// Content-Length
PayloadType::Payload(PayloadDecoder::length(len))
} else if has_upgrade || msg.inner.head.method == Method::CONNECT {
// upgrade(websocket) or connect
PayloadType::Stream(PayloadDecoder::eof())
} else if src.len() >= MAX_BUFFER_SIZE {
error!("MAX_BUFFER_SIZE unprocessed data reached, closing");
return Err(ParseError::TooLarge);
} else {
PayloadType::None
let decoder = match len {
PayloadLength::Chunked => {
// Chunked encoding
PayloadType::Payload(PayloadDecoder::chunked())
}
PayloadLength::Length(len) => {
// Content-Length
PayloadType::Payload(PayloadDecoder::length(len))
}
PayloadLength::Upgrade => {
// upgrade(websocket) or connect
PayloadType::Stream(PayloadDecoder::eof())
}
PayloadLength::None => {
if method == Method::CONNECT {
// upgrade(websocket) or connect
PayloadType::Stream(PayloadDecoder::eof())
} else if src.len() >= MAX_BUFFER_SIZE {
error!("MAX_BUFFER_SIZE unprocessed data reached, closing");
return Err(ParseError::TooLarge);
} else {
PayloadType::None
}
}
};
{
let inner = msg.inner_mut();
inner.url.update(&uri);
inner.head.uri = uri;
inner.head.method = method;
inner.head.version = version;
}
Ok(Some((msg, decoder)))
}
}
impl ResponseDecoder {
pub(crate) fn with_pool(pool: &'static MessagePool) -> ResponseDecoder {
ResponseDecoder(pool)
impl MessageTypeDecoder for ClientResponse {
fn keep_alive(&mut self) {
self.head.flags.insert(MessageFlags::KEEPALIVE);
}
}
impl Default for ResponseDecoder {
fn default() -> ResponseDecoder {
ResponseDecoder::with_pool(MessagePool::pool())
fn headers_mut(&mut self) -> &mut HeaderMap {
self.headers_mut()
}
}
impl Decoder for ResponseDecoder {
type Item = (ClientResponse, PayloadType);
type Error = ParseError;
fn decode(src: &mut BytesMut) -> Result<Option<(Self, PayloadType)>, ParseError> {
// Unsafe: we read only this data only after httparse parses headers into.
// performance bump for pipeline benchmarks.
let mut headers: [HeaderIndex; MAX_HEADERS] = unsafe { mem::uninitialized() };
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
// Parse http message
let mut chunked = false;
let mut content_length = None;
let msg = {
// Unsafe: we read only this data only after httparse parses headers into.
// performance bump for pipeline benchmarks.
let mut headers: [HeaderIndex; MAX_HEADERS] =
let (len, version, status, headers_len) = {
let mut parsed: [httparse::Header; MAX_HEADERS] =
unsafe { mem::uninitialized() };
let (len, version, status, headers_len) = {
let mut parsed: [httparse::Header; MAX_HEADERS] =
unsafe { mem::uninitialized() };
let mut res = httparse::Response::new(&mut parsed);
match res.parse(src)? {
httparse::Status::Complete(len) => {
let version = if res.version.unwrap() == 1 {
Version::HTTP_11
} else {
Version::HTTP_10
};
let status = StatusCode::from_u16(res.code.unwrap())
.map_err(|_| ParseError::Status)?;
HeaderIndex::record(src, res.headers, &mut headers);
(len, version, status, res.headers.len())
}
httparse::Status::Partial => return Ok(None),
}
};
let slice = src.split_to(len).freeze();
// convert headers
let mut msg = MessagePool::get_response(self.0);
{
let inner = msg.inner_mut();
inner
.flags
.get_mut()
.set(MessageFlags::KEEPALIVE, version != Version::HTTP_10);
for idx in headers[..headers_len].iter() {
if let Ok(name) =
HeaderName::from_bytes(&slice[idx.name.0..idx.name.1])
{
// Unsafe: httparse check header value for valid utf-8
let value = unsafe {
HeaderValue::from_shared_unchecked(
slice.slice(idx.value.0, idx.value.1),
)
};
match name {
header::CONTENT_LENGTH => {
if let Ok(s) = value.to_str() {
if let Ok(len) = s.parse::<u64>() {
content_length = Some(len);
} else {
debug!("illegal Content-Length: {:?}", len);
return Err(ParseError::Header);
}
} else {
debug!("illegal Content-Length: {:?}", len);
return Err(ParseError::Header);
}
}
// transfer-encoding
header::TRANSFER_ENCODING => {
if let Ok(s) = value.to_str() {
chunked = s.to_lowercase().contains("chunked");
} else {
return Err(ParseError::Header);
}
}
// connection keep-alive state
header::CONNECTION => {
let ka = if let Ok(conn) = value.to_str() {
if version == Version::HTTP_10
&& conn.contains("keep-alive")
{
true
} else {
version == Version::HTTP_11 && !(conn
.contains("close")
|| conn.contains("upgrade"))
}
} else {
false
};
inner.flags.get_mut().set(MessageFlags::KEEPALIVE, ka);
}
_ => (),
}
inner.head.headers.append(name, value);
let mut res = httparse::Response::new(&mut parsed);
match res.parse(src)? {
httparse::Status::Complete(len) => {
let version = if res.version.unwrap() == 1 {
Version::HTTP_11
} else {
return Err(ParseError::Header);
}
}
Version::HTTP_10
};
let status = StatusCode::from_u16(res.code.unwrap())
.map_err(|_| ParseError::Status)?;
HeaderIndex::record(src, res.headers, &mut headers);
inner.status = status;
inner.head.version = version;
(len, version, status, res.headers.len())
}
httparse::Status::Partial => return Ok(None),
}
msg
};
let mut msg = ClientResponse::new();
// convert headers
let len = msg.process_headers(
&src.split_to(len).freeze(),
version,
&headers[..headers_len],
)?;
// https://tools.ietf.org/html/rfc7230#section-3.3.3
let decoder = if chunked {
// Chunked encoding
PayloadType::Payload(PayloadDecoder::chunked())
} else if let Some(len) = content_length {
// Content-Length
PayloadType::Payload(PayloadDecoder::length(len))
} else if msg.inner.status == StatusCode::SWITCHING_PROTOCOLS
|| msg.inner.head.method == Method::CONNECT
{
// switching protocol or connect
PayloadType::Stream(PayloadDecoder::eof())
} else if src.len() >= MAX_BUFFER_SIZE {
error!("MAX_BUFFER_SIZE unprocessed data reached, closing");
return Err(ParseError::TooLarge);
} else {
PayloadType::None
let decoder = match len {
PayloadLength::Chunked => {
// Chunked encoding
PayloadType::Payload(PayloadDecoder::chunked())
}
PayloadLength::Length(len) => {
// Content-Length
PayloadType::Payload(PayloadDecoder::length(len))
}
_ => {
if status == StatusCode::SWITCHING_PROTOCOLS {
// switching protocol or connect
PayloadType::Stream(PayloadDecoder::eof())
} else if src.len() >= MAX_BUFFER_SIZE {
error!("MAX_BUFFER_SIZE unprocessed data reached, closing");
return Err(ParseError::TooLarge);
} else {
PayloadType::None
}
}
};
msg.head.status = status;
msg.head.version = Some(version);
Ok(Some((msg, decoder)))
}
}
@ -690,7 +669,7 @@ mod tests {
macro_rules! parse_ready {
($e:expr) => {{
match RequestDecoder::default().decode($e) {
match MessageDecoder::<Request>::default().decode($e) {
Ok(Some((msg, _))) => msg,
Ok(_) => unreachable!("Eof during parsing http request"),
Err(err) => unreachable!("Error during parsing http request: {:?}", err),
@ -700,7 +679,7 @@ mod tests {
macro_rules! expect_parse_err {
($e:expr) => {{
match RequestDecoder::default().decode($e) {
match MessageDecoder::<Request>::default().decode($e) {
Err(err) => match err {
ParseError::Io(_) => unreachable!("Parse error expected"),
_ => (),
@ -779,7 +758,7 @@ mod tests {
fn test_parse() {
let mut buf = BytesMut::from("GET /test HTTP/1.1\r\n\r\n");
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
match reader.decode(&mut buf) {
Ok(Some((req, _))) => {
assert_eq!(req.version(), Version::HTTP_11);
@ -794,7 +773,7 @@ mod tests {
fn test_parse_partial() {
let mut buf = BytesMut::from("PUT /test HTTP/1");
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
assert!(reader.decode(&mut buf).unwrap().is_none());
buf.extend(b".1\r\n\r\n");
@ -808,7 +787,7 @@ mod tests {
fn test_parse_post() {
let mut buf = BytesMut::from("POST /test2 HTTP/1.0\r\n\r\n");
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, _) = reader.decode(&mut buf).unwrap().unwrap();
assert_eq!(req.version(), Version::HTTP_10);
assert_eq!(*req.method(), Method::POST);
@ -820,7 +799,7 @@ mod tests {
let mut buf =
BytesMut::from("GET /test HTTP/1.1\r\nContent-Length: 4\r\n\r\nbody");
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, pl) = reader.decode(&mut buf).unwrap().unwrap();
let mut pl = pl.unwrap();
assert_eq!(req.version(), Version::HTTP_11);
@ -837,7 +816,7 @@ mod tests {
let mut buf =
BytesMut::from("\r\nGET /test HTTP/1.1\r\nContent-Length: 4\r\n\r\nbody");
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, pl) = reader.decode(&mut buf).unwrap().unwrap();
let mut pl = pl.unwrap();
assert_eq!(req.version(), Version::HTTP_11);
@ -852,7 +831,7 @@ mod tests {
#[test]
fn test_parse_partial_eof() {
let mut buf = BytesMut::from("GET /test HTTP/1.1\r\n");
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
assert!(reader.decode(&mut buf).unwrap().is_none());
buf.extend(b"\r\n");
@ -866,7 +845,7 @@ mod tests {
fn test_headers_split_field() {
let mut buf = BytesMut::from("GET /test HTTP/1.1\r\n");
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
assert!{ reader.decode(&mut buf).unwrap().is_none() }
buf.extend(b"t");
@ -890,7 +869,7 @@ mod tests {
Set-Cookie: c1=cookie1\r\n\
Set-Cookie: c2=cookie2\r\n\r\n",
);
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, _) = reader.decode(&mut buf).unwrap().unwrap();
let val: Vec<_> = req
@ -1090,7 +1069,7 @@ mod tests {
upgrade: websocket\r\n\r\n\
some raw data",
);
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, pl) = reader.decode(&mut buf).unwrap().unwrap();
assert!(!req.keep_alive());
assert!(req.upgrade());
@ -1139,7 +1118,7 @@ mod tests {
"GET /test HTTP/1.1\r\n\
transfer-encoding: chunked\r\n\r\n",
);
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, pl) = reader.decode(&mut buf).unwrap().unwrap();
let mut pl = pl.unwrap();
assert!(req.chunked().unwrap());
@ -1162,7 +1141,7 @@ mod tests {
"GET /test HTTP/1.1\r\n\
transfer-encoding: chunked\r\n\r\n",
);
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, pl) = reader.decode(&mut buf).unwrap().unwrap();
let mut pl = pl.unwrap();
assert!(req.chunked().unwrap());
@ -1193,7 +1172,7 @@ mod tests {
transfer-encoding: chunked\r\n\r\n",
);
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (req, pl) = reader.decode(&mut buf).unwrap().unwrap();
let mut pl = pl.unwrap();
assert!(req.chunked().unwrap());
@ -1238,7 +1217,7 @@ mod tests {
transfer-encoding: chunked\r\n\r\n"[..],
);
let mut reader = RequestDecoder::default();
let mut reader = MessageDecoder::<Request>::default();
let (msg, pl) = reader.decode(&mut buf).unwrap().unwrap();
let mut pl = pl.unwrap();
assert!(msg.chunked().unwrap());

3
src/h1/encoder.rs
View file

@ -11,7 +11,8 @@ use http::{StatusCode, Version};
use body::{Binary, Body};
use header::ContentEncoding;
use http::Method;
use request::{Request, RequestHead};
use message::RequestHead;
use request::Request;
use response::Response;
#[derive(Debug)]

1
src/h1/mod.rs
View file

@ -11,7 +11,6 @@ mod service;
pub use self::client::{ClientCodec, ClientPayloadCodec};
pub use self::codec::Codec;
pub use self::decoder::{PayloadDecoder, RequestDecoder};
pub use self::dispatcher::Dispatcher;
pub use self::service::{H1Service, H1ServiceHandler, OneRequest};

1
src/lib.rs
View file

@ -117,6 +117,7 @@ mod header;
mod httpcodes;
mod httpmessage;
mod json;
mod message;
mod payload;
mod request;
mod response;

163
src/message.rs Normal file
View file

@ -0,0 +1,163 @@
use std::cell::{Cell, RefCell};
use std::collections::VecDeque;
use std::rc::Rc;
use http::{HeaderMap, Method, StatusCode, Uri, Version};
use extensions::Extensions;
use payload::Payload;
use uri::Url;
#[doc(hidden)]
pub trait Head: Default + 'static {
fn clear(&mut self);
fn pool() -> &'static MessagePool<Self>;
}
bitflags! {
pub(crate) struct MessageFlags: u8 {
const KEEPALIVE = 0b0000_0001;
}
}
pub struct RequestHead {
pub uri: Uri,
pub method: Method,
pub version: Version,
pub headers: HeaderMap,
pub(crate) flags: MessageFlags,
}
impl Default for RequestHead {
fn default() -> RequestHead {
RequestHead {
uri: Uri::default(),
method: Method::default(),
version: Version::HTTP_11,
headers: HeaderMap::with_capacity(16),
flags: MessageFlags::empty(),
}
}
}
impl Head for RequestHead {
fn clear(&mut self) {
self.headers.clear();
self.flags = MessageFlags::empty();
}
fn pool() -> &'static MessagePool<Self> {
REQUEST_POOL.with(|p| *p)
}
}
pub struct ResponseHead {
pub version: Option<Version>,
pub status: StatusCode,
pub headers: HeaderMap,
pub reason: Option<&'static str>,
pub(crate) flags: MessageFlags,
}
impl Default for ResponseHead {
fn default() -> ResponseHead {
ResponseHead {
version: None,
status: StatusCode::OK,
headers: HeaderMap::with_capacity(16),
reason: None,
flags: MessageFlags::empty(),
}
}
}
impl Head for ResponseHead {
fn clear(&mut self) {
self.headers.clear();
self.flags = MessageFlags::empty();
}
fn pool() -> &'static MessagePool<Self> {
RESPONSE_POOL.with(|p| *p)
}
}
pub struct Message<T: Head> {
pub head: T,
pub url: Url,
pub status: StatusCode,
pub extensions: RefCell<Extensions>,
pub payload: RefCell<Option<Payload>>,
pub(crate) pool: &'static MessagePool<T>,
pub(crate) flags: Cell<MessageFlags>,
}
impl<T: Head> Message<T> {
#[inline]
/// Reset request instance
pub fn reset(&mut self) {
self.head.clear();
self.extensions.borrow_mut().clear();
self.flags.set(MessageFlags::empty());
*self.payload.borrow_mut() = None;
}
}
impl<T: Head> Default for Message<T> {
fn default() -> Self {
Message {
pool: T::pool(),
url: Url::default(),
head: T::default(),
status: StatusCode::OK,
flags: Cell::new(MessageFlags::empty()),
payload: RefCell::new(None),
extensions: RefCell::new(Extensions::new()),
}
}
}
#[doc(hidden)]
/// Request's objects pool
pub struct MessagePool<T: Head>(RefCell<VecDeque<Rc<Message<T>>>>);
thread_local!(static REQUEST_POOL: &'static MessagePool<RequestHead> = MessagePool::<RequestHead>::create());
thread_local!(static RESPONSE_POOL: &'static MessagePool<ResponseHead> = MessagePool::<ResponseHead>::create());
impl MessagePool<RequestHead> {
/// Get default request's pool
pub fn pool() -> &'static MessagePool<RequestHead> {
REQUEST_POOL.with(|p| *p)
}
/// Get Request object
#[inline]
pub fn get_message() -> Rc<Message<RequestHead>> {
REQUEST_POOL.with(|pool| {
if let Some(mut msg) = pool.0.borrow_mut().pop_front() {
if let Some(r) = Rc::get_mut(&mut msg) {
r.reset();
}
return msg;
}
Rc::new(Message::default())
})
}
}
impl<T: Head> MessagePool<T> {
fn create() -> &'static MessagePool<T> {
let pool = MessagePool(RefCell::new(VecDeque::with_capacity(128)));
Box::leak(Box::new(pool))
}
#[inline]
/// Release request instance
pub(crate) fn release(&self, msg: Rc<Message<T>>) {
let v = &mut self.0.borrow_mut();
if v.len() < 128 {
v.push_front(msg);
}
}
}

148
src/request.rs
View file

@ -1,65 +1,18 @@
use std::cell::{Cell, Ref, RefCell, RefMut};
use std::collections::VecDeque;
use std::cell::{Ref, RefMut};
use std::fmt;
use std::rc::Rc;
use http::{header, HeaderMap, Method, StatusCode, Uri, Version};
use http::{header, HeaderMap, Method, Uri, Version};
use client::ClientResponse;
use extensions::Extensions;
use httpmessage::HttpMessage;
use payload::Payload;
use uri::Url;
bitflags! {
pub(crate) struct MessageFlags: u8 {
const KEEPALIVE = 0b0000_0001;
const CONN_INFO = 0b0000_0010;
}
}
use message::{Message, MessageFlags, MessagePool, RequestHead};
/// Request
pub struct Request {
pub(crate) inner: Rc<Message>,
}
pub struct RequestHead {
pub uri: Uri,
pub method: Method,
pub version: Version,
pub headers: HeaderMap,
}
impl Default for RequestHead {
fn default() -> RequestHead {
RequestHead {
uri: Uri::default(),
method: Method::default(),
version: Version::HTTP_11,
headers: HeaderMap::with_capacity(16),
}
}
}
pub struct Message {
pub head: RequestHead,
pub url: Url,
pub status: StatusCode,
pub extensions: RefCell<Extensions>,
pub payload: RefCell<Option<Payload>>,
pub(crate) pool: &'static MessagePool,
pub(crate) flags: Cell<MessageFlags>,
}
impl Message {
#[inline]
/// Reset request instance
pub fn reset(&mut self) {
self.head.clear();
self.extensions.borrow_mut().clear();
self.flags.set(MessageFlags::empty());
*self.payload.borrow_mut() = None;
}
pub(crate) inner: Rc<Message<RequestHead>>,
}
impl HttpMessage for Request {
@ -82,33 +35,35 @@ impl HttpMessage for Request {
impl Request {
/// Create new Request instance
pub fn new() -> Request {
Request::with_pool(MessagePool::pool())
}
/// Create new Request instance with pool
pub(crate) fn with_pool(pool: &'static MessagePool) -> Request {
Request {
inner: Rc::new(Message {
pool,
url: Url::default(),
head: RequestHead::default(),
status: StatusCode::OK,
flags: Cell::new(MessageFlags::empty()),
payload: RefCell::new(None),
extensions: RefCell::new(Extensions::new()),
}),
inner: MessagePool::get_message(),
}
}
// /// Create new Request instance with pool
// pub(crate) fn with_pool(pool: &'static MessagePool) -> Request {
// Request {
// inner: Rc::new(Message {
// pool,
// url: Url::default(),
// head: RequestHead::default(),
// status: StatusCode::OK,
// flags: Cell::new(MessageFlags::empty()),
// payload: RefCell::new(None),
// extensions: RefCell::new(Extensions::new()),
// }),
// }
// }
#[inline]
#[doc(hidden)]
pub fn inner(&self) -> &Message {
pub fn inner(&self) -> &Message<RequestHead> {
self.inner.as_ref()
}
#[inline]
#[doc(hidden)]
pub fn inner_mut(&mut self) -> &mut Message {
pub fn inner_mut(&mut self) -> &mut Message<RequestHead> {
Rc::get_mut(&mut self.inner).expect("Multiple copies exist")
}
@ -139,7 +94,11 @@ impl Request {
/// The target path of this Request.
#[inline]
pub fn path(&self) -> &str {
self.inner().url.path()
if let Some(path) = self.inner().url.path() {
path
} else {
self.inner().head.uri.path()
}
}
#[inline]
@ -219,56 +178,3 @@ impl fmt::Debug for Request {
Ok(())
}
}
/// Request's objects pool
pub(crate) struct MessagePool(RefCell<VecDeque<Rc<Message>>>);
thread_local!(static POOL: &'static MessagePool = MessagePool::create());
impl MessagePool {
fn create() -> &'static MessagePool {
let pool = MessagePool(RefCell::new(VecDeque::with_capacity(128)));
Box::leak(Box::new(pool))
}
/// Get default request's pool
pub fn pool() -> &'static MessagePool {
POOL.with(|p| *p)
}
/// Get Request object
#[inline]
pub fn get_request(pool: &'static MessagePool) -> Request {
if let Some(mut msg) = pool.0.borrow_mut().pop_front() {
if let Some(r) = Rc::get_mut(&mut msg) {
r.reset();
}
return Request { inner: msg };
}
Request::with_pool(pool)
}
/// Get Client Response object
#[inline]
pub fn get_response(pool: &'static MessagePool) -> ClientResponse {
if let Some(mut msg) = pool.0.borrow_mut().pop_front() {
if let Some(r) = Rc::get_mut(&mut msg) {
r.reset();
}
return ClientResponse {
inner: msg,
payload: RefCell::new(None),
};
}
ClientResponse::with_pool(pool)
}
#[inline]
/// Release request instance
pub(crate) fn release(&self, msg: Rc<Message>) {
let v = &mut self.0.borrow_mut();
if v.len() < 128 {
v.push_front(msg);
}
}
}

2
src/test.rs
View file

@ -392,7 +392,7 @@ impl TestRequest {
{
let inner = req.inner_mut();
inner.head.method = method;
inner.url = InnerUrl::new(uri);
inner.url = InnerUrl::new(&uri);
inner.head.version = version;
inner.head.headers = headers;
*inner.payload.borrow_mut() = payload;

17
src/uri.rs
View file

@ -37,27 +37,22 @@ lazy_static! {
#[derive(Default, Clone, Debug)]
pub struct Url {
uri: Uri,
path: Option<Rc<String>>,
}
impl Url {
pub fn new(uri: Uri) -> Url {
pub fn new(uri: &Uri) -> Url {
let path = DEFAULT_QUOTER.requote(uri.path().as_bytes());
Url { uri, path }
Url { path }
}
pub fn uri(&self) -> &Uri {
&self.uri
pub(crate) fn update(&mut self, uri: &Uri) {
self.path = DEFAULT_QUOTER.requote(uri.path().as_bytes());
}
pub fn path(&self) -> &str {
if let Some(ref s) = self.path {
s
} else {
self.uri.path()
}
pub fn path(&self) -> Option<&str> {
self.path.as_ref().map(|s| s.as_str())
}
}