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

use objects pool for HttpRequest; optimize nested services call

Browse source
This commit is contained in:
Nikolay Kim 2019-04-07 23:06:21 -07:00
parent 75b213a6f0
commit aa78565453
20 changed files with 343 additions and 295 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
Cargo.toml
View file

@ -68,9 +68,9 @@ rust-tls = ["rustls", "actix-server/rust-tls"]
[dependencies]
actix-codec = "0.1.1"
actix-service = "0.3.4"
actix-service = "0.3.6"
actix-utils = "0.3.4"
actix-router = "0.1.1"
actix-router = "0.1.2"
actix-rt = "0.2.2"
actix-web-codegen = "0.1.0-alpha.1"
actix-http = { version = "0.1.0-alpha.3", features=["fail"] }
@ -124,4 +124,4 @@ actix-web-codegen = { path = "actix-web-codegen" }
actix-web-actors = { path = "actix-web-actors" }
actix-session = { path = "actix-session" }
actix-files = { path = "actix-files" }
awc = { path = "awc" }
awc = { path = "awc" }

3
actix-files/src/lib.rs
View file

@ -423,7 +423,7 @@ impl<P> FilesService<P> {
> {
log::debug!("Files: Failed to handle {}: {}", req.path(), e);
if let Some(ref mut default) = self.default {
Either::B(default.call(ServiceRequest::from_parts(req, payload)))
default.call(ServiceRequest::from_parts(req, payload))
} else {
Either::A(ok(ServiceResponse::from_err(e, req.clone())))
}
@ -955,6 +955,7 @@ mod tests {
.method(Method::POST)
.to_http_request();
let resp = file.respond_to(&req).unwrap();
println!("RES: {:?}", resp);
assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
let file = NamedFile::open("Cargo.toml").unwrap();

3
actix-http/src/client/h2proto.rs
View file

@ -104,9 +104,8 @@ where
let (parts, body) = resp.into_parts();
let payload = if head_req { Payload::None } else { body.into() };
let mut head = ResponseHead::default();
let mut head = ResponseHead::new(parts.status);
head.version = parts.version;
head.status = parts.status;
head.headers = parts.headers.into();
Ok((head, payload))

1
actix-http/src/client/pool.rs
View file

@ -21,6 +21,7 @@ use tokio_timer::{sleep, Delay};
use super::connection::{ConnectionType, IoConnection};
use super::error::ConnectError;
#[allow(dead_code)]
#[derive(Clone, Copy, PartialEq)]
pub enum Protocol {
Http1,

133
actix-http/src/h1/decoder.rs
View file

@ -73,78 +73,75 @@ pub(crate) trait MessageType: Sized {
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>() {
if len != 0 {
content_length = Some(len);
}
} else {
debug!("illegal Content-Length: {:?}", s);
return Err(ParseError::Header);
let name =
HeaderName::from_bytes(&slice[idx.name.0..idx.name.1]).unwrap();
// 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>() {
if len != 0 {
content_length = Some(len);
}
} else {
debug!("illegal Content-Length: {:?}", value);
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().map(|s| s.trim()) {
chunked = s.eq_ignore_ascii_case("chunked");
} else {
return Err(ParseError::Header);
}
}
// transfer-encoding
header::TRANSFER_ENCODING => {
if let Ok(s) = value.to_str().map(|s| s.trim()) {
chunked = s.eq_ignore_ascii_case("chunked");
} else {
return Err(ParseError::Header);
}
// connection keep-alive state
header::CONNECTION => {
ka = if let Ok(conn) = value.to_str().map(|conn| conn.trim())
{
if conn.eq_ignore_ascii_case("keep-alive") {
Some(ConnectionType::KeepAlive)
} else if conn.eq_ignore_ascii_case("close") {
Some(ConnectionType::Close)
} else if conn.eq_ignore_ascii_case("upgrade") {
Some(ConnectionType::Upgrade)
} else {
None
}
}
// connection keep-alive state
header::CONNECTION => {
ka = if let Ok(conn) = value.to_str().map(|conn| conn.trim()) {
if conn.eq_ignore_ascii_case("keep-alive") {
Some(ConnectionType::KeepAlive)
} else if conn.eq_ignore_ascii_case("close") {
Some(ConnectionType::Close)
} else if conn.eq_ignore_ascii_case("upgrade") {
Some(ConnectionType::Upgrade)
} else {
None
};
}
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().map(|val| val.trim()) {
if val.eq_ignore_ascii_case("websocket") {
content_length = None;
}
}
}
header::EXPECT => {
let bytes = value.as_bytes();
if bytes.len() >= 4 && &bytes[0..4] == b"100-" {
expect = true;
}
}
_ => (),
} else {
None
};
}
headers.append(name, value);
} else {
return Err(ParseError::Header);
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().map(|val| val.trim()) {
if val.eq_ignore_ascii_case("websocket") {
content_length = None;
}
}
}
header::EXPECT => {
let bytes = value.as_bytes();
if bytes.len() >= 4 && &bytes[0..4] == b"100-" {
expect = true;
}
}
_ => (),
}
headers.append(name, value);
}
}
self.set_connection_type(ka);
@ -217,10 +214,10 @@ impl MessageType for Request {
let mut msg = Request::new();
// convert headers
let len = msg.set_headers(&src.split_to(len).freeze(), &headers[..h_len])?;
let length = msg.set_headers(&src.split_to(len).freeze(), &headers[..h_len])?;
// payload decoder
let decoder = match len {
let decoder = match length {
PayloadLength::Payload(pl) => pl,
PayloadLength::Upgrade => {
// upgrade(websocket)
@ -287,13 +284,14 @@ impl MessageType for ResponseHead {
}
};
let mut msg = ResponseHead::default();
let mut msg = ResponseHead::new(status);
msg.version = ver;
// convert headers
let len = msg.set_headers(&src.split_to(len).freeze(), &headers[..h_len])?;
let length = msg.set_headers(&src.split_to(len).freeze(), &headers[..h_len])?;
// message payload
let decoder = if let PayloadLength::Payload(pl) = len {
let decoder = if let PayloadLength::Payload(pl) = length {
pl
} else if status == StatusCode::SWITCHING_PROTOCOLS {
// switching protocol or connect
@ -305,9 +303,6 @@ impl MessageType for ResponseHead {
PayloadType::None
};
msg.status = status;
msg.version = ver;
Ok(Some((msg, decoder)))
}
}

2
actix-http/src/h1/dispatcher.rs
View file

@ -218,7 +218,7 @@ where
{
fn can_read(&self) -> bool {
if self.flags.contains(Flags::READ_DISCONNECT) {
return false;
false
} else if let Some(ref info) = self.payload {
info.need_read() == PayloadStatus::Read
} else {

2
actix-http/src/httpcodes.rs
View file

@ -8,7 +8,7 @@ macro_rules! STATIC_RESP {
($name:ident, $status:expr) => {
#[allow(non_snake_case, missing_docs)]
pub fn $name() -> ResponseBuilder {
Response::build($status)
ResponseBuilder::new($status)
}
};
}

6
actix-http/src/lib.rs
View file

@ -1,5 +1,9 @@
//! Basic http primitives for actix-net framework.
#![allow(clippy::type_complexity, clippy::new_without_default)]
#![allow(
clippy::type_complexity,
clippy::new_without_default,
clippy::borrow_interior_mutable_const
)]
#[macro_use]
extern crate log;

36
actix-http/src/message.rs
View file

@ -1,5 +1,4 @@
use std::cell::{Ref, RefCell, RefMut};
use std::collections::VecDeque;
use std::rc::Rc;
use bitflags::bitflags;
@ -171,20 +170,20 @@ pub struct ResponseHead {
flags: Flags,
}
impl Default for ResponseHead {
fn default() -> ResponseHead {
impl ResponseHead {
/// Create new instance of `ResponseHead` type
#[inline]
pub fn new(status: StatusCode) -> ResponseHead {
ResponseHead {
status,
version: Version::default(),
status: StatusCode::OK,
headers: HeaderMap::with_capacity(12),
reason: None,
flags: Flags::empty(),
extensions: RefCell::new(Extensions::new()),
}
}
}
impl ResponseHead {
/// Message extensions
#[inline]
pub fn extensions(&self) -> Ref<Extensions> {
@ -335,8 +334,8 @@ pub(crate) struct BoxedResponseHead {
impl BoxedResponseHead {
/// Get new message from the pool of objects
pub fn new() -> Self {
RESPONSE_POOL.with(|p| p.get_message())
pub fn new(status: StatusCode) -> Self {
RESPONSE_POOL.with(|p| p.get_message(status))
}
}
@ -362,25 +361,25 @@ impl Drop for BoxedResponseHead {
#[doc(hidden)]
/// Request's objects pool
pub struct MessagePool<T: Head>(RefCell<VecDeque<Rc<T>>>);
pub struct MessagePool<T: Head>(RefCell<Vec<Rc<T>>>);
#[doc(hidden)]
/// Request's objects pool
pub struct BoxedResponsePool(RefCell<VecDeque<Box<ResponseHead>>>);
pub struct BoxedResponsePool(RefCell<Vec<Box<ResponseHead>>>);
thread_local!(static REQUEST_POOL: &'static MessagePool<RequestHead> = MessagePool::<RequestHead>::create());
thread_local!(static RESPONSE_POOL: &'static BoxedResponsePool = BoxedResponsePool::create());
impl<T: Head> MessagePool<T> {
fn create() -> &'static MessagePool<T> {
let pool = MessagePool(RefCell::new(VecDeque::with_capacity(128)));
let pool = MessagePool(RefCell::new(Vec::with_capacity(128)));
Box::leak(Box::new(pool))
}
/// Get message from the pool
#[inline]
fn get_message(&'static self) -> Message<T> {
if let Some(mut msg) = self.0.borrow_mut().pop_front() {
if let Some(mut msg) = self.0.borrow_mut().pop() {
if let Some(r) = Rc::get_mut(&mut msg) {
r.clear();
}
@ -397,28 +396,29 @@ impl<T: Head> MessagePool<T> {
fn release(&self, msg: Rc<T>) {
let v = &mut self.0.borrow_mut();
if v.len() < 128 {
v.push_front(msg);
v.push(msg);
}
}
}
impl BoxedResponsePool {
fn create() -> &'static BoxedResponsePool {
let pool = BoxedResponsePool(RefCell::new(VecDeque::with_capacity(128)));
let pool = BoxedResponsePool(RefCell::new(Vec::with_capacity(128)));
Box::leak(Box::new(pool))
}
/// Get message from the pool
#[inline]
fn get_message(&'static self) -> BoxedResponseHead {
if let Some(mut head) = self.0.borrow_mut().pop_front() {
fn get_message(&'static self, status: StatusCode) -> BoxedResponseHead {
if let Some(mut head) = self.0.borrow_mut().pop() {
head.reason = None;
head.status = status;
head.headers.clear();
head.flags = Flags::empty();
BoxedResponseHead { head: Some(head) }
} else {
BoxedResponseHead {
head: Some(Box::alloc().init(ResponseHead::default())),
head: Some(Box::alloc().init(ResponseHead::new(status))),
}
}
}
@ -428,7 +428,7 @@ impl BoxedResponsePool {
fn release(&self, msg: Box<ResponseHead>) {
let v = &mut self.0.borrow_mut();
if v.len() < 128 {
v.push_front(msg);
v.push(msg);
}
}
}

60
actix-http/src/response.rs
View file

@ -41,11 +41,8 @@ impl Response<Body> {
/// Constructs a response
#[inline]
pub fn new(status: StatusCode) -> Response {
let mut head = BoxedResponseHead::new();
head.status = status;
Response {
head,
head: BoxedResponseHead::new(status),
body: ResponseBody::Body(Body::Empty),
error: None,
}
@ -78,6 +75,16 @@ impl Response<Body> {
}
impl<B> Response<B> {
/// Constructs a response with body
#[inline]
pub fn with_body(status: StatusCode, body: B) -> Response<B> {
Response {
head: BoxedResponseHead::new(status),
body: ResponseBody::Body(body),
error: None,
}
}
#[inline]
/// Http message part of the response
pub fn head(&self) -> &ResponseHead {
@ -90,18 +97,6 @@ impl<B> Response<B> {
&mut *self.head
}
/// Constructs a response with body
#[inline]
pub fn with_body(status: StatusCode, body: B) -> Response<B> {
let mut head = BoxedResponseHead::new();
head.status = status;
Response {
head,
body: ResponseBody::Body(body),
error: None,
}
}
/// The source `error` for this response
#[inline]
pub fn error(&self) -> Option<&Error> {
@ -325,13 +320,11 @@ pub struct ResponseBuilder {
}
impl ResponseBuilder {
#[inline]
/// Create response builder
pub fn new(status: StatusCode) -> Self {
let mut head = BoxedResponseHead::new();
head.status = status;
ResponseBuilder {
head: Some(head),
head: Some(BoxedResponseHead::new(status)),
err: None,
cookies: None,
}
@ -555,15 +548,13 @@ impl ResponseBuilder {
/// }
/// ```
pub fn del_cookie<'a>(&mut self, cookie: &Cookie<'a>) -> &mut Self {
{
if self.cookies.is_none() {
self.cookies = Some(CookieJar::new())
}
let jar = self.cookies.as_mut().unwrap();
let cookie = cookie.clone().into_owned();
jar.add_original(cookie.clone());
jar.remove(cookie);
if self.cookies.is_none() {
self.cookies = Some(CookieJar::new())
}
let jar = self.cookies.as_mut().unwrap();
let cookie = cookie.clone().into_owned();
jar.add_original(cookie.clone());
jar.remove(cookie);
self
}
@ -605,6 +596,7 @@ impl ResponseBuilder {
head.extensions.borrow_mut()
}
#[inline]
/// Set a body and generate `Response`.
///
/// `ResponseBuilder` can not be used after this call.
@ -625,9 +617,7 @@ impl ResponseBuilder {
if let Some(ref jar) = self.cookies {
for cookie in jar.delta() {
match HeaderValue::from_str(&cookie.to_string()) {
Ok(val) => {
let _ = response.headers.append(header::SET_COOKIE, val);
}
Ok(val) => response.headers.append(header::SET_COOKIE, val),
Err(e) => return Response::from(Error::from(e)).into_body(),
};
}
@ -652,6 +642,7 @@ impl ResponseBuilder {
self.body(Body::from_message(BodyStream::new(stream)))
}
#[inline]
/// Set a json body and generate `Response`
///
/// `ResponseBuilder` can not be used after this call.
@ -751,11 +742,12 @@ impl<'a> From<&'a ResponseHead> for ResponseBuilder {
}
}
let mut msg = BoxedResponseHead::new();
let mut msg = BoxedResponseHead::new(head.status);
msg.version = head.version;
msg.status = head.status;
msg.reason = head.reason;
// msg.headers = head.headers.clone();
for (k, v) in &head.headers {
msg.headers.append(k.clone(), v.clone());
}
msg.no_chunking(!head.chunked());
ResponseBuilder {

2
actix-http/src/test.rs
View file

@ -178,6 +178,6 @@ impl TestRequest {
}
#[inline]
fn parts<'a>(parts: &'a mut Option<Inner>) -> &'a mut Inner {
fn parts(parts: &mut Option<Inner>) -> &mut Inner {
parts.as_mut().expect("cannot reuse test request builder")
}

4
awc/src/test.rs
View file

@ -3,7 +3,7 @@ use std::fmt::Write as FmtWrite;
use actix_http::cookie::{Cookie, CookieJar};
use actix_http::http::header::{self, Header, HeaderValue, IntoHeaderValue};
use actix_http::http::{HeaderName, HttpTryFrom, Version};
use actix_http::http::{HeaderName, HttpTryFrom, StatusCode, Version};
use actix_http::{h1, Payload, ResponseHead};
use bytes::Bytes;
#[cfg(test)]
@ -49,7 +49,7 @@ pub struct TestResponse {
impl Default for TestResponse {
fn default() -> TestResponse {
TestResponse {
head: ResponseHead::default(),
head: ResponseHead::new(StatusCode::OK),
cookies: CookieJar::new(),
payload: None,
}

41
src/app_service.rs
View file

@ -14,6 +14,7 @@ use crate::config::{AppConfig, ServiceConfig};
use crate::data::{DataFactory, DataFactoryResult};
use crate::error::Error;
use crate::guard::Guard;
use crate::request::{HttpRequest, HttpRequestPool};
use crate::rmap::ResourceMap;
use crate::service::{ServiceFactory, ServiceRequest, ServiceResponse};
@ -21,7 +22,10 @@ type Guards = Vec<Box<Guard>>;
type HttpService<P> = BoxedService<ServiceRequest<P>, ServiceResponse, Error>;
type HttpNewService<P> =
BoxedNewService<(), ServiceRequest<P>, ServiceResponse, Error, ()>;
type BoxedResponse = Box<Future<Item = ServiceResponse, Error = Error>>;
type BoxedResponse = Either<
FutureResult<ServiceResponse, Error>,
Box<Future<Item = ServiceResponse, Error = Error>>,
>;
/// Service factory to convert `Request` to a `ServiceRequest<S>`.
/// It also executes data factories.
@ -191,6 +195,7 @@ where
chain: self.chain.take().unwrap(),
rmap: self.rmap.clone(),
config: self.config.clone(),
pool: HttpRequestPool::create(),
}
.and_then(self.endpoint.take().unwrap()),
))
@ -208,6 +213,7 @@ where
chain: C,
rmap: Rc<ResourceMap>,
config: AppConfig,
pool: &'static HttpRequestPool,
}
impl<C, P> Service for AppInitService<C, P>
@ -224,13 +230,24 @@ where
}
fn call(&mut self, req: Request) -> Self::Future {
let req = ServiceRequest::new(
Path::new(Url::new(req.uri().clone())),
req,
self.rmap.clone(),
self.config.clone(),
);
self.chain.call(req)
let (head, payload) = req.into_parts();
let req = if let Some(mut req) = self.pool.get_request() {
let inner = Rc::get_mut(&mut req.0).unwrap();
inner.path.get_mut().update(&head.uri);
inner.path.reset();
inner.head = head;
req
} else {
HttpRequest::new(
Path::new(Url::new(head.uri.clone())),
head,
self.rmap.clone(),
self.config.clone(),
self.pool,
)
};
self.chain.call(ServiceRequest::from_parts(req, payload))
}
}
@ -353,7 +370,7 @@ impl<P> Service for AppRouting<P> {
type Request = ServiceRequest<P>;
type Response = ServiceResponse;
type Error = Error;
type Future = Either<BoxedResponse, FutureResult<Self::Response, Self::Error>>;
type Future = BoxedResponse;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
if self.ready.is_none() {
@ -376,12 +393,12 @@ impl<P> Service for AppRouting<P> {
});
if let Some((srv, _info)) = res {
Either::A(srv.call(req))
srv.call(req)
} else if let Some(ref mut default) = self.default {
Either::A(default.call(req))
default.call(req)
} else {
let req = req.into_parts().0;
Either::B(ok(ServiceResponse::new(req, Response::NotFound().finish())))
Either::A(ok(ServiceResponse::new(req, Response::NotFound().finish())))
}
}
}

114
src/handler.rs
View file

@ -52,37 +52,21 @@ where
}
}
}
impl<F, T, R> NewService for Handler<F, T, R>
impl<F, T, R> Clone for Handler<F, T, R>
where
F: Factory<T, R>,
R: Responder,
{
type Request = (T, HttpRequest);
type Response = ServiceResponse;
type Error = Void;
type InitError = ();
type Service = HandlerService<F, T, R>;
type Future = FutureResult<Self::Service, ()>;
fn new_service(&self, _: &()) -> Self::Future {
ok(HandlerService {
fn clone(&self) -> Self {
Self {
hnd: self.hnd.clone(),
_t: PhantomData,
})
}
}
}
#[doc(hidden)]
pub struct HandlerService<F, T, R>
where
F: Factory<T, R>,
R: Responder,
{
hnd: F,
_t: PhantomData<(T, R)>,
}
impl<F, T, R> Service for HandlerService<F, T, R>
impl<F, T, R> Service for Handler<F, T, R>
where
F: Factory<T, R>,
R: Responder,
@ -184,41 +168,23 @@ where
}
}
}
impl<F, T, R> NewService for AsyncHandler<F, T, R>
impl<F, T, R> Clone for AsyncHandler<F, T, R>
where
F: AsyncFactory<T, R>,
R: IntoFuture,
R::Item: Into<Response>,
R::Error: Into<Error>,
{
type Request = (T, HttpRequest);
type Response = ServiceResponse;
type Error = Error;
type InitError = ();
type Service = AsyncHandlerService<F, T, R>;
type Future = FutureResult<Self::Service, ()>;
fn new_service(&self, _: &()) -> Self::Future {
ok(AsyncHandlerService {
fn clone(&self) -> Self {
AsyncHandler {
hnd: self.hnd.clone(),
_t: PhantomData,
})
}
}
}
#[doc(hidden)]
pub struct AsyncHandlerService<F, T, R>
where
F: AsyncFactory<T, R>,
R: IntoFuture,
R::Item: Into<Response>,
R::Error: Into<Error>,
{
hnd: F,
_t: PhantomData<(T, R)>,
}
impl<F, T, R> Service for AsyncHandlerService<F, T, R>
impl<F, T, R> Service for AsyncHandler<F, T, R>
where
F: AsyncFactory<T, R>,
R: IntoFuture,
@ -227,7 +193,7 @@ where
{
type Request = (T, HttpRequest);
type Response = ServiceResponse;
type Error = Error;
type Error = Void;
type Future = AsyncHandlerServiceResponse<R::Future>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
@ -255,7 +221,7 @@ where
T::Error: Into<Error>,
{
type Item = ServiceResponse;
type Error = Error;
type Error = Void;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.fut.poll() {
@ -276,46 +242,58 @@ where
}
/// Extract arguments from request
pub struct Extract<P, T: FromRequest<P>> {
pub struct Extract<P, T: FromRequest<P>, S> {
config: Rc<RefCell<Option<Rc<Extensions>>>>,
service: S,
_t: PhantomData<(P, T)>,
}
impl<P, T: FromRequest<P>> Extract<P, T> {
pub fn new(config: Rc<RefCell<Option<Rc<Extensions>>>>) -> Self {
impl<P, T: FromRequest<P>, S> Extract<P, T, S> {
pub fn new(config: Rc<RefCell<Option<Rc<Extensions>>>>, service: S) -> Self {
Extract {
config,
service,
_t: PhantomData,
}
}
}
impl<P, T: FromRequest<P>> NewService for Extract<P, T> {
impl<P, T: FromRequest<P>, S> NewService for Extract<P, T, S>
where
S: Service<Request = (T, HttpRequest), Response = ServiceResponse, Error = Void>
+ Clone,
{
type Request = ServiceRequest<P>;
type Response = (T, HttpRequest);
type Response = ServiceResponse;
type Error = (Error, ServiceRequest<P>);
type InitError = ();
type Service = ExtractService<P, T>;
type Service = ExtractService<P, T, S>;
type Future = FutureResult<Self::Service, ()>;
fn new_service(&self, _: &()) -> Self::Future {
ok(ExtractService {
_t: PhantomData,
config: self.config.borrow().clone(),
service: self.service.clone(),
})
}
}
pub struct ExtractService<P, T: FromRequest<P>> {
pub struct ExtractService<P, T: FromRequest<P>, S> {
config: Option<Rc<Extensions>>,
service: S,
_t: PhantomData<(P, T)>,
}
impl<P, T: FromRequest<P>> Service for ExtractService<P, T> {
impl<P, T: FromRequest<P>, S> Service for ExtractService<P, T, S>
where
S: Service<Request = (T, HttpRequest), Response = ServiceResponse, Error = Void>
+ Clone,
{
type Request = ServiceRequest<P>;
type Response = (T, HttpRequest);
type Response = ServiceResponse;
type Error = (Error, ServiceRequest<P>);
type Future = ExtractResponse<P, T>;
type Future = ExtractResponse<P, T, S>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
@ -328,28 +306,40 @@ impl<P, T: FromRequest<P>> Service for ExtractService<P, T> {
ExtractResponse {
fut,
fut_s: None,
req: Some((req, payload)),
service: self.service.clone(),
}
}
}
pub struct ExtractResponse<P, T: FromRequest<P>> {
pub struct ExtractResponse<P, T: FromRequest<P>, S: Service> {
req: Option<(HttpRequest, Payload<P>)>,
service: S,
fut: <T::Future as IntoFuture>::Future,
fut_s: Option<S::Future>,
}
impl<P, T: FromRequest<P>> Future for ExtractResponse<P, T> {
type Item = (T, HttpRequest);
impl<P, T: FromRequest<P>, S> Future for ExtractResponse<P, T, S>
where
S: Service<Request = (T, HttpRequest), Response = ServiceResponse, Error = Void>,
{
type Item = ServiceResponse;
type Error = (Error, ServiceRequest<P>);
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if let Some(ref mut fut) = self.fut_s {
return fut.poll().map_err(|_| panic!());
}
let item = try_ready!(self.fut.poll().map_err(|e| {
let (req, payload) = self.req.take().unwrap();
let req = ServiceRequest::from_parts(req, payload);
(e.into(), req)
}));
Ok(Async::Ready((item, self.req.take().unwrap().0)))
self.fut_s = Some(self.service.call((item, self.req.take().unwrap().0)));
self.poll()
}
}

80
src/request.rs
View file

@ -1,4 +1,4 @@
use std::cell::{Ref, RefMut};
use std::cell::{Ref, RefCell, RefMut};
use std::fmt;
use std::rc::Rc;
@ -15,29 +15,34 @@ use crate::rmap::ResourceMap;
#[derive(Clone)]
/// An HTTP Request
pub struct HttpRequest {
pub struct HttpRequest(pub(crate) Rc<HttpRequestInner>);
pub(crate) struct HttpRequestInner {
pub(crate) head: Message<RequestHead>,
pub(crate) path: Path<Url>,
rmap: Rc<ResourceMap>,
config: AppConfig,
route_data: Option<Rc<Extensions>>,
pool: &'static HttpRequestPool,
}
impl HttpRequest {
#[inline]
pub(crate) fn new(
head: Message<RequestHead>,
path: Path<Url>,
head: Message<RequestHead>,
rmap: Rc<ResourceMap>,
config: AppConfig,
pool: &'static HttpRequestPool,
) -> HttpRequest {
HttpRequest {
HttpRequest(Rc::new(HttpRequestInner {
head,
path,
rmap,
config,
pool,
route_data: None,
}
}))
}
}
@ -45,7 +50,14 @@ impl HttpRequest {
/// This method returns reference to the request head
#[inline]
pub fn head(&self) -> &RequestHead {
&self.head
&self.0.head
}
/// This method returns muttable reference to the request head.
/// panics if multiple references of http request exists.
#[inline]
pub(crate) fn head_mut(&mut self) -> &mut RequestHead {
&mut Rc::get_mut(&mut self.0).unwrap().head
}
/// Request's uri.
@ -98,7 +110,12 @@ impl HttpRequest {
/// access the matched value for that segment.
#[inline]
pub fn match_info(&self) -> &Path<Url> {
&self.path
&self.0.path
}
#[inline]
pub(crate) fn match_info_mut(&mut self) -> &mut Path<Url> {
&mut Rc::get_mut(&mut self.0).unwrap().path
}
/// Request extensions
@ -141,7 +158,7 @@ impl HttpRequest {
U: IntoIterator<Item = I>,
I: AsRef<str>,
{
self.rmap.url_for(&self, name, elements)
self.0.rmap.url_for(&self, name, elements)
}
/// Generate url for named resource
@ -162,13 +179,13 @@ impl HttpRequest {
/// App config
#[inline]
pub fn app_config(&self) -> &AppConfig {
&self.config
&self.0.config
}
/// Get an application data stored with `App::data()` method during
/// application configuration.
pub fn app_data<T: 'static>(&self) -> Option<Data<T>> {
if let Some(st) = self.config.extensions().get::<Data<T>>() {
if let Some(st) = self.0.config.extensions().get::<Data<T>>() {
Some(st.clone())
} else {
None
@ -178,7 +195,7 @@ impl HttpRequest {
/// Load route data. Route data could be set during
/// route configuration with `Route::data()` method.
pub fn route_data<T: 'static>(&self) -> Option<&RouteData<T>> {
if let Some(ref ext) = self.route_data {
if let Some(ref ext) = self.0.route_data {
ext.get::<RouteData<T>>()
} else {
None
@ -186,7 +203,7 @@ impl HttpRequest {
}
pub(crate) fn set_route_data(&mut self, data: Option<Rc<Extensions>>) {
self.route_data = data;
Rc::get_mut(&mut self.0).unwrap().route_data = data;
}
}
@ -202,13 +219,13 @@ impl HttpMessage for HttpRequest {
/// Request extensions
#[inline]
fn extensions(&self) -> Ref<Extensions> {
self.head.extensions()
self.0.head.extensions()
}
/// Mutable reference to a the request's extensions
#[inline]
fn extensions_mut(&self) -> RefMut<Extensions> {
self.head.extensions_mut()
self.0.head.extensions_mut()
}
#[inline]
@ -217,6 +234,17 @@ impl HttpMessage for HttpRequest {
}
}
impl Drop for HttpRequest {
fn drop(&mut self) {
if Rc::strong_count(&self.0) == 1 {
let v = &mut self.0.pool.0.borrow_mut();
if v.len() < 128 {
v.push(self.0.clone());
}
}
}
}
/// It is possible to get `HttpRequest` as an extractor handler parameter
///
/// ## Example
@ -252,8 +280,8 @@ impl fmt::Debug for HttpRequest {
writeln!(
f,
"\nHttpRequest {:?} {}:{}",
self.head.version,
self.head.method,
self.0.head.version,
self.0.head.method,
self.path()
)?;
if !self.query_string().is_empty() {
@ -270,6 +298,26 @@ impl fmt::Debug for HttpRequest {
}
}
/// Request's objects pool
pub(crate) struct HttpRequestPool(RefCell<Vec<Rc<HttpRequestInner>>>);
impl HttpRequestPool {
pub(crate) fn create() -> &'static HttpRequestPool {
let pool = HttpRequestPool(RefCell::new(Vec::with_capacity(128)));
Box::leak(Box::new(pool))
}
/// Get message from the pool
#[inline]
pub(crate) fn get_request(&self) -> Option<HttpRequest> {
if let Some(inner) = self.0.borrow_mut().pop() {
Some(HttpRequest(inner))
} else {
None
}
}
}
#[cfg(test)]
mod tests {
use super::*;

13
src/resource.rs
View file

@ -487,11 +487,8 @@ impl<P> Service for ResourceService<P> {
type Response = ServiceResponse;
type Error = Error;
type Future = Either<
FutureResult<ServiceResponse, Error>,
Box<Future<Item = ServiceResponse, Error = Error>>,
Either<
Box<Future<Item = Self::Response, Error = Self::Error>>,
FutureResult<Self::Response, Self::Error>,
>,
>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
@ -501,17 +498,17 @@ impl<P> Service for ResourceService<P> {
fn call(&mut self, mut req: ServiceRequest<P>) -> Self::Future {
for route in self.routes.iter_mut() {
if route.check(&mut req) {
return Either::A(route.call(req));
return route.call(req);
}
}
if let Some(ref mut default) = self.default {
Either::B(Either::A(default.call(req)))
default.call(req)
} else {
let req = req.into_parts().0;
Either::B(Either::B(ok(ServiceResponse::new(
Either::A(ok(ServiceResponse::new(
req,
Response::MethodNotAllowed().finish(),
))))
)))
}
}
}

54
src/route.rs
View file

@ -4,6 +4,7 @@ use std::rc::Rc;
use actix_http::{http::Method, Error, Extensions, Response};
use actix_service::{NewService, Service};
use futures::future::{ok, Either, FutureResult};
use futures::{Async, Future, IntoFuture, Poll};
use crate::data::RouteData;
@ -19,7 +20,10 @@ type BoxedRouteService<Req, Res> = Box<
Request = Req,
Response = Res,
Error = Error,
Future = Box<Future<Item = Res, Error = Error>>,
Future = Either<
FutureResult<Res, Error>,
Box<Future<Item = Res, Error = Error>>,
>,
>,
>;
@ -50,11 +54,10 @@ impl<P: 'static> Route<P> {
pub fn new() -> Route<P> {
let data_ref = Rc::new(RefCell::new(None));
Route {
service: Box::new(RouteNewService::new(
Extract::new(data_ref.clone()).and_then(
Handler::new(HttpResponse::NotFound).map_err(|_| panic!()),
),
)),
service: Box::new(RouteNewService::new(Extract::new(
data_ref.clone(),
Handler::new(|| HttpResponse::NotFound()),
))),
guards: Rc::new(Vec::new()),
data: None,
data_ref,
@ -131,7 +134,10 @@ impl<P> Service for RouteService<P> {
type Request = ServiceRequest<P>;
type Response = ServiceResponse;
type Error = Error;
type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;
type Future = Either<
FutureResult<Self::Response, Self::Error>,
Box<Future<Item = Self::Response, Error = Self::Error>>,
>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
self.service.poll_ready()
@ -235,10 +241,10 @@ impl<P: 'static> Route<P> {
T: FromRequest<P> + 'static,
R: Responder + 'static,
{
self.service = Box::new(RouteNewService::new(
Extract::new(self.data_ref.clone())
.and_then(Handler::new(handler).map_err(|_| panic!())),
));
self.service = Box::new(RouteNewService::new(Extract::new(
self.data_ref.clone(),
Handler::new(handler),
)));
self
}
@ -277,10 +283,10 @@ impl<P: 'static> Route<P> {
R::Item: Into<Response>,
R::Error: Into<Error>,
{
self.service = Box::new(RouteNewService::new(
Extract::new(self.data_ref.clone())
.and_then(AsyncHandler::new(handler).map_err(|_| panic!())),
));
self.service = Box::new(RouteNewService::new(Extract::new(
self.data_ref.clone(),
AsyncHandler::new(handler),
)));
self
}
@ -394,17 +400,25 @@ where
type Request = ServiceRequest<P>;
type Response = ServiceResponse;
type Error = Error;
type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;
type Future = Either<
FutureResult<Self::Response, Self::Error>,
Box<Future<Item = Self::Response, Error = Self::Error>>,
>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
self.service.poll_ready().map_err(|(e, _)| e)
}
fn call(&mut self, req: ServiceRequest<P>) -> Self::Future {
Box::new(self.service.call(req).then(|res| match res {
Ok(res) => Ok(res),
Err((err, req)) => Ok(req.error_response(err)),
}))
let mut fut = self.service.call(req);
match fut.poll() {
Ok(Async::Ready(res)) => Either::A(ok(res)),
Err((e, req)) => Either::A(ok(req.error_response(e))),
Ok(Async::NotReady) => Either::B(Box::new(fut.then(|res| match res {
Ok(res) => Ok(res),
Err((err, req)) => Ok(req.error_response(err)),
}))),
}
}
}

5
src/scope.rs
View file

@ -24,7 +24,10 @@ type Guards = Vec<Box<Guard>>;
type HttpService<P> = BoxedService<ServiceRequest<P>, ServiceResponse, Error>;
type HttpNewService<P> =
BoxedNewService<(), ServiceRequest<P>, ServiceResponse, Error, ()>;
type BoxedResponse = Box<Future<Item = ServiceResponse, Error = Error>>;
type BoxedResponse = Either<
FutureResult<ServiceResponse, Error>,
Box<Future<Item = ServiceResponse, Error = Error>>,
>;
/// Resources scope.
///

31
src/service.rs
View file

@ -1,13 +1,12 @@
use std::cell::{Ref, RefMut};
use std::fmt;
use std::marker::PhantomData;
use std::rc::Rc;
use actix_http::body::{Body, MessageBody, ResponseBody};
use actix_http::http::{HeaderMap, Method, StatusCode, Uri, Version};
use actix_http::{
Error, Extensions, HttpMessage, Payload, PayloadStream, Request, RequestHead,
Response, ResponseHead,
Error, Extensions, HttpMessage, Payload, PayloadStream, RequestHead, Response,
ResponseHead,
};
use actix_router::{Path, Resource, Url};
use futures::future::{ok, FutureResult, IntoFuture};
@ -15,7 +14,6 @@ use futures::future::{ok, FutureResult, IntoFuture};
use crate::config::{AppConfig, ServiceConfig};
use crate::data::Data;
use crate::request::HttpRequest;
use crate::rmap::ResourceMap;
pub trait HttpServiceFactory<P> {
fn register(self, config: &mut ServiceConfig<P>);
@ -56,19 +54,6 @@ pub struct ServiceRequest<P = PayloadStream> {
}
impl<P> ServiceRequest<P> {
pub(crate) fn new(
path: Path<Url>,
request: Request<P>,
rmap: Rc<ResourceMap>,
config: AppConfig,
) -> Self {
let (head, payload) = request.into_parts();
ServiceRequest {
payload,
req: HttpRequest::new(head, path, rmap, config),
}
}
/// Construct service request from parts
pub fn from_parts(req: HttpRequest, payload: Payload<P>) -> Self {
ServiceRequest { req, payload }
@ -95,13 +80,13 @@ impl<P> ServiceRequest<P> {
/// This method returns reference to the request head
#[inline]
pub fn head(&self) -> &RequestHead {
&self.req.head
&self.req.head()
}
/// This method returns reference to the request head
#[inline]
pub fn head_mut(&mut self) -> &mut RequestHead {
&mut self.req.head
self.req.head_mut()
}
/// Request's uri.
@ -160,12 +145,12 @@ impl<P> ServiceRequest<P> {
/// access the matched value for that segment.
#[inline]
pub fn match_info(&self) -> &Path<Url> {
&self.req.path
self.req.match_info()
}
#[inline]
pub fn match_info_mut(&mut self) -> &mut Path<Url> {
&mut self.req.path
self.req.match_info_mut()
}
/// Service configuration
@ -203,13 +188,13 @@ impl<P> HttpMessage for ServiceRequest<P> {
/// Request extensions
#[inline]
fn extensions(&self) -> Ref<Extensions> {
self.req.head.extensions()
self.req.extensions()
}
/// Mutable reference to a the request's extensions
#[inline]
fn extensions_mut(&self) -> RefMut<Extensions> {
self.req.head.extensions_mut()
self.req.extensions_mut()
}
#[inline]

42
src/test.rs
View file

@ -17,6 +17,7 @@ use futures::future::{lazy, Future};
use crate::config::{AppConfig, AppConfigInner};
use crate::data::RouteData;
use crate::dev::{Body, Payload};
use crate::request::HttpRequestPool;
use crate::rmap::ResourceMap;
use crate::service::{ServiceRequest, ServiceResponse};
use crate::{Error, HttpRequest, HttpResponse};
@ -326,14 +327,17 @@ impl TestRequest {
/// Complete request creation and generate `ServiceRequest` instance
pub fn to_srv_request(mut self) -> ServiceRequest<PayloadStream> {
let req = self.req.finish();
let (head, payload) = self.req.finish().into_parts();
ServiceRequest::new(
Path::new(Url::new(req.uri().clone())),
req,
let req = HttpRequest::new(
Path::new(Url::new(head.uri.clone())),
head,
Rc::new(self.rmap),
AppConfig::new(self.config),
)
HttpRequestPool::create(),
);
ServiceRequest::from_parts(req, payload)
}
/// Complete request creation and generate `ServiceResponse` instance
@ -343,34 +347,32 @@ impl TestRequest {
/// Complete request creation and generate `HttpRequest` instance
pub fn to_http_request(mut self) -> HttpRequest {
let req = self.req.finish();
let (head, _) = self.req.finish().into_parts();
let mut req = ServiceRequest::new(
Path::new(Url::new(req.uri().clone())),
req,
let mut req = HttpRequest::new(
Path::new(Url::new(head.uri.clone())),
head,
Rc::new(self.rmap),
AppConfig::new(self.config),
)
.into_parts()
.0;
HttpRequestPool::create(),
);
req.set_route_data(Some(Rc::new(self.route_data)));
req
}
/// Complete request creation and generate `HttpRequest` and `Payload` instances
pub fn to_http_parts(mut self) -> (HttpRequest, Payload) {
let req = self.req.finish();
let (head, payload) = self.req.finish().into_parts();
let (mut req, pl) = ServiceRequest::new(
Path::new(Url::new(req.uri().clone())),
req,
let mut req = HttpRequest::new(
Path::new(Url::new(head.uri.clone())),
head,
Rc::new(self.rmap),
AppConfig::new(self.config),
)
.into_parts();
HttpRequestPool::create(),
);
req.set_route_data(Some(Rc::new(self.route_data)));
(req, pl)
(req, payload)
}
/// Runs the provided future, blocking the current thread until the future