mirrors/actix-web
1
0
Fork 0
mirror of https://github.com/actix/actix-web.git synced 2024-11-26 19:41:12 +00:00
Code Issues Projects Releases Wiki Activity

Merge branch 'master' into ranges

Browse source
This commit is contained in:
Nikolay Kim 2018-05-17 11:16:08 -07:00 committed by GitHub
commit a5692d4ecf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
42 changed files with 1910 additions and 549 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.appveyor.yml
View file

@ -31,13 +31,13 @@ environment:
CHANNEL: beta
# Nightly channel
- TARGET: i686-pc-windows-gnu
CHANNEL: nightly-2017-12-21
CHANNEL: nightly
- TARGET: i686-pc-windows-msvc
CHANNEL: nightly-2017-12-21
CHANNEL: nightly
- TARGET: x86_64-pc-windows-gnu
CHANNEL: nightly-2017-12-21
CHANNEL: nightly
- TARGET: x86_64-pc-windows-msvc
CHANNEL: nightly-2017-12-21
CHANNEL: nightly
# Install Rust and Cargo
# (Based on from https://github.com/rust-lang/libc/blob/master/appveyor.yml)

17
.travis.yml
View file

@ -31,8 +31,17 @@ before_script:
script:
- |
if [[ "$TRAVIS_RUST_VERSION" != "beta" ]]; then
cargo clean
cargo test --features="alpn,tls" -- --nocapture
fi
- |
if [[ "$TRAVIS_RUST_VERSION" == "beta" ]]; then
bash <(curl https://raw.githubusercontent.com/xd009642/tarpaulin/master/travis-install.sh)
USE_SKEPTIC=1 cargo tarpaulin --out Xml --no-count
bash <(curl -s https://codecov.io/bash)
echo "Uploaded code coverage"
fi
# Upload docs
after_success:
@ -44,11 +53,3 @@ after_success:
./ghp-import/ghp_import.py -n -p -f -m "Documentation upload" -r https://"$GH_TOKEN"@github.com/"$TRAVIS_REPO_SLUG.git" target/doc &&
echo "Uploaded documentation"
fi
- |
if [[ "$TRAVIS_OS_NAME" == "linux" && "$TRAVIS_RUST_VERSION" == "beta" ]]; then
bash <(curl https://raw.githubusercontent.com/xd009642/tarpaulin/master/travis-install.sh)
USE_SKEPTIC=1 cargo tarpaulin --out Xml
bash <(curl -s https://codecov.io/bash)
echo "Uploaded code coverage"
fi

46
CHANGES.md
View file

@ -1,6 +1,50 @@
# Changes
## 0.6.0 (...)
## 0.6.6 (2018-05-17)
* Panic during middleware execution #226
* Add support for listen_tls/listen_ssl #224
* Implement extractor for `Session`
## 0.6.5 (2018-05-15)
* Fix error handling during request decoding #222
## 0.6.4 (2018-05-11)
* Fix segfault in ServerSettings::get_response_builder()
## 0.6.3 (2018-05-10)
* Add `Router::with_async()` method for async handler registration.
* Added error response functions for 501,502,503,504
* Fix client request timeout handling
## 0.6.2 (2018-05-09)
* WsWriter trait is optional.
## 0.6.1 (2018-05-08)
* Fix http/2 payload streaming #215
* Fix connector's default `keep-alive` and `lifetime` settings #212
* Send `ErrorNotFound` instead of `ErrorBadRequest` when path extractor fails #214
* Allow to exclude certain endpoints from logging #211
## 0.6.0 (2018-05-08)
* Add route scopes #202

2
Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "actix-web"
version = "0.6.0-dev"
version = "0.6.6"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix web is a simple, pragmatic and extremely fast web framework for Rust."
readme = "README.md"

18
MIGRATION.md
View file

@ -1,5 +1,7 @@
## Migration from 0.5 to 0.6
* `Path<T>` extractor return `ErrorNotFound` on failure instead of `ErrorBadRequest`
* `ws::Message::Close` now includes optional close reason.
`ws::CloseCode::Status` and `ws::CloseCode::Empty` have been removed.
@ -11,6 +13,17 @@
* `HttpRequest::extensions()` returns read only reference to the request's Extension
`HttpRequest::extensions_mut()` returns mutable reference.
* Instead of
`use actix_web::middleware::{
CookieSessionBackend, CookieSessionError, RequestSession,
Session, SessionBackend, SessionImpl, SessionStorage};`
use `actix_web::middleware::session`
`use actix_web::middleware::session{CookieSessionBackend, CookieSessionError,
RequestSession, Session, SessionBackend, SessionImpl, SessionStorage};`
* `FromRequest::from_request()` accepts mutable reference to a request
* `FromRequest::Result` has to implement `Into<Reply<Self>>`
@ -19,7 +32,7 @@
https://actix.rs/actix-web/actix_web/trait.Responder.html#tymethod.respond_to)
is generic over `S`
* `HttpRequest::query()` is deprecated. Use `Query` extractor.
* Use `Query` extractor instead of HttpRequest::query()`.
```rust
fn index(q: Query<HashMap<String, String>>) -> Result<..> {
@ -33,6 +46,9 @@
let q = Query::<HashMap<String, String>>::extract(req);
```
* Websocket operations are implemented as `WsWriter` trait.
you need to use `use actix_web::ws::WsWriter`
## Migration from 0.4 to 0.5

3
README.md
View file

@ -18,6 +18,7 @@ Actix web is a simple, pragmatic and extremely fast web framework for Rust.
[DefaultHeaders](https://actix.rs/book/actix-web/sec-9-middlewares.html#default-headers),
[CORS](https://actix.rs/actix-web/actix_web/middleware/cors/index.html),
[CSRF](https://actix.rs/actix-web/actix_web/middleware/csrf/index.html))
* Includes an asynchronous [HTTP client](https://github.com/actix/actix-web/blob/master/src/client/mod.rs)
* Built on top of [Actix actor framework](https://github.com/actix/actix)
## Documentation & community resources
@ -36,7 +37,7 @@ extern crate actix_web;
use actix_web::{http, server, App, Path};
fn index(info: Path<(u32, String)>) -> String {
format!("Hello {}! id:{}", info.0, info.1)
format!("Hello {}! id:{}", info.1, info.0)
}
fn main() {

9
build.rs
View file

@ -1,8 +1,15 @@
extern crate version_check;
fn main() {
match version_check::is_min_version("1.26.0") {
Some((true, _)) => println!("cargo:rustc-cfg=actix_impl_trait"),
_ => (),
};
match version_check::is_nightly() {
Some(true) => println!("cargo:rustc-cfg=actix_nightly"),
Some(true) => {
println!("cargo:rustc-cfg=actix_nightly");
println!("cargo:rustc-cfg=actix_impl_trait");
}
Some(false) => (),
None => (),
};

111
src/application.rs
View file

@ -9,6 +9,7 @@ use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::Middleware;
use pipeline::{HandlerType, Pipeline, PipelineHandler};
use pred::Predicate;
use resource::ResourceHandler;
use router::{Resource, Router};
use scope::Scope;
@ -29,7 +30,12 @@ pub(crate) struct Inner<S> {
default: ResourceHandler<S>,
encoding: ContentEncoding,
resources: Vec<ResourceHandler<S>>,
handlers: Vec<(String, Box<RouteHandler<S>>)>,
handlers: Vec<PrefixHandlerType<S>>,
}
enum PrefixHandlerType<S> {
Handler(String, Box<RouteHandler<S>>),
Scope(Resource, Box<RouteHandler<S>>, Vec<Box<Predicate<S>>>),
}
impl<S: 'static> PipelineHandler<S> for Inner<S> {
@ -44,7 +50,10 @@ impl<S: 'static> PipelineHandler<S> for Inner<S> {
HandlerType::Normal(idx) => {
self.resources[idx].handle(req, Some(&mut self.default))
}
HandlerType::Handler(idx) => self.handlers[idx].1.handle(req),
HandlerType::Handler(idx) => match self.handlers[idx] {
PrefixHandlerType::Handler(_, ref mut hnd) => hnd.handle(req),
PrefixHandlerType::Scope(_, ref mut hnd, _) => hnd.handle(req),
},
HandlerType::Default => self.default.handle(req, None),
}
}
@ -62,27 +71,55 @@ impl<S: 'static> HttpApplication<S> {
HandlerType::Normal(idx)
} else {
let inner = self.as_ref();
for idx in 0..inner.handlers.len() {
let &(ref prefix, _) = &inner.handlers[idx];
let m = {
let path = &req.path()[inner.prefix..];
path.starts_with(prefix)
&& (path.len() == prefix.len()
|| path.split_at(prefix.len()).1.starts_with('/'))
};
let path: &'static str =
unsafe { &*(&req.path()[inner.prefix..] as *const _) };
let path_len = path.len();
'outer: for idx in 0..inner.handlers.len() {
match inner.handlers[idx] {
PrefixHandlerType::Handler(ref prefix, _) => {
let m = {
path.starts_with(prefix)
&& (path_len == prefix.len()
|| path.split_at(prefix.len()).1.starts_with('/'))
};
if m {
let prefix_len = inner.prefix + prefix.len();
let path: &'static str =
unsafe { &*(&req.path()[prefix_len..] as *const _) };
if m {
let prefix_len = inner.prefix + prefix.len();
let path: &'static str =
unsafe { &*(&req.path()[prefix_len..] as *const _) };
req.set_prefix_len(prefix_len as u16);
if path.is_empty() {
req.match_info_mut().add("tail", "/");
} else {
req.match_info_mut().add("tail", path);
req.set_prefix_len(prefix_len as u16);
if path.is_empty() {
req.match_info_mut().add("tail", "/");
} else {
req.match_info_mut().add("tail", path);
}
return HandlerType::Handler(idx);
}
}
PrefixHandlerType::Scope(ref pattern, _, ref filters) => {
if let Some(prefix_len) =
pattern.match_prefix_with_params(path, req.match_info_mut())
{
for filter in filters {
if !filter.check(req) {
continue 'outer;
}
}
let prefix_len = inner.prefix + prefix_len - 1;
let path: &'static str =
unsafe { &*(&req.path()[prefix_len..] as *const _) };
req.set_prefix_len(prefix_len as u16);
if path.is_empty() {
req.match_info_mut().set("tail", "/");
} else {
req.match_info_mut().set("tail", path);
}
return HandlerType::Handler(idx);
}
}
return HandlerType::Handler(idx);
}
}
HandlerType::Default
@ -131,7 +168,7 @@ struct ApplicationParts<S> {
settings: ServerSettings,
default: ResourceHandler<S>,
resources: Vec<(Resource, Option<ResourceHandler<S>>)>,
handlers: Vec<(String, Box<RouteHandler<S>>)>,
handlers: Vec<PrefixHandlerType<S>>,
external: HashMap<String, Resource>,
encoding: ContentEncoding,
middlewares: Vec<Box<Middleware<S>>>,
@ -305,7 +342,7 @@ where
/// Configure scope for common root path.
///
/// Scopes collect multiple paths under a common path prefix.
/// Scope path can not contain variable path segments as resources.
/// Scope path can contain variable path segments as resources.
///
/// ```rust
/// # extern crate actix_web;
@ -313,7 +350,7 @@ where
///
/// fn main() {
/// let app = App::new()
/// .scope("/app", |scope| {
/// .scope("/{project_id}", |scope| {
/// scope.resource("/path1", |r| r.f(|_| HttpResponse::Ok()))
/// .resource("/path2", |r| r.f(|_| HttpResponse::Ok()))
/// .resource("/path3", |r| r.f(|_| HttpResponse::MethodNotAllowed()))
@ -322,24 +359,32 @@ where
/// ```
///
/// In the above example, three routes get added:
/// * /app/path1
/// * /app/path2
/// * /app/path3
/// * /{project_id}/path1
/// * /{project_id}/path2
/// * /{project_id}/path3
///
pub fn scope<F>(mut self, path: &str, f: F) -> App<S>
where
F: FnOnce(Scope<S>) -> Scope<S>,
{
{
let scope = Box::new(f(Scope::new()));
let mut scope = Box::new(f(Scope::new()));
let mut path = path.trim().trim_right_matches('/').to_owned();
if !path.is_empty() && !path.starts_with('/') {
path.insert(0, '/')
}
if !path.ends_with('/') {
path.push('/');
}
let parts = self.parts.as_mut().expect("Use after finish");
parts.handlers.push((path, scope));
let filters = scope.take_filters();
parts.handlers.push(PrefixHandlerType::Scope(
Resource::prefix("", &path),
scope,
filters,
));
}
self
}
@ -496,11 +541,15 @@ where
if !path.is_empty() && !path.starts_with('/') {
path.insert(0, '/')
}
if path.len() > 1 && path.ends_with('/') {
path.pop();
}
let parts = self.parts.as_mut().expect("Use after finish");
parts
.handlers
.push((path, Box::new(WrapHandler::new(handler))));
parts.handlers.push(PrefixHandlerType::Handler(
path,
Box::new(WrapHandler::new(handler)),
));
}
self
}

18
src/body.rs
View file

@ -62,10 +62,28 @@ impl Body {
}
}
/// Is this binary empy.
#[inline]
pub fn is_empty(&self) -> bool {
match *self {
Body::Empty => true,
_ => false,
}
}
/// Create body from slice (copy)
pub fn from_slice(s: &[u8]) -> Body {
Body::Binary(Binary::Bytes(Bytes::from(s)))
}
/// Is this binary body.
#[inline]
pub(crate) fn binary(self) -> Binary {
match self {
Body::Binary(b) => b,
_ => panic!(),
}
}
}
impl PartialEq for Body {

10
src/client/connector.rs
View file

@ -223,8 +223,8 @@ impl Default for ClientConnector {
pool: Rc::new(Pool::new(Rc::clone(&_modified))),
pool_modified: _modified,
connector: builder.build().unwrap(),
conn_lifetime: Duration::from_secs(15),
conn_keep_alive: Duration::from_secs(75),
conn_lifetime: Duration::from_secs(75),
conn_keep_alive: Duration::from_secs(15),
limit: 100,
limit_per_host: 0,
acquired: 0,
@ -243,8 +243,8 @@ impl Default for ClientConnector {
subscriber: None,
pool: Rc::new(Pool::new(Rc::clone(&_modified))),
pool_modified: _modified,
conn_lifetime: Duration::from_secs(15),
conn_keep_alive: Duration::from_secs(75),
conn_lifetime: Duration::from_secs(75),
conn_keep_alive: Duration::from_secs(15),
limit: 100,
limit_per_host: 0,
acquired: 0,
@ -347,6 +347,7 @@ impl ClientConnector {
/// Keep-alive period is the period between connection usage. If
/// the delay between repeated usages of the same connection
/// exceeds this period, the connection is closed.
/// Default keep-alive period is 15 seconds.
pub fn conn_keep_alive(mut self, dur: Duration) -> Self {
self.conn_keep_alive = dur;
self
@ -356,6 +357,7 @@ impl ClientConnector {
///
/// Connection lifetime is max lifetime of any opened connection
/// until it is closed regardless of keep-alive period.
/// Default lifetime period is 75 seconds.
pub fn conn_lifetime(mut self, dur: Duration) -> Self {
self.conn_lifetime = dur;
self

1
src/client/mod.rs
View file

@ -49,6 +49,7 @@ use httpresponse::HttpResponse;
impl ResponseError for SendRequestError {
fn error_response(&self) -> HttpResponse {
match *self {
SendRequestError::Timeout => HttpResponse::GatewayTimeout(),
SendRequestError::Connector(_) => HttpResponse::BadGateway(),
_ => HttpResponse::InternalServerError(),
}.into()

12
src/client/pipeline.rs
View file

@ -194,6 +194,7 @@ impl Future for SendRequest {
self.state = State::Send(pl);
}
State::Send(mut pl) => {
pl.poll_timeout()?;
pl.poll_write().map_err(|e| {
io::Error::new(io::ErrorKind::Other, format!("{}", e).as_str())
})?;
@ -315,7 +316,7 @@ impl Pipeline {
{
Async::NotReady => need_run = true,
Async::Ready(_) => {
let _ = self.poll_timeout().map_err(|e| {
self.poll_timeout().map_err(|e| {
io::Error::new(io::ErrorKind::Other, format!("{}", e))
})?;
}
@ -371,16 +372,15 @@ impl Pipeline {
}
}
fn poll_timeout(&mut self) -> Poll<(), SendRequestError> {
fn poll_timeout(&mut self) -> Result<(), SendRequestError> {
if self.timeout.is_some() {
match self.timeout.as_mut().unwrap().poll() {
Ok(Async::Ready(())) => Err(SendRequestError::Timeout),
Ok(Async::NotReady) => Ok(Async::NotReady),
Ok(Async::Ready(())) => return Err(SendRequestError::Timeout),
Ok(Async::NotReady) => (),
Err(_) => unreachable!(),
}
} else {
Ok(Async::NotReady)
}
Ok(())
}
#[inline]

99
src/error.rs
View file

@ -88,7 +88,7 @@ impl fmt::Debug for Error {
}
}
/// `HttpResponse` for `Error`
/// Convert `Error` to a `HttpResponse` instance
impl From<Error> for HttpResponse {
fn from(err: Error) -> Self {
HttpResponse::from_error(err)
@ -317,10 +317,7 @@ pub enum HttpRangeError {
/// Return `BadRequest` for `HttpRangeError`
impl ResponseError for HttpRangeError {
fn error_response(&self) -> HttpResponse {
HttpResponse::with_body(
StatusCode::BAD_REQUEST,
"Invalid Range header provided",
)
HttpResponse::with_body(StatusCode::BAD_REQUEST, "Invalid Range header provided")
}
}
@ -757,6 +754,46 @@ where
InternalError::new(err, StatusCode::INTERNAL_SERVER_ERROR).into()
}
/// Helper function that creates wrapper of any error and
/// generate *NOT IMPLEMENTED* response.
#[allow(non_snake_case)]
pub fn ErrorNotImplemented<T>(err: T) -> Error
where
T: Send + Sync + fmt::Debug + fmt::Display + 'static,
{
InternalError::new(err, StatusCode::NOT_IMPLEMENTED).into()
}
/// Helper function that creates wrapper of any error and
/// generate *BAD GATEWAY* response.
#[allow(non_snake_case)]
pub fn ErrorBadGateway<T>(err: T) -> Error
where
T: Send + Sync + fmt::Debug + fmt::Display + 'static,
{
InternalError::new(err, StatusCode::BAD_GATEWAY).into()
}
/// Helper function that creates wrapper of any error and
/// generate *SERVICE UNAVAILABLE* response.
#[allow(non_snake_case)]
pub fn ErrorServiceUnavailable<T>(err: T) -> Error
where
T: Send + Sync + fmt::Debug + fmt::Display + 'static,
{
InternalError::new(err, StatusCode::SERVICE_UNAVAILABLE).into()
}
/// Helper function that creates wrapper of any error and
/// generate *GATEWAY TIMEOUT* response.
#[allow(non_snake_case)]
pub fn ErrorGatewayTimeout<T>(err: T) -> Error
where
T: Send + Sync + fmt::Debug + fmt::Display + 'static,
{
InternalError::new(err, StatusCode::GATEWAY_TIMEOUT).into()
}
#[cfg(test)]
mod tests {
use super::*;
@ -850,7 +887,7 @@ mod tests {
}
macro_rules! from {
($from: expr => $error: pat) => {
($from:expr => $error:pat) => {
match ParseError::from($from) {
e @ $error => {
assert!(format!("{}", e).len() >= 5);
@ -861,7 +898,7 @@ mod tests {
}
macro_rules! from_and_cause {
($from: expr => $error: pat) => {
($from:expr => $error:pat) => {
match ParseError::from($from) {
e @ $error => {
let desc = format!("{}", e.cause().unwrap());
@ -912,4 +949,52 @@ mod tests {
let resp: HttpResponse = err.error_response();
assert_eq!(resp.status(), StatusCode::OK);
}
#[test]
fn test_error_helpers() {
let r: HttpResponse = ErrorBadRequest("err").into();
assert_eq!(r.status(), StatusCode::BAD_REQUEST);
let r: HttpResponse = ErrorUnauthorized("err").into();
assert_eq!(r.status(), StatusCode::UNAUTHORIZED);
let r: HttpResponse = ErrorForbidden("err").into();
assert_eq!(r.status(), StatusCode::FORBIDDEN);
let r: HttpResponse = ErrorNotFound("err").into();
assert_eq!(r.status(), StatusCode::NOT_FOUND);
let r: HttpResponse = ErrorMethodNotAllowed("err").into();
assert_eq!(r.status(), StatusCode::METHOD_NOT_ALLOWED);
let r: HttpResponse = ErrorRequestTimeout("err").into();
assert_eq!(r.status(), StatusCode::REQUEST_TIMEOUT);
let r: HttpResponse = ErrorConflict("err").into();
assert_eq!(r.status(), StatusCode::CONFLICT);
let r: HttpResponse = ErrorGone("err").into();
assert_eq!(r.status(), StatusCode::GONE);
let r: HttpResponse = ErrorPreconditionFailed("err").into();
assert_eq!(r.status(), StatusCode::PRECONDITION_FAILED);
let r: HttpResponse = ErrorExpectationFailed("err").into();
assert_eq!(r.status(), StatusCode::EXPECTATION_FAILED);
let r: HttpResponse = ErrorInternalServerError("err").into();
assert_eq!(r.status(), StatusCode::INTERNAL_SERVER_ERROR);
let r: HttpResponse = ErrorNotImplemented("err").into();
assert_eq!(r.status(), StatusCode::NOT_IMPLEMENTED);
let r: HttpResponse = ErrorBadGateway("err").into();
assert_eq!(r.status(), StatusCode::BAD_GATEWAY);
let r: HttpResponse = ErrorServiceUnavailable("err").into();
assert_eq!(r.status(), StatusCode::SERVICE_UNAVAILABLE);
let r: HttpResponse = ErrorGatewayTimeout("err").into();
assert_eq!(r.status(), StatusCode::GATEWAY_TIMEOUT);
}
}

4
src/extractor.rs
View file

@ -11,7 +11,7 @@ use serde::de::{self, DeserializeOwned};
use serde_urlencoded;
use de::PathDeserializer;
use error::{Error, ErrorBadRequest};
use error::{Error, ErrorNotFound, ErrorBadRequest};
use handler::{AsyncResult, FromRequest};
use httpmessage::{HttpMessage, MessageBody, UrlEncoded};
use httprequest::HttpRequest;
@ -108,7 +108,7 @@ where
fn from_request(req: &HttpRequest<S>, _: &Self::Config) -> Self::Result {
let req = req.clone();
de::Deserialize::deserialize(PathDeserializer::new(&req))
.map_err(|e| e.into())
.map_err(ErrorNotFound)
.map(|inner| Path { inner })
}
}

7
src/handler.rs
View file

@ -492,14 +492,15 @@ where
/// }
///
/// /// extract path info using serde
/// fn index(state: State<MyApp>, info: Path<Info>) -> String {
/// format!("{} {}!", state.msg, info.username)
/// fn index(data: (State<MyApp>, Path<Info>)) -> String {
/// let (state, path) = data;
/// format!("{} {}!", state.msg, path.username)
/// }
///
/// fn main() {
/// let app = App::with_state(MyApp{msg: "Welcome"}).resource(
/// "/{username}/index.html", // <- define path parameters
/// |r| r.method(http::Method::GET).with2(index)); // <- use `with` extractor
/// |r| r.method(http::Method::GET).with(index)); // <- use `with` extractor
/// }
/// ```
pub struct State<S>(HttpRequest<S>);

60
src/httprequest.rs
View file

@ -27,7 +27,6 @@ use uri::Url as InnerUrl;
bitflags! {
pub(crate) struct MessageFlags: u8 {
const QUERY = 0b0000_0001;
const KEEPALIVE = 0b0000_0010;
}
}
@ -40,8 +39,6 @@ pub struct HttpInnerMessage {
pub headers: HeaderMap,
pub extensions: Extensions,
pub params: Params<'static>,
pub cookies: Option<Vec<Cookie<'static>>>,
pub query: Params<'static>,
pub addr: Option<SocketAddr>,
pub payload: Option<Payload>,
pub info: Option<ConnectionInfo<'static>>,
@ -49,6 +46,9 @@ pub struct HttpInnerMessage {
resource: RouterResource,
}
struct Query(Params<'static>);
struct Cookies(Vec<Cookie<'static>>);
#[derive(Debug, Copy, Clone, PartialEq)]
enum RouterResource {
Notset,
@ -64,9 +64,7 @@ impl Default for HttpInnerMessage {
headers: HeaderMap::with_capacity(16),
flags: MessageFlags::empty(),
params: Params::new(),
query: Params::new(),
addr: None,
cookies: None,
payload: None,
extensions: Extensions::new(),
info: None,
@ -91,7 +89,6 @@ impl HttpInnerMessage {
self.addr = None;
self.info = None;
self.flags = MessageFlags::empty();
self.cookies = None;
self.payload = None;
self.prefix = 0;
self.resource = RouterResource::Notset;
@ -109,7 +106,10 @@ impl HttpRequest<()> {
/// Construct a new Request.
#[inline]
pub fn new(
method: Method, uri: Uri, version: Version, headers: HeaderMap,
method: Method,
uri: Uri,
version: Version,
headers: HeaderMap,
payload: Option<Payload>,
) -> HttpRequest {
let url = InnerUrl::new(uri);
@ -121,9 +121,7 @@ impl HttpRequest<()> {
headers,
payload,
params: Params::new(),
query: Params::new(),
extensions: Extensions::new(),
cookies: None,
addr: None,
info: None,
prefix: 0,
@ -306,7 +304,9 @@ impl<S> HttpRequest<S> {
/// }
/// ```
pub fn url_for<U, I>(
&self, name: &str, elements: U,
&self,
name: &str,
elements: U,
) -> Result<Url, UrlGenerationError>
where
U: IntoIterator<Item = I>,
@ -369,20 +369,20 @@ impl<S> HttpRequest<S> {
}
#[doc(hidden)]
#[deprecated(since = "0.6.0", note = "please use `Query<T>` extractor")]
/// Get a reference to the Params object.
/// Params is a container for url query parameters.
pub fn query(&self) -> &Params {
if !self.as_ref().flags.contains(MessageFlags::QUERY) {
self.as_mut().flags.insert(MessageFlags::QUERY);
let params: &mut Params =
unsafe { mem::transmute(&mut self.as_mut().query) };
params.clear();
pub fn query<'a>(&'a self) -> &'a Params {
if let None = self.extensions().get::<Query>() {
let mut params: Params<'a> = Params::new();
for (key, val) in form_urlencoded::parse(self.query_string().as_ref()) {
params.add(key, val);
}
let params: Params<'static> = unsafe { mem::transmute(params) };
self.as_mut().extensions.insert(Query(params));
}
unsafe { mem::transmute(&self.as_ref().query) }
let params: &Params<'a> =
unsafe { mem::transmute(&self.extensions().get::<Query>().unwrap().0) };
params
}
/// The query string in the URL.
@ -399,7 +399,7 @@ impl<S> HttpRequest<S> {
/// Load request cookies.
pub fn cookies(&self) -> Result<&Vec<Cookie<'static>>, CookieParseError> {
if self.as_ref().cookies.is_none() {
if let None = self.extensions().get::<Query>() {
let msg = self.as_mut();
let mut cookies = Vec::new();
for hdr in msg.headers.get_all(header::COOKIE) {
@ -410,9 +410,9 @@ impl<S> HttpRequest<S> {
}
}
}
msg.cookies = Some(cookies);
msg.extensions.insert(Cookies(cookies));
}
Ok(&self.as_ref().cookies.as_ref().unwrap())
Ok(&self.extensions().get::<Cookies>().unwrap().0)
}
/// Return request cookie.
@ -427,6 +427,12 @@ impl<S> HttpRequest<S> {
None
}
pub(crate) fn set_cookies(&mut self, cookies: Option<Vec<Cookie<'static>>>) {
if let Some(cookies) = cookies {
self.extensions_mut().insert(Cookies(cookies));
}
}
/// Get a reference to the Params object.
///
/// Params is a container for url parameters.
@ -664,10 +670,8 @@ mod tests {
let mut resource = ResourceHandler::<()>::default();
resource.name("index");
let routes = vec![(
Resource::new("index", "/user/{name}.{ext}"),
Some(resource),
)];
let routes =
vec![(Resource::new("index", "/user/{name}.{ext}"), Some(resource))];
let (router, _) = Router::new("/", ServerSettings::default(), routes);
assert!(router.has_route("/user/test.html"));
assert!(!router.has_route("/test/unknown"));
@ -696,10 +700,8 @@ mod tests {
let mut resource = ResourceHandler::<()>::default();
resource.name("index");
let routes = vec![(
Resource::new("index", "/user/{name}.{ext}"),
Some(resource),
)];
let routes =
vec![(Resource::new("index", "/user/{name}.{ext}"), Some(resource))];
let (router, _) = Router::new("/prefix/", ServerSettings::default(), routes);
assert!(router.has_route("/user/test.html"));
assert!(!router.has_route("/prefix/user/test.html"));

95
src/httpresponse.rs
View file

@ -466,10 +466,7 @@ impl HttpResponseBuilder {
jar.add(cookie.into_owned());
self.cookies = Some(jar)
} else {
self.cookies
.as_mut()
.unwrap()
.add(cookie.into_owned());
self.cookies.as_mut().unwrap().add(cookie.into_owned());
}
self
}
@ -534,9 +531,7 @@ impl HttpResponseBuilder {
if let Some(e) = self.err.take() {
return Error::from(e).into();
}
let mut response = self.response
.take()
.expect("cannot reuse response builder");
let mut response = self.response.take().expect("cannot reuse response builder");
if let Some(ref jar) = self.cookies {
for cookie in jar.delta() {
match HeaderValue::from_str(&cookie.to_string()) {
@ -558,9 +553,7 @@ impl HttpResponseBuilder {
S: Stream<Item = Bytes, Error = E> + 'static,
E: Into<Error>,
{
self.body(Body::Streaming(Box::new(
stream.map_err(|e| e.into()),
)))
self.body(Body::Streaming(Box::new(stream.map_err(|e| e.into()))))
}
/// Set a json body and generate `HttpResponse`
@ -607,7 +600,8 @@ impl HttpResponseBuilder {
#[inline]
#[cfg_attr(feature = "cargo-clippy", allow(borrowed_box))]
fn parts<'a>(
parts: &'a mut Option<Box<InnerHttpResponse>>, err: &Option<HttpError>,
parts: &'a mut Option<Box<InnerHttpResponse>>,
err: &Option<HttpError>,
) -> Option<&'a mut Box<InnerHttpResponse>> {
if err.is_some() {
return None;
@ -822,14 +816,15 @@ thread_local!(static POOL: Rc<UnsafeCell<HttpResponsePool>> = HttpResponsePool::
impl HttpResponsePool {
pub fn pool() -> Rc<UnsafeCell<HttpResponsePool>> {
Rc::new(UnsafeCell::new(HttpResponsePool(
VecDeque::with_capacity(128),
)))
Rc::new(UnsafeCell::new(HttpResponsePool(VecDeque::with_capacity(
128,
))))
}
#[inline]
pub fn get_builder(
pool: &Rc<UnsafeCell<HttpResponsePool>>, status: StatusCode,
pool: &Rc<UnsafeCell<HttpResponsePool>>,
status: StatusCode,
) -> HttpResponseBuilder {
let p = unsafe { &mut *pool.as_ref().get() };
if let Some(mut msg) = p.0.pop_front() {
@ -853,7 +848,9 @@ impl HttpResponsePool {
#[inline]
pub fn get_response(
pool: &Rc<UnsafeCell<HttpResponsePool>>, status: StatusCode, body: Body,
pool: &Rc<UnsafeCell<HttpResponsePool>>,
status: StatusCode,
body: Body,
) -> HttpResponse {
let p = unsafe { &mut *pool.as_ref().get() };
if let Some(mut msg) = p.0.pop_front() {
@ -879,7 +876,8 @@ impl HttpResponsePool {
#[inline(always)]
#[cfg_attr(feature = "cargo-clippy", allow(boxed_local, inline_always))]
fn release(
pool: &Rc<UnsafeCell<HttpResponsePool>>, mut inner: Box<InnerHttpResponse>,
pool: &Rc<UnsafeCell<HttpResponsePool>>,
mut inner: Box<InnerHttpResponse>,
) {
let pool = unsafe { &mut *pool.as_ref().get() };
if pool.0.len() < 128 {
@ -975,9 +973,7 @@ mod tests {
#[test]
fn test_force_close() {
let resp = HttpResponse::build(StatusCode::OK)
.force_close()
.finish();
let resp = HttpResponse::build(StatusCode::OK).force_close().finish();
assert!(!resp.keep_alive().unwrap())
}
@ -986,10 +982,7 @@ mod tests {
let resp = HttpResponse::build(StatusCode::OK)
.content_type("text/plain")
.body(Body::Empty);
assert_eq!(
resp.headers().get(CONTENT_TYPE).unwrap(),
"text/plain"
)
assert_eq!(resp.headers().get(CONTENT_TYPE).unwrap(), "text/plain")
}
#[test]
@ -1036,10 +1029,10 @@ mod tests {
}
impl Body {
pub(crate) fn binary(&self) -> Option<&Binary> {
pub(crate) fn bin_ref(&self) -> &Binary {
match *self {
Body::Binary(ref bin) => Some(bin),
_ => None,
Body::Binary(ref bin) => bin,
_ => panic!(),
}
}
}
@ -1055,7 +1048,7 @@ mod tests {
HeaderValue::from_static("text/plain; charset=utf-8")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(resp.body().binary().unwrap(), &Binary::from("test"));
assert_eq!(resp.body().bin_ref(), &Binary::from("test"));
let resp: HttpResponse = "test".respond_to(&req).ok().unwrap();
assert_eq!(resp.status(), StatusCode::OK);
@ -1064,7 +1057,7 @@ mod tests {
HeaderValue::from_static("text/plain; charset=utf-8")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(resp.body().binary().unwrap(), &Binary::from("test"));
assert_eq!(resp.body().bin_ref(), &Binary::from("test"));
let resp: HttpResponse = b"test".as_ref().into();
assert_eq!(resp.status(), StatusCode::OK);
@ -1073,10 +1066,7 @@ mod tests {
HeaderValue::from_static("application/octet-stream")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from(b"test".as_ref())
);
assert_eq!(resp.body().bin_ref(), &Binary::from(b"test".as_ref()));
let resp: HttpResponse = b"test".as_ref().respond_to(&req).ok().unwrap();
assert_eq!(resp.status(), StatusCode::OK);
@ -1085,10 +1075,7 @@ mod tests {
HeaderValue::from_static("application/octet-stream")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from(b"test".as_ref())
);
assert_eq!(resp.body().bin_ref(), &Binary::from(b"test".as_ref()));
let resp: HttpResponse = "test".to_owned().into();
assert_eq!(resp.status(), StatusCode::OK);
@ -1097,10 +1084,7 @@ mod tests {
HeaderValue::from_static("text/plain; charset=utf-8")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from("test".to_owned())
);
assert_eq!(resp.body().bin_ref(), &Binary::from("test".to_owned()));
let resp: HttpResponse = "test".to_owned().respond_to(&req).ok().unwrap();
assert_eq!(resp.status(), StatusCode::OK);
@ -1109,10 +1093,7 @@ mod tests {
HeaderValue::from_static("text/plain; charset=utf-8")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from("test".to_owned())
);
assert_eq!(resp.body().bin_ref(), &Binary::from("test".to_owned()));
let resp: HttpResponse = (&"test".to_owned()).into();
assert_eq!(resp.status(), StatusCode::OK);
@ -1121,10 +1102,7 @@ mod tests {
HeaderValue::from_static("text/plain; charset=utf-8")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from(&"test".to_owned())
);
assert_eq!(resp.body().bin_ref(), &Binary::from(&"test".to_owned()));
let resp: HttpResponse = (&"test".to_owned()).respond_to(&req).ok().unwrap();
assert_eq!(resp.status(), StatusCode::OK);
@ -1133,10 +1111,7 @@ mod tests {
HeaderValue::from_static("text/plain; charset=utf-8")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from(&"test".to_owned())
);
assert_eq!(resp.body().bin_ref(), &Binary::from(&"test".to_owned()));
let b = Bytes::from_static(b"test");
let resp: HttpResponse = b.into();
@ -1147,7 +1122,7 @@ mod tests {
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
resp.body().bin_ref(),
&Binary::from(Bytes::from_static(b"test"))
);
@ -1160,7 +1135,7 @@ mod tests {
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
resp.body().bin_ref(),
&Binary::from(Bytes::from_static(b"test"))
);
@ -1172,10 +1147,7 @@ mod tests {
HeaderValue::from_static("application/octet-stream")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from(BytesMut::from("test"))
);
assert_eq!(resp.body().bin_ref(), &Binary::from(BytesMut::from("test")));
let b = BytesMut::from("test");
let resp: HttpResponse = b.respond_to(&req).ok().unwrap();
@ -1185,10 +1157,7 @@ mod tests {
HeaderValue::from_static("application/octet-stream")
);
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.body().binary().unwrap(),
&Binary::from(BytesMut::from("test"))
);
assert_eq!(resp.body().bin_ref(), &Binary::from(BytesMut::from("test")));
}
#[test]

24
src/lib.rs
View file

@ -60,7 +60,21 @@
//! * Middlewares (`Logger`, `Session`, `CORS`, `CSRF`, `DefaultHeaders`)
//! * Built on top of [Actix actor framework](https://github.com/actix/actix)
//! * Supported Rust version: 1.24 or later
//!
//! ## Package feature
//!
//! * `tls` - enables ssl support via `native-tls` crate
//! * `alpn` - enables ssl support via `openssl` crate, require for `http/2`
//! support
//! * `session` - enables session support, includes `ring` crate as
//! dependency
//! * `brotli` - enables `brotli` compression support, requires `c`
//! compiler
//! * `flate-c` - enables `gzip`, `deflate` compression support, requires
//! `c` compiler
//! * `flate-rust` - experimental rust based implementation for
//! `gzip`, `deflate` compression.
//!
#![cfg_attr(actix_nightly, feature(
specialization, // for impl ErrorResponse for std::error::Error
))]
@ -169,12 +183,17 @@ pub use body::{Binary, Body};
pub use context::HttpContext;
pub use error::{Error, ResponseError, Result};
pub use extractor::{Form, Path, Query};
pub use handler::{AsyncResponder, Either, FromRequest, FutureResponse, Responder, State};
pub use handler::{
AsyncResponder, Either, FromRequest, FutureResponse, Responder, State,
};
pub use httpmessage::HttpMessage;
pub use httprequest::HttpRequest;
pub use httpresponse::HttpResponse;
pub use json::Json;
pub use scope::Scope;
#[doc(hidden)]
#[deprecated(since = "0.6.2", note = "please use `use actix_web::ws::WsWriter`")]
pub use ws::WsWriter;
#[cfg(feature = "openssl")]
@ -210,6 +229,7 @@ pub mod dev {
pub use resource::ResourceHandler;
pub use route::Route;
pub use router::{Resource, ResourceType, Router};
pub use with::ExtractorConfig;
}
pub mod http {

35
src/middleware/logger.rs
View file

@ -1,7 +1,7 @@
//! Request logging middleware
use std::collections::HashSet;
use std::env;
use std::fmt;
use std::fmt::{Display, Formatter};
use std::fmt::{self, Display, Formatter};
use libc;
use regex::Regex;
@ -74,6 +74,7 @@ use middleware::{Finished, Middleware, Started};
///
pub struct Logger {
format: Format,
exclude: HashSet<String>,
}
impl Logger {
@ -81,8 +82,15 @@ impl Logger {
pub fn new(format: &str) -> Logger {
Logger {
format: Format::new(format),
exclude: HashSet::new(),
}
}
/// Ignore and do not log access info for specified path.
pub fn exclude<T: Into<String>>(mut self, path: T) -> Self {
self.exclude.insert(path.into());
self
}
}
impl Default for Logger {
@ -94,6 +102,7 @@ impl Default for Logger {
fn default() -> Logger {
Logger {
format: Format::default(),
exclude: HashSet::new(),
}
}
}
@ -102,21 +111,23 @@ struct StartTime(time::Tm);
impl Logger {
fn log<S>(&self, req: &mut HttpRequest<S>, resp: &HttpResponse) {
let entry_time = req.extensions().get::<StartTime>().unwrap().0;
let render = |fmt: &mut Formatter| {
for unit in &self.format.0 {
unit.render(fmt, req, resp, entry_time)?;
}
Ok(())
};
info!("{}", FormatDisplay(&render));
if let Some(entry_time) = req.extensions().get::<StartTime>() {
let render = |fmt: &mut Formatter| {
for unit in &self.format.0 {
unit.render(fmt, req, resp, entry_time.0)?;
}
Ok(())
};
info!("{}", FormatDisplay(&render));
}
}
}
impl<S> Middleware<S> for Logger {
fn start(&self, req: &mut HttpRequest<S>) -> Result<Started> {
req.extensions_mut().insert(StartTime(time::now()));
if !self.exclude.contains(req.path()) {
req.extensions_mut().insert(StartTime(time::now()));
}
Ok(Started::Done)
}

76
src/middleware/session.rs
View file

@ -80,6 +80,7 @@ use serde_json::error::Error as JsonError;
use time::Duration;
use error::{Error, ResponseError, Result};
use handler::FromRequest;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Middleware, Response, Started};
@ -190,6 +191,34 @@ impl Session {
}
}
/// Extractor implementation for Session type.
///
/// ```rust
/// # use actix_web::*;
/// use actix_web::middleware::session::Session;
///
/// fn index(session: Session) -> Result<&'static str> {
/// // access session data
/// if let Some(count) = session.get::<i32>("counter")? {
/// session.set("counter", count+1)?;
/// } else {
/// session.set("counter", 1)?;
/// }
///
/// Ok("Welcome!")
/// }
/// # fn main() {}
/// ```
impl<S> FromRequest<S> for Session {
type Config = ();
type Result = Session;
#[inline]
fn from_request(req: &HttpRequest<S>, _: &Self::Config) -> Self::Result {
req.session()
}
}
struct SessionImplCell(RefCell<Box<SessionImpl>>);
#[doc(hidden)]
@ -226,22 +255,21 @@ impl<S: 'static, T: SessionBackend<S>> Middleware<S> for SessionStorage<T, S> {
fn start(&self, req: &mut HttpRequest<S>) -> Result<Started> {
let mut req = req.clone();
let fut = self.0
.from_request(&mut req)
.then(move |res| match res {
Ok(sess) => {
req.extensions_mut().insert(Arc::new(SessionImplCell(
RefCell::new(Box::new(sess)),
)));
FutOk(None)
}
Err(err) => FutErr(err),
});
let fut = self.0.from_request(&mut req).then(move |res| match res {
Ok(sess) => {
req.extensions_mut()
.insert(Arc::new(SessionImplCell(RefCell::new(Box::new(sess)))));
FutOk(None)
}
Err(err) => FutErr(err),
});
Ok(Started::Future(Box::new(fut)))
}
fn response(
&self, req: &mut HttpRequest<S>, resp: HttpResponse,
&self,
req: &mut HttpRequest<S>,
resp: HttpResponse,
) -> Result<Response> {
if let Some(s_box) = req.extensions_mut().remove::<Arc<SessionImplCell>>() {
s_box.0.borrow_mut().write(resp)
@ -357,7 +385,9 @@ impl CookieSessionInner {
}
fn set_cookie(
&self, resp: &mut HttpResponse, state: &HashMap<String, String>,
&self,
resp: &mut HttpResponse,
state: &HashMap<String, String>,
) -> Result<()> {
let value =
serde_json::to_string(&state).map_err(CookieSessionError::Serialize)?;
@ -551,4 +581,24 @@ mod tests {
let response = srv.execute(request.send()).unwrap();
assert!(response.cookie("actix-session").is_some());
}
#[test]
fn cookie_session_extractor() {
let mut srv = test::TestServer::with_factory(|| {
App::new()
.middleware(SessionStorage::new(
CookieSessionBackend::signed(&[0; 32]).secure(false),
))
.resource("/", |r| {
r.with(|ses: Session| {
let _ = ses.set("counter", 100);
"test"
})
})
});
let request = srv.get().uri(srv.url("/")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.cookie("actix-session").is_some());
}
}

69
src/pipeline.rs
View file

@ -29,7 +29,9 @@ pub(crate) trait PipelineHandler<S> {
fn encoding(&self) -> ContentEncoding;
fn handle(
&mut self, req: HttpRequest<S>, htype: HandlerType,
&mut self,
req: HttpRequest<S>,
htype: HandlerType,
) -> AsyncResult<HttpResponse>;
}
@ -120,8 +122,10 @@ impl<S> PipelineInfo<S> {
impl<S: 'static, H: PipelineHandler<S>> Pipeline<S, H> {
pub fn new(
req: HttpRequest<S>, mws: Rc<Vec<Box<Middleware<S>>>>,
handler: Rc<UnsafeCell<H>>, htype: HandlerType,
req: HttpRequest<S>,
mws: Rc<Vec<Box<Middleware<S>>>>,
handler: Rc<UnsafeCell<H>>,
htype: HandlerType,
) -> Pipeline<S, H> {
let mut info = PipelineInfo {
mws,
@ -148,6 +152,7 @@ impl Pipeline<(), Inner<()>> {
}
impl<S: 'static, H> Pipeline<S, H> {
#[inline]
fn is_done(&self) -> bool {
match self.1 {
PipelineState::None
@ -192,7 +197,9 @@ impl<S: 'static, H: PipelineHandler<S>> HttpHandlerTask for Pipeline<S, H> {
match self.1 {
PipelineState::None => return Ok(Async::Ready(true)),
PipelineState::Error => {
return Err(io::Error::new(io::ErrorKind::Other, "Internal error").into())
return Err(
io::Error::new(io::ErrorKind::Other, "Internal error").into()
)
}
_ => (),
}
@ -236,7 +243,9 @@ struct StartMiddlewares<S, H> {
impl<S: 'static, H: PipelineHandler<S>> StartMiddlewares<S, H> {
fn init(
info: &mut PipelineInfo<S>, hnd: Rc<UnsafeCell<H>>, htype: HandlerType,
info: &mut PipelineInfo<S>,
hnd: Rc<UnsafeCell<H>>,
htype: HandlerType,
) -> PipelineState<S, H> {
// execute middlewares, we need this stage because middlewares could be
// non-async and we can move to next state immediately
@ -251,23 +260,14 @@ impl<S: 'static, H: PipelineHandler<S>> StartMiddlewares<S, H> {
Ok(Started::Response(resp)) => {
return RunMiddlewares::init(info, resp)
}
Ok(Started::Future(mut fut)) => match fut.poll() {
Ok(Async::NotReady) => {
return PipelineState::Starting(StartMiddlewares {
hnd,
htype,
fut: Some(fut),
_s: PhantomData,
})
}
Ok(Async::Ready(resp)) => {
if let Some(resp) = resp {
return RunMiddlewares::init(info, resp);
}
info.count += 1;
}
Err(err) => return ProcessResponse::init(err.into()),
},
Ok(Started::Future(fut)) => {
return PipelineState::Starting(StartMiddlewares {
hnd,
htype,
fut: Some(fut),
_s: PhantomData,
})
}
Err(err) => return ProcessResponse::init(err.into()),
}
}
@ -284,12 +284,12 @@ impl<S: 'static, H: PipelineHandler<S>> StartMiddlewares<S, H> {
if let Some(resp) = resp {
return Some(RunMiddlewares::init(info, resp));
}
if info.count == len {
let reply = unsafe { &mut *self.hnd.get() }
.handle(info.req().clone(), self.htype);
return Some(WaitingResponse::init(info, reply));
} else {
loop {
loop {
if info.count == len {
let reply = unsafe { &mut *self.hnd.get() }
.handle(info.req().clone(), self.htype);
return Some(WaitingResponse::init(info, reply));
} else {
match info.mws[info.count as usize].start(info.req_mut()) {
Ok(Started::Done) => info.count += 1,
Ok(Started::Response(resp)) => {
@ -322,7 +322,8 @@ struct WaitingResponse<S, H> {
impl<S: 'static, H> WaitingResponse<S, H> {
#[inline]
fn init(
info: &mut PipelineInfo<S>, reply: AsyncResult<HttpResponse>,
info: &mut PipelineInfo<S>,
reply: AsyncResult<HttpResponse>,
) -> PipelineState<S, H> {
match reply.into() {
AsyncResultItem::Err(err) => RunMiddlewares::init(info, err.into()),
@ -353,6 +354,7 @@ struct RunMiddlewares<S, H> {
}
impl<S: 'static, H> RunMiddlewares<S, H> {
#[inline]
fn init(info: &mut PipelineInfo<S>, mut resp: HttpResponse) -> PipelineState<S, H> {
if info.count == 0 {
return ProcessResponse::init(resp);
@ -473,7 +475,9 @@ impl<S: 'static, H> ProcessResponse<S, H> {
}
fn poll_io(
mut self, io: &mut Writer, info: &mut PipelineInfo<S>,
mut self,
io: &mut Writer,
info: &mut PipelineInfo<S>,
) -> Result<PipelineState<S, H>, PipelineState<S, H>> {
loop {
if self.drain.is_none() && self.running != RunningState::Paused {
@ -501,8 +505,8 @@ impl<S: 'static, H> ProcessResponse<S, H> {
if let Some(err) = self.resp.error() {
if self.resp.status().is_server_error() {
error!(
"Error occured during request handling: {}",
err
"Error occured during request handling, status: {} {}",
self.resp.status(), err
);
} else {
warn!(
@ -685,6 +689,7 @@ struct FinishingMiddlewares<S, H> {
}
impl<S: 'static, H> FinishingMiddlewares<S, H> {
#[inline]
fn init(info: &mut PipelineInfo<S>, resp: HttpResponse) -> PipelineState<S, H> {
if info.count == 0 {
Completed::init(info)

21
src/resource.rs
View file

@ -1,9 +1,11 @@
use std::marker::PhantomData;
use std::rc::Rc;
use futures::Future;
use http::{Method, StatusCode};
use smallvec::SmallVec;
use error::Error;
use handler::{AsyncResult, FromRequest, Handler, Responder};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
@ -183,6 +185,25 @@ impl<S: 'static> ResourceHandler<S> {
self.routes.last_mut().unwrap().with(handler);
}
/// Register a new route and add async handler.
///
/// This is shortcut for:
///
/// ```rust,ignore
/// Application::resource("/", |r| r.route().with_async(index)
/// ```
pub fn with_async<T, F, R, I, E>(&mut self, handler: F)
where
F: Fn(T) -> R + 'static,
R: Future<Item = I, Error = E> + 'static,
I: Responder + 'static,
E: Into<Error> + 'static,
T: FromRequest<S> + 'static,
{
self.routes.push(Route::default());
self.routes.last_mut().unwrap().with_async(handler);
}
/// Register a resource middleware
///
/// This is similar to `App's` middlewares, but

142
src/route.rs
View file

@ -5,15 +5,19 @@ use std::rc::Rc;
use futures::{Async, Future, Poll};
use error::Error;
use handler::{AsyncHandler, AsyncResult, AsyncResultItem, FromRequest, Handler,
Responder, RouteHandler, WrapHandler};
use handler::{
AsyncHandler, AsyncResult, AsyncResultItem, FromRequest, Handler, Responder,
RouteHandler, WrapHandler,
};
use http::StatusCode;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Finished as MiddlewareFinished, Middleware,
Response as MiddlewareResponse, Started as MiddlewareStarted};
use middleware::{
Finished as MiddlewareFinished, Middleware, Response as MiddlewareResponse,
Started as MiddlewareStarted,
};
use pred::Predicate;
use with::{ExtractorConfig, With, With2, With3};
use with::{ExtractorConfig, With, With2, With3, WithAsync};
/// Resource route definition
///
@ -51,7 +55,9 @@ impl<S: 'static> Route<S> {
#[inline]
pub(crate) fn compose(
&mut self, req: HttpRequest<S>, mws: Rc<Vec<Box<Middleware<S>>>>,
&mut self,
req: HttpRequest<S>,
mws: Rc<Vec<Box<Middleware<S>>>>,
) -> AsyncResult<HttpResponse> {
AsyncResult::async(Box::new(Compose::new(req, mws, self.handler.clone())))
}
@ -129,6 +135,34 @@ impl<S: 'static> Route<S> {
/// |r| r.method(http::Method::GET).with(index)); // <- use `with` extractor
/// }
/// ```
///
/// It is possible to use tuples for specifing multiple extractors for one
/// handler function.
///
/// ```rust
/// # extern crate bytes;
/// # extern crate actix_web;
/// # extern crate futures;
/// #[macro_use] extern crate serde_derive;
/// # use std::collections::HashMap;
/// use actix_web::{http, App, Query, Path, Result, Json};
///
/// #[derive(Deserialize)]
/// struct Info {
/// username: String,
/// }
///
/// /// extract path info using serde
/// fn index(info: (Path<Info>, Query<HashMap<String, String>>, Json<Info>)) -> Result<String> {
/// Ok(format!("Welcome {}!", info.0.username))
/// }
///
/// fn main() {
/// let app = App::new().resource(
/// "/{username}/index.html", // <- define path parameters
/// |r| r.method(http::Method::GET).with(index)); // <- use `with` extractor
/// }
/// ```
pub fn with<T, F, R>(&mut self, handler: F) -> ExtractorConfig<S, T>
where
F: Fn(T) -> R + 'static,
@ -140,6 +174,49 @@ impl<S: 'static> Route<S> {
cfg
}
/// Set async handler function, use request extractor for parameters.
/// Also this method needs to be used if your handler function returns
/// `impl Future<>`
///
/// ```rust
/// # extern crate bytes;
/// # extern crate actix_web;
/// # extern crate futures;
/// #[macro_use] extern crate serde_derive;
/// use actix_web::{App, Path, Error, http};
/// use futures::Future;
///
/// #[derive(Deserialize)]
/// struct Info {
/// username: String,
/// }
///
/// /// extract path info using serde
/// fn index(info: Path<Info>) -> Box<Future<Item=&'static str, Error=Error>> {
/// unimplemented!()
/// }
///
/// fn main() {
/// let app = App::new().resource(
/// "/{username}/index.html", // <- define path parameters
/// |r| r.method(http::Method::GET)
/// .with_async(index)); // <- use `with` extractor
/// }
/// ```
pub fn with_async<T, F, R, I, E>(&mut self, handler: F) -> ExtractorConfig<S, T>
where
F: Fn(T) -> R + 'static,
R: Future<Item = I, Error = E> + 'static,
I: Responder + 'static,
E: Into<Error> + 'static,
T: FromRequest<S> + 'static,
{
let cfg = ExtractorConfig::default();
self.h(WithAsync::new(handler, Clone::clone(&cfg)));
cfg
}
#[doc(hidden)]
/// Set handler function, use request extractor for both parameters.
///
/// ```rust
@ -171,7 +248,8 @@ impl<S: 'static> Route<S> {
/// }
/// ```
pub fn with2<T1, T2, F, R>(
&mut self, handler: F,
&mut self,
handler: F,
) -> (ExtractorConfig<S, T1>, ExtractorConfig<S, T2>)
where
F: Fn(T1, T2) -> R + 'static,
@ -189,9 +267,11 @@ impl<S: 'static> Route<S> {
(cfg1, cfg2)
}
#[doc(hidden)]
/// Set handler function, use request extractor for all parameters.
pub fn with3<T1, T2, T3, F, R>(
&mut self, handler: F,
&mut self,
handler: F,
) -> (
ExtractorConfig<S, T1>,
ExtractorConfig<S, T2>,
@ -224,9 +304,7 @@ struct InnerHandler<S>(Rc<UnsafeCell<Box<RouteHandler<S>>>>);
impl<S: 'static> InnerHandler<S> {
#[inline]
fn new<H: Handler<S>>(h: H) -> Self {
InnerHandler(Rc::new(UnsafeCell::new(Box::new(WrapHandler::new(
h,
)))))
InnerHandler(Rc::new(UnsafeCell::new(Box::new(WrapHandler::new(h)))))
}
#[inline]
@ -237,9 +315,7 @@ impl<S: 'static> InnerHandler<S> {
R: Responder + 'static,
E: Into<Error> + 'static,
{
InnerHandler(Rc::new(UnsafeCell::new(Box::new(AsyncHandler::new(
h,
)))))
InnerHandler(Rc::new(UnsafeCell::new(Box::new(AsyncHandler::new(h)))))
}
#[inline]
@ -292,7 +368,9 @@ impl<S: 'static> ComposeState<S> {
impl<S: 'static> Compose<S> {
fn new(
req: HttpRequest<S>, mws: Rc<Vec<Box<Middleware<S>>>>, handler: InnerHandler<S>,
req: HttpRequest<S>,
mws: Rc<Vec<Box<Middleware<S>>>>,
handler: InnerHandler<S>,
) -> Self {
let mut info = ComposeInfo {
count: 0,
@ -346,21 +424,12 @@ impl<S: 'static> StartMiddlewares<S> {
Ok(MiddlewareStarted::Response(resp)) => {
return RunMiddlewares::init(info, resp)
}
Ok(MiddlewareStarted::Future(mut fut)) => match fut.poll() {
Ok(Async::NotReady) => {
return ComposeState::Starting(StartMiddlewares {
fut: Some(fut),
_s: PhantomData,
})
}
Ok(Async::Ready(resp)) => {
if let Some(resp) = resp {
return RunMiddlewares::init(info, resp);
}
info.count += 1;
}
Err(err) => return FinishingMiddlewares::init(info, err.into()),
},
Ok(MiddlewareStarted::Future(fut)) => {
return ComposeState::Starting(StartMiddlewares {
fut: Some(fut),
_s: PhantomData,
})
}
Err(err) => return FinishingMiddlewares::init(info, err.into()),
}
}
@ -377,11 +446,11 @@ impl<S: 'static> StartMiddlewares<S> {
if let Some(resp) = resp {
return Some(RunMiddlewares::init(info, resp));
}
if info.count == len {
let reply = info.handler.handle(info.req.clone());
return Some(WaitingResponse::init(info, reply));
} else {
loop {
loop {
if info.count == len {
let reply = info.handler.handle(info.req.clone());
return Some(WaitingResponse::init(info, reply));
} else {
match info.mws[info.count].start(&mut info.req) {
Ok(MiddlewareStarted::Done) => info.count += 1,
Ok(MiddlewareStarted::Response(resp)) => {
@ -416,7 +485,8 @@ struct WaitingResponse<S> {
impl<S: 'static> WaitingResponse<S> {
#[inline]
fn init(
info: &mut ComposeInfo<S>, reply: AsyncResult<HttpResponse>,
info: &mut ComposeInfo<S>,
reply: AsyncResult<HttpResponse>,
) -> ComposeState<S> {
match reply.into() {
AsyncResultItem::Err(err) => RunMiddlewares::init(info, err.into()),

114
src/router.rs
View file

@ -142,7 +142,8 @@ enum PatternElement {
#[derive(Clone, Debug)]
enum PatternType {
Static(String),
Dynamic(Regex, Vec<String>),
Prefix(String),
Dynamic(Regex, Vec<String>, usize),
}
#[derive(Debug, Copy, Clone, PartialEq)]
@ -173,14 +174,23 @@ impl Resource {
///
/// Panics if path pattern is wrong.
pub fn new(name: &str, path: &str) -> Self {
Resource::with_prefix(name, path, "/")
Resource::with_prefix(name, path, "/", false)
}
/// Parse path pattern and create new `Resource` instance.
///
/// Use `prefix` type instead of `static`.
///
/// Panics if path regex pattern is wrong.
pub fn prefix(name: &str, path: &str) -> Self {
Resource::with_prefix(name, path, "/", true)
}
/// Construct external resource
///
/// Panics if path pattern is wrong.
pub fn external(name: &str, path: &str) -> Self {
let mut resource = Resource::with_prefix(name, path, "/");
let mut resource = Resource::with_prefix(name, path, "/", false);
resource.rtp = ResourceType::External;
resource
}
@ -197,8 +207,9 @@ impl Resource {
}
/// Parse path pattern and create new `Resource` instance with custom prefix
pub fn with_prefix(name: &str, path: &str, prefix: &str) -> Self {
let (pattern, elements, is_dynamic) = Resource::parse(path, prefix);
pub fn with_prefix(name: &str, path: &str, prefix: &str, for_prefix: bool) -> Self {
let (pattern, elements, is_dynamic, len) =
Resource::parse(path, prefix, for_prefix);
let tp = if is_dynamic {
let re = match Regex::new(&pattern) {
@ -208,7 +219,9 @@ impl Resource {
let names = re.capture_names()
.filter_map(|name| name.map(|name| name.to_owned()))
.collect();
PatternType::Dynamic(re, names)
PatternType::Dynamic(re, names, len)
} else if for_prefix {
PatternType::Prefix(pattern.clone())
} else {
PatternType::Static(pattern.clone())
};
@ -240,7 +253,8 @@ impl Resource {
pub fn is_match(&self, path: &str) -> bool {
match self.tp {
PatternType::Static(ref s) => s == path,
PatternType::Dynamic(ref re, _) => re.is_match(path),
PatternType::Dynamic(ref re, _, _) => re.is_match(path),
PatternType::Prefix(ref s) => path.starts_with(s),
}
}
@ -249,7 +263,7 @@ impl Resource {
) -> bool {
match self.tp {
PatternType::Static(ref s) => s == path,
PatternType::Dynamic(ref re, ref names) => {
PatternType::Dynamic(ref re, ref names, _) => {
if let Some(captures) = re.captures(path) {
let mut idx = 0;
for capture in captures.iter() {
@ -265,6 +279,42 @@ impl Resource {
false
}
}
PatternType::Prefix(ref s) => path.starts_with(s),
}
}
pub fn match_prefix_with_params<'a>(
&'a self, path: &'a str, params: &'a mut Params<'a>,
) -> Option<usize> {
match self.tp {
PatternType::Static(ref s) => if s == path {
Some(s.len())
} else {
None
},
PatternType::Dynamic(ref re, ref names, len) => {
if let Some(captures) = re.captures(path) {
let mut idx = 0;
let mut pos = 0;
for capture in captures.iter() {
if let Some(ref m) = capture {
if idx != 0 {
params.add(names[idx - 1].as_str(), m.as_str());
}
idx += 1;
pos = m.end();
}
}
Some(pos + len)
} else {
None
}
}
PatternType::Prefix(ref s) => if path.starts_with(s) {
Some(s.len())
} else {
None
},
}
}
@ -297,7 +347,9 @@ impl Resource {
Ok(path)
}
fn parse(pattern: &str, prefix: &str) -> (String, Vec<PatternElement>, bool) {
fn parse(
pattern: &str, prefix: &str, for_prefix: bool,
) -> (String, Vec<PatternElement>, bool, usize) {
const DEFAULT_PATTERN: &str = "[^/]+";
let mut re1 = String::from("^") + prefix;
@ -309,6 +361,7 @@ impl Resource {
let mut param_pattern = String::from(DEFAULT_PATTERN);
let mut is_dynamic = false;
let mut elems = Vec::new();
let mut len = 0;
for (index, ch) in pattern.chars().enumerate() {
// All routes must have a leading slash so its optional to have one
@ -325,6 +378,7 @@ impl Resource {
param_name.clear();
param_pattern = String::from(DEFAULT_PATTERN);
len = 0;
in_param_pattern = false;
in_param = false;
} else if ch == ':' {
@ -348,16 +402,19 @@ impl Resource {
re1.push_str(escape(&ch.to_string()).as_str());
re2.push(ch);
el.push(ch);
len += 1;
}
}
let re = if is_dynamic {
re1.push('$');
if !for_prefix {
re1.push('$');
}
re1
} else {
re2
};
(re, elems, is_dynamic)
(re, elems, is_dynamic, len)
}
}
@ -570,6 +627,41 @@ mod tests {
assert_eq!(req.match_info().get("id").unwrap(), "adahg32");
}
#[test]
fn test_resource_prefix() {
let re = Resource::prefix("test", "/name");
assert!(re.is_match("/name"));
assert!(re.is_match("/name/"));
assert!(re.is_match("/name/test/test"));
assert!(re.is_match("/name1"));
assert!(re.is_match("/name~"));
let re = Resource::prefix("test", "/name/");
assert!(re.is_match("/name/"));
assert!(re.is_match("/name/gs"));
assert!(!re.is_match("/name"));
}
#[test]
fn test_reousrce_prefix_dynamic() {
let re = Resource::prefix("test", "/{name}/");
assert!(re.is_match("/name/"));
assert!(re.is_match("/name/gs"));
assert!(!re.is_match("/name"));
let mut req = TestRequest::with_uri("/test2/").finish();
assert!(re.match_with_params("/test2/", req.match_info_mut()));
assert_eq!(&req.match_info()["name"], "test2");
let mut req =
TestRequest::with_uri("/test2/subpath1/subpath2/index.html").finish();
assert!(re.match_with_params(
"/test2/subpath1/subpath2/index.html",
req.match_info_mut()
));
assert_eq!(&req.match_info()["name"], "test2");
}
#[test]
fn test_request_resource() {
let routes = vec![

402
src/scope.rs
View file

@ -1,5 +1,6 @@
use std::cell::UnsafeCell;
use std::marker::PhantomData;
use std::mem;
use std::rc::Rc;
use futures::{Async, Future, Poll};
@ -9,19 +10,28 @@ use handler::{AsyncResult, AsyncResultItem, FromRequest, Responder, RouteHandler
use http::Method;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Finished as MiddlewareFinished, Middleware,
Response as MiddlewareResponse, Started as MiddlewareStarted};
use middleware::{
Finished as MiddlewareFinished, Middleware, Response as MiddlewareResponse,
Started as MiddlewareStarted,
};
use pred::Predicate;
use resource::ResourceHandler;
use router::Resource;
type Route<S> = UnsafeCell<Box<RouteHandler<S>>>;
type ScopeResources<S> = Rc<Vec<(Resource, Rc<UnsafeCell<ResourceHandler<S>>>)>>;
type NestedInfo<S> = (Resource, Route<S>, Vec<Box<Predicate<S>>>);
/// Resources scope
///
/// Scope is a set of resources with common root path.
/// Scopes collect multiple paths under a common path prefix.
/// Scope path can not contain variable path segments as resources.
/// Scope path can contain variable path segments as resources.
/// Scope prefix is always complete path segment, i.e `/app` would
/// be converted to a `/app/` and it would not match `/app` path.
///
/// You can get variable path segments from `HttpRequest::match_info()`.
/// `Path` extractor also is able to extract scope level variable segments.
///
/// ```rust
/// # extern crate actix_web;
@ -29,7 +39,7 @@ type ScopeResources<S> = Rc<Vec<(Resource, Rc<UnsafeCell<ResourceHandler<S>>>)>>
///
/// fn main() {
/// let app = App::new()
/// .scope("/app", |scope| {
/// .scope("/{project_id}/", |scope| {
/// scope.resource("/path1", |r| r.f(|_| HttpResponse::Ok()))
/// .resource("/path2", |r| r.f(|_| HttpResponse::Ok()))
/// .resource("/path3", |r| r.f(|_| HttpResponse::MethodNotAllowed()))
@ -38,26 +48,24 @@ type ScopeResources<S> = Rc<Vec<(Resource, Rc<UnsafeCell<ResourceHandler<S>>>)>>
/// ```
///
/// In the above example three routes get registered:
/// * /app/path1 - reponds to all http method
/// * /app/path2 - `GET` requests
/// * /app/path3 - `HEAD` requests
/// * /{project_id}/path1 - reponds to all http method
/// * /{project_id}/path2 - `GET` requests
/// * /{project_id}/path3 - `HEAD` requests
///
#[derive(Default)]
pub struct Scope<S: 'static> {
nested: Vec<(String, Route<S>)>,
filters: Vec<Box<Predicate<S>>>,
nested: Vec<NestedInfo<S>>,
middlewares: Rc<Vec<Box<Middleware<S>>>>,
default: Rc<UnsafeCell<ResourceHandler<S>>>,
resources: ScopeResources<S>,
}
impl<S: 'static> Default for Scope<S> {
fn default() -> Scope<S> {
Scope::new()
}
}
#[cfg_attr(feature = "cargo-clippy", allow(new_without_default_derive))]
impl<S: 'static> Scope<S> {
pub fn new() -> Scope<S> {
Scope {
filters: Vec::new(),
nested: Vec::new(),
resources: Rc::new(Vec::new()),
middlewares: Rc::new(Vec::new()),
@ -65,6 +73,36 @@ impl<S: 'static> Scope<S> {
}
}
#[inline]
pub(crate) fn take_filters(&mut self) -> Vec<Box<Predicate<S>>> {
mem::replace(&mut self.filters, Vec::new())
}
/// Add match predicate to scope.
///
/// ```rust
/// # extern crate actix_web;
/// use actix_web::{http, pred, App, HttpRequest, HttpResponse, Path};
///
/// fn index(data: Path<(String, String)>) -> &'static str {
/// "Welcome!"
/// }
///
/// fn main() {
/// let app = App::new()
/// .scope("/app", |scope| {
/// scope.filter(pred::Header("content-type", "text/plain"))
/// .route("/test1", http::Method::GET, index)
/// .route("/test2", http::Method::POST,
/// |_: HttpRequest| HttpResponse::MethodNotAllowed())
/// });
/// }
/// ```
pub fn filter<T: Predicate<S> + 'static>(mut self, p: T) -> Self {
self.filters.push(Box::new(p));
self
}
/// Create nested scope with new state.
///
/// ```rust
@ -91,23 +129,30 @@ impl<S: 'static> Scope<S> {
F: FnOnce(Scope<T>) -> Scope<T>,
{
let scope = Scope {
filters: Vec::new(),
nested: Vec::new(),
resources: Rc::new(Vec::new()),
middlewares: Rc::new(Vec::new()),
default: Rc::new(UnsafeCell::new(ResourceHandler::default_not_found())),
};
let scope = f(scope);
let mut scope = f(scope);
let mut path = path.trim().trim_right_matches('/').to_owned();
if !path.is_empty() && !path.starts_with('/') {
path.insert(0, '/')
}
if !path.ends_with('/') {
path.push('/');
}
let handler = UnsafeCell::new(Box::new(Wrapper {
scope,
state: Rc::new(state),
}));
self.nested.push((path, handler));
let state = Rc::new(state);
let filters: Vec<Box<Predicate<S>>> = vec![Box::new(FiltersWrapper {
state: Rc::clone(&state),
filters: scope.take_filters(),
})];
let handler = UnsafeCell::new(Box::new(Wrapper { scope, state }));
self.nested
.push((Resource::prefix("", &path), handler, filters));
self
}
@ -138,20 +183,28 @@ impl<S: 'static> Scope<S> {
F: FnOnce(Scope<S>) -> Scope<S>,
{
let scope = Scope {
filters: Vec::new(),
nested: Vec::new(),
resources: Rc::new(Vec::new()),
middlewares: Rc::new(Vec::new()),
default: Rc::new(UnsafeCell::new(ResourceHandler::default_not_found())),
};
let scope = f(scope);
let mut scope = f(scope);
let mut path = path.trim().trim_right_matches('/').to_owned();
if !path.is_empty() && !path.starts_with('/') {
path.insert(0, '/')
}
if !path.ends_with('/') {
path.push('/');
}
self.nested
.push((path, UnsafeCell::new(Box::new(scope))));
let filters = scope.take_filters();
self.nested.push((
Resource::prefix("", &path),
UnsafeCell::new(Box::new(scope)),
filters,
));
self
}
@ -264,7 +317,7 @@ impl<S: 'static> Scope<S> {
/// middlewares get invoked on scope level.
///
/// *Note* `Middleware::finish()` fires right after response get
/// prepared. It does not wait until body get sent to peer.
/// prepared. It does not wait until body get sent to the peer.
pub fn middleware<M: Middleware<S>>(mut self, mw: M) -> Scope<S> {
Rc::get_mut(&mut self.middlewares)
.expect("Can not use after configuration")
@ -278,7 +331,7 @@ impl<S: 'static> RouteHandler<S> for Scope<S> {
let path = unsafe { &*(&req.match_info()["tail"] as *const _) };
let path = if path == "" { "/" } else { path };
// recognize paths
// recognize resources
for &(ref pattern, ref resource) in self.resources.iter() {
if pattern.match_with_params(path, req.match_info_mut()) {
let default = unsafe { &mut *self.default.as_ref().get() };
@ -298,17 +351,19 @@ impl<S: 'static> RouteHandler<S> for Scope<S> {
}
// nested scopes
for &(ref prefix, ref handler) in &self.nested {
let len = req.prefix_len() as usize;
let m = {
let path = &req.path()[len..];
path.starts_with(prefix)
&& (path.len() == prefix.len()
|| path.split_at(prefix.len()).1.starts_with('/'))
};
let len = req.prefix_len() as usize;
let path: &'static str = unsafe { &*(&req.path()[len..] as *const _) };
if m {
let prefix_len = len + prefix.len();
'outer: for &(ref prefix, ref handler, ref filters) in &self.nested {
if let Some(prefix_len) =
prefix.match_prefix_with_params(path, req.match_info_mut())
{
for filter in filters {
if !filter.check(&mut req) {
continue 'outer;
}
}
let prefix_len = len + prefix_len - 1;
let path: &'static str =
unsafe { &*(&req.path()[prefix_len..] as *const _) };
@ -347,8 +402,24 @@ struct Wrapper<S: 'static> {
impl<S: 'static, S2: 'static> RouteHandler<S2> for Wrapper<S> {
fn handle(&mut self, req: HttpRequest<S2>) -> AsyncResult<HttpResponse> {
self.scope
.handle(req.change_state(Rc::clone(&self.state)))
self.scope.handle(req.change_state(Rc::clone(&self.state)))
}
}
struct FiltersWrapper<S: 'static> {
state: Rc<S>,
filters: Vec<Box<Predicate<S>>>,
}
impl<S: 'static, S2: 'static> Predicate<S2> for FiltersWrapper<S> {
fn check(&self, req: &mut HttpRequest<S2>) -> bool {
let mut req = req.change_state(Rc::clone(&self.state));
for filter in &self.filters {
if !filter.check(&mut req) {
return false;
}
}
true
}
}
@ -388,7 +459,8 @@ impl<S: 'static> ComposeState<S> {
impl<S: 'static> Compose<S> {
fn new(
req: HttpRequest<S>, mws: Rc<Vec<Box<Middleware<S>>>>,
req: HttpRequest<S>,
mws: Rc<Vec<Box<Middleware<S>>>>,
resource: Rc<UnsafeCell<ResourceHandler<S>>>,
default: Option<Rc<UnsafeCell<ResourceHandler<S>>>>,
) -> Self {
@ -451,21 +523,12 @@ impl<S: 'static> StartMiddlewares<S> {
Ok(MiddlewareStarted::Response(resp)) => {
return RunMiddlewares::init(info, resp)
}
Ok(MiddlewareStarted::Future(mut fut)) => match fut.poll() {
Ok(Async::NotReady) => {
return ComposeState::Starting(StartMiddlewares {
fut: Some(fut),
_s: PhantomData,
})
}
Ok(Async::Ready(resp)) => {
if let Some(resp) = resp {
return RunMiddlewares::init(info, resp);
}
info.count += 1;
}
Err(err) => return Response::init(err.into()),
},
Ok(MiddlewareStarted::Future(fut)) => {
return ComposeState::Starting(StartMiddlewares {
fut: Some(fut),
_s: PhantomData,
})
}
Err(err) => return Response::init(err.into()),
}
}
@ -482,17 +545,17 @@ impl<S: 'static> StartMiddlewares<S> {
if let Some(resp) = resp {
return Some(RunMiddlewares::init(info, resp));
}
if info.count == len {
let resource = unsafe { &mut *info.resource.get() };
let reply = if let Some(ref default) = info.default {
let d = unsafe { &mut *default.as_ref().get() };
resource.handle(info.req.clone(), Some(d))
loop {
if info.count == len {
let resource = unsafe { &mut *info.resource.get() };
let reply = if let Some(ref default) = info.default {
let d = unsafe { &mut *default.as_ref().get() };
resource.handle(info.req.clone(), Some(d))
} else {
resource.handle(info.req.clone(), None)
};
return Some(WaitingResponse::init(info, reply));
} else {
resource.handle(info.req.clone(), None)
};
return Some(WaitingResponse::init(info, reply));
} else {
loop {
match info.mws[info.count].start(&mut info.req) {
Ok(MiddlewareStarted::Done) => info.count += 1,
Ok(MiddlewareStarted::Response(resp)) => {
@ -522,7 +585,8 @@ struct WaitingResponse<S> {
impl<S: 'static> WaitingResponse<S> {
#[inline]
fn init(
info: &mut ComposeInfo<S>, reply: AsyncResult<HttpResponse>,
info: &mut ComposeInfo<S>,
reply: AsyncResult<HttpResponse>,
) -> ComposeState<S> {
match reply.into() {
AsyncResultItem::Ok(resp) => RunMiddlewares::init(info, resp),
@ -698,9 +762,14 @@ impl<S: 'static> Response<S> {
#[cfg(test)]
mod tests {
use bytes::Bytes;
use application::App;
use http::StatusCode;
use body::Body;
use http::{Method, StatusCode};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use pred;
use test::TestRequest;
#[test]
@ -716,6 +785,90 @@ mod tests {
assert_eq!(resp.as_msg().status(), StatusCode::OK);
}
#[test]
fn test_scope_route() {
let mut app = App::new()
.scope("app", |scope| {
scope
.route("/path1", Method::GET, |_: HttpRequest<_>| {
HttpResponse::Ok()
})
.route("/path1", Method::DELETE, |_: HttpRequest<_>| {
HttpResponse::Ok()
})
})
.finish();
let req = TestRequest::with_uri("/app/path1").finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::OK);
let req = TestRequest::with_uri("/app/path1")
.method(Method::DELETE)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::OK);
let req = TestRequest::with_uri("/app/path1")
.method(Method::POST)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::NOT_FOUND);
}
#[test]
fn test_scope_filter() {
let mut app = App::new()
.scope("/app", |scope| {
scope
.filter(pred::Get())
.resource("/path1", |r| r.f(|_| HttpResponse::Ok()))
})
.finish();
let req = TestRequest::with_uri("/app/path1")
.method(Method::POST)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::NOT_FOUND);
let req = TestRequest::with_uri("/app/path1")
.method(Method::GET)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::OK);
}
#[test]
fn test_scope_variable_segment() {
let mut app = App::new()
.scope("/ab-{project}", |scope| {
scope.resource("/path1", |r| {
r.f(|r| {
HttpResponse::Ok()
.body(format!("project: {}", &r.match_info()["project"]))
})
})
})
.finish();
let req = TestRequest::with_uri("/ab-project1/path1").finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::OK);
match resp.as_msg().body() {
&Body::Binary(ref b) => {
let bytes: Bytes = b.clone().into();
assert_eq!(bytes, Bytes::from_static(b"project: project1"));
}
_ => panic!(),
}
let req = TestRequest::with_uri("/aa-project1/path1").finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::NOT_FOUND);
}
#[test]
fn test_scope_with_state() {
struct State;
@ -733,6 +886,33 @@ mod tests {
assert_eq!(resp.as_msg().status(), StatusCode::CREATED);
}
#[test]
fn test_scope_with_state_filter() {
struct State;
let mut app = App::new()
.scope("/app", |scope| {
scope.with_state("/t1", State, |scope| {
scope
.filter(pred::Get())
.resource("/path1", |r| r.f(|_| HttpResponse::Ok()))
})
})
.finish();
let req = TestRequest::with_uri("/app/t1/path1")
.method(Method::POST)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::NOT_FOUND);
let req = TestRequest::with_uri("/app/t1/path1")
.method(Method::GET)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::OK);
}
#[test]
fn test_nested_scope() {
let mut app = App::new()
@ -748,6 +928,98 @@ mod tests {
assert_eq!(resp.as_msg().status(), StatusCode::CREATED);
}
#[test]
fn test_nested_scope_filter() {
let mut app = App::new()
.scope("/app", |scope| {
scope.nested("/t1", |scope| {
scope
.filter(pred::Get())
.resource("/path1", |r| r.f(|_| HttpResponse::Ok()))
})
})
.finish();
let req = TestRequest::with_uri("/app/t1/path1")
.method(Method::POST)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::NOT_FOUND);
let req = TestRequest::with_uri("/app/t1/path1")
.method(Method::GET)
.finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::OK);
}
#[test]
fn test_nested_scope_with_variable_segment() {
let mut app = App::new()
.scope("/app", |scope| {
scope.nested("/{project_id}", |scope| {
scope.resource("/path1", |r| {
r.f(|r| {
HttpResponse::Created().body(format!(
"project: {}",
&r.match_info()["project_id"]
))
})
})
})
})
.finish();
let req = TestRequest::with_uri("/app/project_1/path1").finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::CREATED);
match resp.as_msg().body() {
&Body::Binary(ref b) => {
let bytes: Bytes = b.clone().into();
assert_eq!(bytes, Bytes::from_static(b"project: project_1"));
}
_ => panic!(),
}
}
#[test]
fn test_nested2_scope_with_variable_segment() {
let mut app = App::new()
.scope("/app", |scope| {
scope.nested("/{project}", |scope| {
scope.nested("/{id}", |scope| {
scope.resource("/path1", |r| {
r.f(|r| {
HttpResponse::Created().body(format!(
"project: {} - {}",
&r.match_info()["project"],
&r.match_info()["id"],
))
})
})
})
})
})
.finish();
let req = TestRequest::with_uri("/app/test/1/path1").finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::CREATED);
match resp.as_msg().body() {
&Body::Binary(ref b) => {
let bytes: Bytes = b.clone().into();
assert_eq!(bytes, Bytes::from_static(b"project: test - 1"));
}
_ => panic!(),
}
let req = TestRequest::with_uri("/app/test/1/path2").finish();
let resp = app.run(req);
assert_eq!(resp.as_msg().status(), StatusCode::NOT_FOUND);
}
#[test]
fn test_default_resource() {
let mut app = App::new()

23
src/server/channel.rs
View file

@ -38,7 +38,9 @@ where
H: HttpHandler + 'static,
{
pub(crate) fn new(
settings: Rc<WorkerSettings<H>>, mut io: T, peer: Option<SocketAddr>,
settings: Rc<WorkerSettings<H>>,
mut io: T,
peer: Option<SocketAddr>,
http2: bool,
) -> HttpChannel<T, H> {
settings.add_channel();
@ -61,7 +63,7 @@ where
settings,
peer,
io,
BytesMut::with_capacity(4096),
BytesMut::with_capacity(8192),
)),
}
}
@ -93,12 +95,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,9 +170,8 @@ 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 => {

89
src/server/encoding.rs
View file

@ -12,8 +12,10 @@ use flate2::read::GzDecoder;
use flate2::write::{DeflateDecoder, DeflateEncoder, GzEncoder};
#[cfg(feature = "flate2")]
use flate2::Compression;
use http::header::{HeaderMap, HeaderValue, ACCEPT_ENCODING, CONTENT_ENCODING,
CONTENT_LENGTH, TRANSFER_ENCODING};
use http::header::{
HeaderMap, HeaderValue, ACCEPT_ENCODING, CONTENT_ENCODING, CONTENT_LENGTH,
TRANSFER_ENCODING,
};
use http::{HttpTryFrom, Method, Version};
use body::{Binary, Body};
@ -378,16 +380,19 @@ impl ContentEncoder {
}
pub fn for_server(
buf: SharedBytes, req: &HttpInnerMessage, resp: &mut HttpResponse,
buf: SharedBytes,
req: &HttpInnerMessage,
resp: &mut HttpResponse,
response_encoding: ContentEncoding,
) -> ContentEncoder {
let version = resp.version().unwrap_or_else(|| req.version);
let is_head = req.method == Method::HEAD;
let mut body = resp.replace_body(Body::Empty);
let has_body = match body {
Body::Empty => false,
Body::Binary(ref bin) => {
!(response_encoding == ContentEncoding::Auto && bin.len() < 96)
let mut len = 0;
let has_body = match resp.body() {
&Body::Empty => false,
&Body::Binary(ref bin) => {
len = bin.len();
!(response_encoding == ContentEncoding::Auto && len < 96)
}
_ => true,
};
@ -421,14 +426,14 @@ impl ContentEncoder {
ContentEncoding::Identity
};
let mut transfer = match body {
Body::Empty => {
let mut transfer = match resp.body() {
&Body::Empty => {
if req.method != Method::HEAD {
resp.headers_mut().remove(CONTENT_LENGTH);
}
TransferEncoding::length(0, buf)
}
Body::Binary(ref mut bytes) => {
&Body::Binary(_) => {
if !(encoding == ContentEncoding::Identity
|| encoding == ContentEncoding::Auto)
{
@ -448,19 +453,26 @@ impl ContentEncoder {
ContentEncoding::Br => {
ContentEncoder::Br(BrotliEncoder::new(transfer, 3))
}
ContentEncoding::Identity => ContentEncoder::Identity(transfer),
ContentEncoding::Auto => unreachable!(),
ContentEncoding::Identity | ContentEncoding::Auto => {
unreachable!()
}
};
// TODO return error!
let _ = enc.write(bytes.clone());
let _ = enc.write_eof();
*bytes = Binary::from(tmp.take());
let bin = resp.replace_body(Body::Empty).binary();
// TODO return error!
let _ = enc.write(bin);
let _ = enc.write_eof();
let body = tmp.take();
len = body.len();
encoding = ContentEncoding::Identity;
resp.replace_body(Binary::from(body));
}
if is_head {
let mut b = BytesMut::new();
let _ = write!(b, "{}", bytes.len());
let _ = write!(b, "{}", len);
resp.headers_mut().insert(
CONTENT_LENGTH,
HeaderValue::try_from(b.freeze()).unwrap(),
@ -470,7 +482,7 @@ impl ContentEncoder {
}
TransferEncoding::eof(buf)
}
Body::Streaming(_) | Body::Actor(_) => {
&Body::Streaming(_) | &Body::Actor(_) => {
if resp.upgrade() {
if version == Version::HTTP_2 {
error!("Connection upgrade is forbidden for HTTP/2");
@ -485,11 +497,10 @@ impl ContentEncoder {
}
}
};
//
// check for head response
if is_head {
resp.set_body(Body::Empty);
transfer.kind = TransferEncodingKind::Length(0);
} else {
resp.replace_body(body);
}
match encoding {
@ -511,7 +522,9 @@ impl ContentEncoder {
}
fn streaming_encoding(
buf: SharedBytes, version: Version, resp: &mut HttpResponse,
buf: SharedBytes,
version: Version,
resp: &mut HttpResponse,
) -> TransferEncoding {
match resp.chunked() {
Some(true) => {
@ -590,7 +603,7 @@ impl ContentEncoder {
#[cfg_attr(feature = "cargo-clippy", allow(inline_always))]
#[inline(always)]
pub fn write_eof(&mut self) -> Result<(), io::Error> {
pub fn write_eof(&mut self) -> Result<bool, io::Error> {
let encoder = mem::replace(
self,
ContentEncoder::Identity(TransferEncoding::eof(SharedBytes::empty())),
@ -602,7 +615,7 @@ impl ContentEncoder {
Ok(mut writer) => {
writer.encode_eof();
*self = ContentEncoder::Identity(writer);
Ok(())
Ok(true)
}
Err(err) => Err(err),
},
@ -611,7 +624,7 @@ impl ContentEncoder {
Ok(mut writer) => {
writer.encode_eof();
*self = ContentEncoder::Identity(writer);
Ok(())
Ok(true)
}
Err(err) => Err(err),
},
@ -620,14 +633,14 @@ impl ContentEncoder {
Ok(mut writer) => {
writer.encode_eof();
*self = ContentEncoder::Identity(writer);
Ok(())
Ok(true)
}
Err(err) => Err(err),
},
ContentEncoder::Identity(mut writer) => {
writer.encode_eof();
let res = writer.encode_eof();
*self = ContentEncoder::Identity(writer);
Ok(())
Ok(res)
}
}
}
@ -763,8 +776,7 @@ 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)
@ -777,14 +789,16 @@ impl TransferEncoding {
/// Encode eof. Return `EOF` state of encoder
#[inline]
pub fn encode_eof(&mut self) {
pub fn encode_eof(&mut self) -> bool {
match self.kind {
TransferEncodingKind::Eof | TransferEncodingKind::Length(_) => (),
TransferEncodingKind::Eof => true,
TransferEncodingKind::Length(rem) => rem == 0,
TransferEncodingKind::Chunked(ref mut eof) => {
if !*eof {
*eof = true;
self.buffer.extend_from_slice(b"0\r\n\r\n");
}
true
}
}
}
@ -848,10 +862,7 @@ impl AcceptEncoding {
Err(_) => 0.0,
},
};
Some(AcceptEncoding {
encoding,
quality,
})
Some(AcceptEncoding { encoding, quality })
}
/// Parse a raw Accept-Encoding header value into an ordered list.
@ -879,9 +890,7 @@ 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(),

182
src/server/h1.rs
View file

@ -67,7 +67,9 @@ where
H: HttpHandler + 'static,
{
pub fn new(
settings: Rc<WorkerSettings<H>>, stream: T, addr: Option<SocketAddr>,
settings: Rc<WorkerSettings<H>>,
stream: T,
addr: Option<SocketAddr>,
buf: BytesMut,
) -> Self {
let bytes = settings.get_shared_bytes();
@ -146,10 +148,12 @@ where
}
#[inline]
/// read data from stream
pub fn poll_io(&mut self) {
// read io from socket
if !self.flags.intersects(Flags::ERROR)
&& self.tasks.len() < MAX_PIPELINED_MESSAGES && self.can_read()
&& self.tasks.len() < MAX_PIPELINED_MESSAGES
&& self.can_read()
{
if self.read() {
// notify all tasks
@ -158,7 +162,6 @@ where
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
@ -205,10 +208,9 @@ 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);
item.flags.insert(EntryFlags::EOF);
}
}
// no more IO for this iteration
@ -324,7 +326,36 @@ where
// search handler for request
for h in self.settings.handlers().iter_mut() {
req = match h.handle(req) {
Ok(pipe) => {
Ok(mut pipe) => {
if self.tasks.is_empty() {
match pipe.poll_io(&mut self.stream) {
Ok(Async::Ready(ready)) => {
// override keep-alive state
if self.stream.keepalive() {
self.flags.insert(Flags::KEEPALIVE);
} else {
self.flags.remove(Flags::KEEPALIVE);
}
// prepare stream for next response
self.stream.reset();
if !ready {
let item = Entry {
pipe,
flags: EntryFlags::EOF,
};
self.tasks.push_back(item);
}
continue 'outer;
}
Ok(Async::NotReady) => {}
Err(err) => {
error!("Unhandled error: {}", err);
self.flags.insert(Flags::ERROR);
return;
}
}
}
self.tasks.push_back(Entry {
pipe,
flags: EntryFlags::empty(),
@ -347,6 +378,7 @@ where
} else {
error!("Internal server error: unexpected payload chunk");
self.flags.insert(Flags::ERROR);
break;
}
}
Ok(Some(Message::Eof)) => {
@ -355,6 +387,7 @@ where
} else {
error!("Internal server error: unexpected eof");
self.flags.insert(Flags::ERROR);
break;
}
}
Ok(None) => break,
@ -367,6 +400,7 @@ where
};
payload.set_error(e);
}
break;
}
}
}
@ -398,8 +432,12 @@ where
#[cfg(test)]
mod tests {
use bytes::{Bytes, BytesMut};
use std::net::Shutdown;
use std::{cmp, time};
use bytes::{Buf, Bytes, BytesMut};
use http::{Method, Version};
use tokio_io::{AsyncRead, AsyncWrite};
use super::*;
use application::HttpApplication;
@ -463,6 +501,101 @@ mod tests {
}};
}
struct Buffer {
buf: Bytes,
err: Option<io::Error>,
}
impl Buffer {
fn new(data: &'static str) -> Buffer {
Buffer {
buf: Bytes::from(data),
err: None,
}
}
fn feed_data(&mut self, data: &'static str) {
let mut b = BytesMut::from(self.buf.as_ref());
b.extend(data.as_bytes());
self.buf = b.take().freeze();
}
}
impl AsyncRead for Buffer {}
impl io::Read for Buffer {
fn read(&mut self, dst: &mut [u8]) -> Result<usize, io::Error> {
if self.buf.is_empty() {
if self.err.is_some() {
Err(self.err.take().unwrap())
} else {
Err(io::Error::new(io::ErrorKind::WouldBlock, ""))
}
} else {
let size = cmp::min(self.buf.len(), dst.len());
let b = self.buf.split_to(size);
dst[..size].copy_from_slice(&b);
Ok(size)
}
}
}
impl IoStream for Buffer {
fn shutdown(&mut self, _: Shutdown) -> io::Result<()> {
Ok(())
}
fn set_nodelay(&mut self, _: bool) -> io::Result<()> {
Ok(())
}
fn set_linger(&mut self, _: Option<time::Duration>) -> io::Result<()> {
Ok(())
}
}
impl io::Write for Buffer {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl AsyncWrite for Buffer {
fn shutdown(&mut self) -> Poll<(), io::Error> {
Ok(Async::Ready(()))
}
fn write_buf<B: Buf>(&mut self, _: &mut B) -> Poll<usize, io::Error> {
Ok(Async::NotReady)
}
}
#[test]
fn test_req_parse() {
let buf = Buffer::new("GET /test HTTP/1.1\r\n\r\n");
let readbuf = BytesMut::new();
let settings = Rc::new(WorkerSettings::<HttpApplication>::new(
Vec::new(),
KeepAlive::Os,
));
let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf);
h1.poll_io();
h1.parse();
assert_eq!(h1.tasks.len(), 1);
}
#[test]
fn test_req_parse_err() {
let buf = Buffer::new("GET /test HTTP/1\r\n\r\n");
let readbuf = BytesMut::new();
let settings = Rc::new(WorkerSettings::<HttpApplication>::new(
Vec::new(),
KeepAlive::Os,
));
let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf);
h1.poll_io();
h1.parse();
assert!(h1.flags.contains(Flags::ERROR));
}
#[test]
fn test_parse() {
let mut buf = BytesMut::from("GET /test HTTP/1.1\r\n\r\n");
@ -614,10 +747,7 @@ 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"),
}
@ -918,13 +1048,7 @@ mod tests {
.as_ref(),
b"line"
);
assert!(
reader
.decode(&mut buf, &settings)
.unwrap()
.unwrap()
.eof()
);
assert!(reader.decode(&mut buf, &settings).unwrap().unwrap().eof());
}
#[test]
@ -1005,13 +1129,7 @@ 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]
@ -1029,17 +1147,9 @@ 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());

39
src/server/h1decoder.rs
View file

@ -46,7 +46,9 @@ impl H1Decoder {
}
pub fn decode<H>(
&mut self, src: &mut BytesMut, settings: &WorkerSettings<H>,
&mut self,
src: &mut BytesMut,
settings: &WorkerSettings<H>,
) -> Result<Option<Message>, DecoderError> {
// read payload
if self.decoder.is_some() {
@ -64,18 +66,11 @@ impl H1Decoder {
.map_err(DecoderError::Error)?
{
Async::Ready((msg, decoder)) => {
if let Some(decoder) = decoder {
self.decoder = Some(decoder);
Ok(Some(Message::Message {
msg,
payload: true,
}))
} else {
Ok(Some(Message::Message {
msg,
payload: false,
}))
}
self.decoder = decoder;
Ok(Some(Message::Message {
msg,
payload: self.decoder.is_some(),
}))
}
Async::NotReady => {
if src.len() >= MAX_BUFFER_SIZE {
@ -89,7 +84,9 @@ impl H1Decoder {
}
fn parse_message<H>(
&self, buf: &mut BytesMut, settings: &WorkerSettings<H>,
&self,
buf: &mut BytesMut,
settings: &WorkerSettings<H>,
) -> Poll<(SharedHttpInnerMessage, Option<EncodingDecoder>), ParseError> {
// Parse http message
let mut has_upgrade = false;
@ -148,7 +145,7 @@ impl H1Decoder {
header::CONTENT_LENGTH => {
if let Ok(s) = value.to_str() {
if let Ok(len) = s.parse::<u64>() {
content_length = Some(len)
content_length = Some(len);
} else {
debug!("illegal Content-Length: {:?}", len);
return Err(ParseError::Header);
@ -351,7 +348,10 @@ macro_rules! byte (
impl ChunkedState {
fn step(
&self, body: &mut BytesMut, size: &mut u64, buf: &mut Option<Bytes>,
&self,
body: &mut BytesMut,
size: &mut u64,
buf: &mut Option<Bytes>,
) -> Poll<ChunkedState, io::Error> {
use self::ChunkedState::*;
match *self {
@ -414,7 +414,8 @@ impl ChunkedState {
}
}
fn read_size_lf(
rdr: &mut BytesMut, size: &mut u64,
rdr: &mut BytesMut,
size: &mut u64,
) -> Poll<ChunkedState, io::Error> {
match byte!(rdr) {
b'\n' if *size > 0 => Ok(Async::Ready(ChunkedState::Body)),
@ -427,7 +428,9 @@ impl ChunkedState {
}
fn read_body(
rdr: &mut BytesMut, rem: &mut u64, buf: &mut Option<Bytes>,
rdr: &mut BytesMut,
rem: &mut u64,
buf: &mut Option<Bytes>,
) -> Poll<ChunkedState, io::Error> {
trace!("Chunked read, remaining={:?}", rem);

20
src/server/h1writer.rs
View file

@ -42,7 +42,9 @@ pub(crate) struct H1Writer<T: AsyncWrite, H: 'static> {
impl<T: AsyncWrite, H: 'static> H1Writer<T, H> {
pub fn new(
stream: T, buf: SharedBytes, settings: Rc<WorkerSettings<H>>,
stream: T,
buf: SharedBytes,
settings: Rc<WorkerSettings<H>>,
) -> H1Writer<T, H> {
H1Writer {
flags: Flags::empty(),
@ -101,7 +103,9 @@ impl<T: AsyncWrite, H: 'static> Writer for H1Writer<T, H> {
}
fn start(
&mut self, req: &mut HttpInnerMessage, msg: &mut HttpResponse,
&mut self,
req: &mut HttpInnerMessage,
msg: &mut HttpResponse,
encoding: ContentEncoding,
) -> io::Result<WriterState> {
// prepare task
@ -138,7 +142,9 @@ impl<T: AsyncWrite, H: 'static> Writer for H1Writer<T, H> {
let reason = msg.reason().as_bytes();
let mut is_bin = if let Body::Binary(ref bytes) = body {
buffer.reserve(
256 + msg.headers().len() * AVERAGE_HEADER_SIZE + bytes.len()
256
+ msg.headers().len() * AVERAGE_HEADER_SIZE
+ bytes.len()
+ reason.len(),
);
true
@ -255,9 +261,7 @@ impl<T: AsyncWrite, H: 'static> Writer for H1Writer<T, H> {
}
fn write_eof(&mut self) -> io::Result<WriterState> {
self.encoder.write_eof()?;
if !self.encoder.is_eof() {
if !self.encoder.write_eof()? {
Err(io::Error::new(
io::ErrorKind::Other,
"Last payload item, but eof is not reached",
@ -276,7 +280,9 @@ impl<T: AsyncWrite, H: 'static> Writer for H1Writer<T, H> {
unsafe { &mut *(self.buffer.as_ref() as *const _ as *mut _) };
let written = self.write_data(buf)?;
let _ = self.buffer.split_to(written);
if self.buffer.len() > self.buffer_capacity {
if shutdown && !self.buffer.is_empty()
|| (self.buffer.len() > self.buffer_capacity)
{
return Ok(Async::NotReady);
}
}

25
src/server/h2.rs
View file

@ -343,24 +343,27 @@ impl<H: 'static> Entry<H> {
}
fn poll_payload(&mut self) {
if !self.flags.contains(EntryFlags::REOF) {
if self.payload.need_read() == PayloadStatus::Read {
if let Err(err) = self.recv.release_capacity().release_capacity(32_768) {
self.payload.set_error(PayloadError::Http2(err))
}
} else if let Err(err) = self.recv.release_capacity().release_capacity(0) {
self.payload.set_error(PayloadError::Http2(err))
}
while !self.flags.contains(EntryFlags::REOF)
&& self.payload.need_read() == PayloadStatus::Read
{
match self.recv.poll() {
Ok(Async::Ready(Some(chunk))) => {
let l = chunk.len();
self.payload.feed_data(chunk);
if let Err(err) = self.recv.release_capacity().release_capacity(l) {
self.payload.set_error(PayloadError::Http2(err));
break;
}
}
Ok(Async::Ready(None)) => {
self.flags.insert(EntryFlags::REOF);
self.payload.feed_eof();
}
Ok(Async::NotReady) => break,
Err(err) => {
self.payload.set_error(PayloadError::Http2(err));
break;
}
Ok(Async::NotReady) => (),
Err(err) => self.payload.set_error(PayloadError::Http2(err)),
}
}
}

13
src/server/settings.rs
View file

@ -16,7 +16,6 @@ use body::Body;
use httpresponse::{HttpResponse, HttpResponseBuilder, HttpResponsePool};
/// Various server settings
#[derive(Clone)]
pub struct ServerSettings {
addr: Option<net::SocketAddr>,
secure: bool,
@ -28,6 +27,18 @@ pub struct ServerSettings {
unsafe impl Sync for ServerSettings {}
unsafe impl Send for ServerSettings {}
impl Clone for ServerSettings {
fn clone(&self) -> Self {
ServerSettings {
addr: self.addr,
secure: self.secure,
host: self.host.clone(),
cpu_pool: self.cpu_pool.clone(),
responses: HttpResponsePool::pool(),
}
}
}
struct InnerCpuPool {
cpu_pool: UnsafeCell<Option<CpuPool>>,
}

68
src/server/srv.rs
View file

@ -25,6 +25,20 @@ use super::worker::{Conn, SocketInfo, StopWorker, StreamHandlerType, Worker};
use super::{IntoHttpHandler, IoStream, KeepAlive};
use super::{PauseServer, ResumeServer, StopServer};
#[cfg(feature = "alpn")]
fn configure_alpn(builder: &mut SslAcceptorBuilder) -> io::Result<()> {
builder.set_alpn_protos(b"\x02h2\x08http/1.1")?;
builder.set_alpn_select_callback(|_, protos| {
const H2: &[u8] = b"\x02h2";
if protos.windows(3).any(|window| window == H2) {
Ok(b"h2")
} else {
Err(AlpnError::NOACK)
}
});
Ok(())
}
/// An HTTP Server
pub struct HttpServer<H>
where
@ -197,6 +211,16 @@ where
self.sockets.iter().map(|s| s.addr).collect()
}
/// Get addresses of bound sockets and the scheme for it.
///
/// This is useful when the server is bound from different sources
/// with some sockets listening on http and some listening on https
/// and the user should be presented with an enumeration of which
/// socket requires which protocol.
pub fn addrs_with_scheme(&self) -> Vec<(net::SocketAddr, &str)> {
self.sockets.iter().map(|s| (s.addr, s.tp.scheme())).collect()
}
/// Use listener for accepting incoming connection requests
///
/// HttpServer does not change any configuration for TcpListener,
@ -211,6 +235,40 @@ where
self
}
#[cfg(feature = "tls")]
/// Use listener for accepting incoming tls connection requests
///
/// HttpServer does not change any configuration for TcpListener,
/// it needs to be configured before passing it to listen() method.
pub fn listen_tls(mut self, lst: net::TcpListener, acceptor: TlsAcceptor) -> Self {
let addr = lst.local_addr().unwrap();
self.sockets.push(Socket {
addr,
lst,
tp: StreamHandlerType::Tls(acceptor.clone()),
});
self
}
#[cfg(feature = "alpn")]
/// Use listener for accepting incoming tls connection requests
///
/// This method sets alpn protocols to "h2" and "http/1.1"
pub fn listen_ssl(mut self, lst: net::TcpListener, mut builder: SslAcceptorBuilder) -> io::Result<Self> {
// alpn support
if !self.no_http2 {
configure_alpn(&mut builder)?;
}
let acceptor = builder.build();
let addr = lst.local_addr().unwrap();
self.sockets.push(Socket {
addr,
lst,
tp: StreamHandlerType::Alpn(acceptor.clone()),
});
Ok(self)
}
fn bind2<S: net::ToSocketAddrs>(&mut self, addr: S) -> io::Result<Vec<Socket>> {
let mut err = None;
let mut succ = false;
@ -277,15 +335,7 @@ where
) -> io::Result<Self> {
// alpn support
if !self.no_http2 {
builder.set_alpn_protos(b"\x02h2\x08http/1.1")?;
builder.set_alpn_select_callback(|_, protos| {
const H2: &[u8] = b"\x02h2";
if protos.windows(3).any(|window| window == H2) {
Ok(b"h2")
} else {
Err(AlpnError::NOACK)
}
});
configure_alpn(&mut builder)?;
}
let acceptor = builder.build();

10
src/server/worker.rs
View file

@ -239,4 +239,14 @@ impl StreamHandlerType {
}
}
}
pub(crate) fn scheme(&self) -> &'static str {
match *self {
StreamHandlerType::Normal => "http",
#[cfg(feature = "tls")]
StreamHandlerType::Tls(ref acceptor) => "https",
#[cfg(feature = "alpn")]
StreamHandlerType::Alpn(ref acceptor) => "https",
}
}
}

15
src/test.rs
View file

@ -355,12 +355,7 @@ impl<S: 'static> TestApp<S> {
/// Register handler for "/"
pub fn handler<H: Handler<S>>(&mut self, handler: H) {
self.app = Some(
self.app
.take()
.unwrap()
.resource("/", |r| r.h(handler)),
);
self.app = Some(self.app.take().unwrap().resource("/", |r| r.h(handler)));
}
/// Register middleware
@ -562,8 +557,8 @@ impl<S: 'static> TestRequest<S> {
cookies,
payload,
} = self;
let req = HttpRequest::new(method, uri, version, headers, payload);
req.as_mut().cookies = cookies;
let mut req = HttpRequest::new(method, uri, version, headers, payload);
req.set_cookies(cookies);
req.as_mut().params = params;
let (router, _) = Router::new::<S>("/", ServerSettings::default(), Vec::new());
req.with_state(Rc::new(state), router)
@ -583,8 +578,8 @@ impl<S: 'static> TestRequest<S> {
payload,
} = self;
let req = HttpRequest::new(method, uri, version, headers, payload);
req.as_mut().cookies = cookies;
let mut req = HttpRequest::new(method, uri, version, headers, payload);
req.set_cookies(cookies);
req.as_mut().params = params;
req.with_state(Rc::new(state), router)
}

195
src/with.rs
View file

@ -9,6 +9,62 @@ use handler::{AsyncResult, AsyncResultItem, FromRequest, Handler, Responder};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
/// Extractor configuration
///
/// `Route::with()` and `Route::with_async()` returns instance
/// of the `ExtractorConfig` type. It could be used for extractor configuration.
///
/// In this example `Form<FormData>` configured.
///
/// ```rust
/// # extern crate actix_web;
/// #[macro_use] extern crate serde_derive;
/// use actix_web::{App, Form, Result, http};
///
/// #[derive(Deserialize)]
/// struct FormData {
/// username: String,
/// }
///
/// fn index(form: Form<FormData>) -> Result<String> {
/// Ok(format!("Welcome {}!", form.username))
/// }
///
/// fn main() {
/// let app = App::new().resource(
/// "/index.html", |r| {
/// r.method(http::Method::GET)
/// .with(index)
/// .limit(4096);} // <- change form extractor configuration
/// );
/// }
/// ```
///
/// Same could be donce with multiple extractors
///
/// ```rust
/// # extern crate actix_web;
/// #[macro_use] extern crate serde_derive;
/// use actix_web::{App, Form, Path, Result, http};
///
/// #[derive(Deserialize)]
/// struct FormData {
/// username: String,
/// }
///
/// fn index(data: (Path<(String,)>, Form<FormData>)) -> Result<String> {
/// Ok(format!("Welcome {}!", data.1.username))
/// }
///
/// fn main() {
/// let app = App::new().resource(
/// "/index.html", |r| {
/// r.method(http::Method::GET)
/// .with(index)
/// .1.limit(4096);} // <- change form extractor configuration
/// );
/// }
/// ```
pub struct ExtractorConfig<S: 'static, T: FromRequest<S>> {
cfg: Rc<UnsafeCell<T::Config>>,
}
@ -167,6 +223,145 @@ where
}
}
pub struct WithAsync<T, S, F, R, I, E>
where
F: Fn(T) -> R,
R: Future<Item = I, Error = E>,
I: Responder,
E: Into<E>,
T: FromRequest<S>,
S: 'static,
{
hnd: Rc<UnsafeCell<F>>,
cfg: ExtractorConfig<S, T>,
_s: PhantomData<S>,
}
impl<T, S, F, R, I, E> WithAsync<T, S, F, R, I, E>
where
F: Fn(T) -> R,
R: Future<Item = I, Error = E>,
I: Responder,
E: Into<Error>,
T: FromRequest<S>,
S: 'static,
{
pub fn new(f: F, cfg: ExtractorConfig<S, T>) -> Self {
WithAsync {
cfg,
hnd: Rc::new(UnsafeCell::new(f)),
_s: PhantomData,
}
}
}
impl<T, S, F, R, I, E> Handler<S> for WithAsync<T, S, F, R, I, E>
where
F: Fn(T) -> R + 'static,
R: Future<Item = I, Error = E> + 'static,
I: Responder + 'static,
E: Into<Error> + 'static,
T: FromRequest<S> + 'static,
S: 'static,
{
type Result = AsyncResult<HttpResponse>;
fn handle(&mut self, req: HttpRequest<S>) -> Self::Result {
let mut fut = WithAsyncHandlerFut {
req,
started: false,
hnd: Rc::clone(&self.hnd),
cfg: self.cfg.clone(),
fut1: None,
fut2: None,
fut3: None,
};
match fut.poll() {
Ok(Async::Ready(resp)) => AsyncResult::ok(resp),
Ok(Async::NotReady) => AsyncResult::async(Box::new(fut)),
Err(e) => AsyncResult::err(e),
}
}
}
struct WithAsyncHandlerFut<T, S, F, R, I, E>
where
F: Fn(T) -> R,
R: Future<Item = I, Error = E> + 'static,
I: Responder + 'static,
E: Into<Error> + 'static,
T: FromRequest<S> + 'static,
S: 'static,
{
started: bool,
hnd: Rc<UnsafeCell<F>>,
cfg: ExtractorConfig<S, T>,
req: HttpRequest<S>,
fut1: Option<Box<Future<Item = T, Error = Error>>>,
fut2: Option<R>,
fut3: Option<Box<Future<Item = HttpResponse, Error = Error>>>,
}
impl<T, S, F, R, I, E> Future for WithAsyncHandlerFut<T, S, F, R, I, E>
where
F: Fn(T) -> R,
R: Future<Item = I, Error = E> + 'static,
I: Responder + 'static,
E: Into<Error> + 'static,
T: FromRequest<S> + 'static,
S: 'static,
{
type Item = HttpResponse;
type Error = Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if let Some(ref mut fut) = self.fut3 {
return fut.poll();
}
if self.fut2.is_some() {
return match self.fut2.as_mut().unwrap().poll() {
Ok(Async::NotReady) => Ok(Async::NotReady),
Ok(Async::Ready(r)) => match r.respond_to(&self.req) {
Ok(r) => match r.into().into() {
AsyncResultItem::Err(err) => Err(err),
AsyncResultItem::Ok(resp) => Ok(Async::Ready(resp)),
AsyncResultItem::Future(fut) => {
self.fut3 = Some(fut);
self.poll()
}
},
Err(e) => Err(e.into()),
},
Err(e) => Err(e.into()),
};
}
let item = if !self.started {
self.started = true;
let reply = T::from_request(&self.req, self.cfg.as_ref()).into();
match reply.into() {
AsyncResultItem::Err(err) => return Err(err),
AsyncResultItem::Ok(msg) => msg,
AsyncResultItem::Future(fut) => {
self.fut1 = Some(fut);
return self.poll();
}
}
} else {
match self.fut1.as_mut().unwrap().poll()? {
Async::Ready(item) => item,
Async::NotReady => return Ok(Async::NotReady),
}
};
let hnd: &mut F = unsafe { &mut *self.hnd.get() };
self.fut2 = Some((*hnd)(item));
self.poll()
}
}
pub struct With2<T1, T2, S, F, R>
where
F: Fn(T1, T2) -> R,

44
src/ws/client.rs
View file

@ -518,24 +518,22 @@ impl ClientWriter {
fn as_mut(&mut self) -> &mut Inner {
unsafe { &mut *self.inner.get() }
}
}
impl WsWriter for ClientWriter {
/// Send text frame
#[inline]
fn text<T: Into<Binary>>(&mut self, text: T) {
pub fn text<T: Into<Binary>>(&mut self, text: T) {
self.write(Frame::message(text.into(), OpCode::Text, true, true));
}
/// Send binary frame
#[inline]
fn binary<B: Into<Binary>>(&mut self, data: B) {
pub fn binary<B: Into<Binary>>(&mut self, data: B) {
self.write(Frame::message(data, OpCode::Binary, true, true));
}
/// Send ping frame
#[inline]
fn ping(&mut self, message: &str) {
pub fn ping(&mut self, message: &str) {
self.write(Frame::message(
Vec::from(message),
OpCode::Ping,
@ -546,7 +544,7 @@ impl WsWriter for ClientWriter {
/// Send pong frame
#[inline]
fn pong(&mut self, message: &str) {
pub fn pong(&mut self, message: &str) {
self.write(Frame::message(
Vec::from(message),
OpCode::Pong,
@ -557,7 +555,39 @@ impl WsWriter for ClientWriter {
/// Send close frame
#[inline]
fn close(&mut self, reason: Option<CloseReason>) {
pub fn close(&mut self, reason: Option<CloseReason>) {
self.write(Frame::close(reason, true));
}
}
impl WsWriter for ClientWriter {
/// Send text frame
#[inline]
fn send_text<T: Into<Binary>>(&mut self, text: T) {
self.text(text)
}
/// Send binary frame
#[inline]
fn send_binary<B: Into<Binary>>(&mut self, data: B) {
self.binary(data)
}
/// Send ping frame
#[inline]
fn send_ping(&mut self, message: &str) {
self.ping(message)
}
/// Send pong frame
#[inline]
fn send_pong(&mut self, message: &str) {
self.pong(message)
}
/// Send close frame
#[inline]
fn send_close(&mut self, reason: Option<CloseReason>) {
self.close(reason);
}
}

72
src/ws/context.rs
View file

@ -13,9 +13,9 @@ use context::{ActorHttpContext, Drain, Frame as ContextFrame};
use error::{Error, ErrorInternalServerError};
use httprequest::HttpRequest;
use ws::WsWriter;
use ws::frame::Frame;
use ws::proto::{CloseReason, OpCode};
use ws::WsWriter;
/// Execution context for `WebSockets` actors
pub struct WebsocketContext<A, S = ()>
@ -149,6 +149,46 @@ where
Drain::new(rx)
}
/// Send text frame
#[inline]
pub fn text<T: Into<Binary>>(&mut self, text: T) {
self.write(Frame::message(text.into(), OpCode::Text, true, false));
}
/// Send binary frame
#[inline]
pub fn binary<B: Into<Binary>>(&mut self, data: B) {
self.write(Frame::message(data, OpCode::Binary, true, false));
}
/// Send ping frame
#[inline]
pub fn ping(&mut self, message: &str) {
self.write(Frame::message(
Vec::from(message),
OpCode::Ping,
true,
false,
));
}
/// Send pong frame
#[inline]
pub fn pong(&mut self, message: &str) {
self.write(Frame::message(
Vec::from(message),
OpCode::Pong,
true,
false,
));
}
/// Send close frame
#[inline]
pub fn close(&mut self, reason: Option<CloseReason>) {
self.write(Frame::close(reason, false));
}
/// Check if connection still open
#[inline]
pub fn connected(&self) -> bool {
@ -181,42 +221,32 @@ where
{
/// Send text frame
#[inline]
fn text<T: Into<Binary>>(&mut self, text: T) {
self.write(Frame::message(text.into(), OpCode::Text, true, false));
fn send_text<T: Into<Binary>>(&mut self, text: T) {
self.text(text)
}
/// Send binary frame
#[inline]
fn binary<B: Into<Binary>>(&mut self, data: B) {
self.write(Frame::message(data, OpCode::Binary, true, false));
fn send_binary<B: Into<Binary>>(&mut self, data: B) {
self.binary(data)
}
/// Send ping frame
#[inline]
fn ping(&mut self, message: &str) {
self.write(Frame::message(
Vec::from(message),
OpCode::Ping,
true,
false,
));
fn send_ping(&mut self, message: &str) {
self.ping(message)
}
/// Send pong frame
#[inline]
fn pong(&mut self, message: &str) {
self.write(Frame::message(
Vec::from(message),
OpCode::Pong,
true,
false,
));
fn send_pong(&mut self, message: &str) {
self.pong(message)
}
/// Send close frame
#[inline]
fn close(&mut self, reason: Option<CloseReason>) {
self.write(Frame::close(reason, false));
fn send_close(&mut self, reason: Option<CloseReason>) {
self.close(reason)
}
}

10
src/ws/mod.rs
View file

@ -343,15 +343,15 @@ where
/// Common writing methods for a websocket.
pub trait WsWriter {
/// Send a text
fn text<T: Into<Binary>>(&mut self, text: T);
fn send_text<T: Into<Binary>>(&mut self, text: T);
/// Send a binary
fn binary<B: Into<Binary>>(&mut self, data: B);
fn send_binary<B: Into<Binary>>(&mut self, data: B);
/// Send a ping message
fn ping(&mut self, message: &str);
fn send_ping(&mut self, message: &str);
/// Send a pong message
fn pong(&mut self, message: &str);
fn send_pong(&mut self, message: &str);
/// Close the connection
fn close(&mut self, reason: Option<CloseReason>);
fn send_close(&mut self, reason: Option<CloseReason>);
}
#[cfg(test)]

78
tests/test_handlers.rs
View file

@ -9,6 +9,7 @@ extern crate tokio_core;
extern crate serde_derive;
extern crate serde_json;
use std::io;
use std::time::Duration;
use actix::*;
@ -377,6 +378,83 @@ fn test_path_and_query_extractor2_async4() {
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}
#[cfg(actix_impl_trait)]
fn test_impl_trait(
data: (Json<Value>, Path<PParam>, Query<PParam>),
) -> impl Future<Item = String, Error = io::Error> {
Timeout::new(Duration::from_millis(10), &Arbiter::handle())
.unwrap()
.and_then(move |_| {
Ok(format!(
"Welcome {} - {}!",
data.1.username,
(data.0).0
))
})
}
#[cfg(actix_impl_trait)]
fn test_impl_trait_err(
_data: (Json<Value>, Path<PParam>, Query<PParam>),
) -> impl Future<Item = String, Error = io::Error> {
Timeout::new(Duration::from_millis(10), &Arbiter::handle())
.unwrap()
.and_then(move |_| Err(io::Error::new(io::ErrorKind::Other, "other")))
}
#[cfg(actix_impl_trait)]
#[test]
fn test_path_and_query_extractor2_async4_impl_trait() {
let mut srv = test::TestServer::new(|app| {
app.resource("/{username}/index.html", |r| {
r.route().with_async(test_impl_trait)
});
});
// client request
let request = srv.post()
.uri(srv.url("/test1/index.html?username=test2"))
.header("content-type", "application/json")
.body("{\"test\": 1}")
.unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
// read response
let bytes = srv.execute(response.body()).unwrap();
assert_eq!(
bytes,
Bytes::from_static(b"Welcome test1 - {\"test\":1}!")
);
// client request
let request = srv.get()
.uri(srv.url("/test1/index.html"))
.finish()
.unwrap();
let response = srv.execute(request.send()).unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}
#[cfg(actix_impl_trait)]
#[test]
fn test_path_and_query_extractor2_async4_impl_trait_err() {
let mut srv = test::TestServer::new(|app| {
app.resource("/{username}/index.html", |r| {
r.route().with_async(test_impl_trait_err)
});
});
// client request
let request = srv.post()
.uri(srv.url("/test1/index.html?username=test2"))
.header("content-type", "application/json")
.body("{\"test\": 1}")
.unwrap();
let response = srv.execute(request.send()).unwrap();
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
}
#[test]
fn test_non_ascii_route() {
let mut srv = test::TestServer::new(|app| {

171
tests/test_middleware.rs
View file

@ -21,28 +21,24 @@ struct MiddlewareTest {
impl<S> middleware::Middleware<S> for MiddlewareTest {
fn start(&self, _: &mut HttpRequest<S>) -> Result<middleware::Started> {
self.start.store(
self.start.load(Ordering::Relaxed) + 1,
Ordering::Relaxed,
);
self.start
.store(self.start.load(Ordering::Relaxed) + 1, Ordering::Relaxed);
Ok(middleware::Started::Done)
}
fn response(
&self, _: &mut HttpRequest<S>, resp: HttpResponse,
&self,
_: &mut HttpRequest<S>,
resp: HttpResponse,
) -> Result<middleware::Response> {
self.response.store(
self.response.load(Ordering::Relaxed) + 1,
Ordering::Relaxed,
);
self.response
.store(self.response.load(Ordering::Relaxed) + 1, Ordering::Relaxed);
Ok(middleware::Response::Done(resp))
}
fn finish(&self, _: &mut HttpRequest<S>, _: &HttpResponse) -> middleware::Finished {
self.finish.store(
self.finish.load(Ordering::Relaxed) + 1,
Ordering::Relaxed,
);
self.finish
.store(self.finish.load(Ordering::Relaxed) + 1, Ordering::Relaxed);
middleware::Finished::Done
}
}
@ -187,10 +183,7 @@ fn test_scope_middleware() {
})
});
let request = srv.get()
.uri(srv.url("/scope/test"))
.finish()
.unwrap();
let request = srv.get().uri(srv.url("/scope/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
@ -226,10 +219,7 @@ fn test_scope_middleware_multiple() {
})
});
let request = srv.get()
.uri(srv.url("/scope/test"))
.finish()
.unwrap();
let request = srv.get().uri(srv.url("/scope/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
@ -337,10 +327,7 @@ fn test_scope_middleware_async_handler() {
})
});
let request = srv.get()
.uri(srv.url("/scope/test"))
.finish()
.unwrap();
let request = srv.get().uri(srv.url("/scope/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
@ -402,10 +389,7 @@ fn test_scope_middleware_async_error() {
})
});
let request = srv.get()
.uri(srv.url("/scope/test"))
.finish()
.unwrap();
let request = srv.get().uri(srv.url("/scope/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert_eq!(response.status(), http::StatusCode::BAD_REQUEST);
@ -466,7 +450,9 @@ impl<S> middleware::Middleware<S> for MiddlewareAsyncTest {
}
fn response(
&self, _: &mut HttpRequest<S>, resp: HttpResponse,
&self,
_: &mut HttpRequest<S>,
resp: HttpResponse,
) -> Result<middleware::Response> {
let to = Timeout::new(Duration::from_millis(10), &Arbiter::handle()).unwrap();
@ -551,7 +537,43 @@ fn test_async_middleware_multiple() {
assert_eq!(num1.load(Ordering::Relaxed), 2);
assert_eq!(num2.load(Ordering::Relaxed), 2);
thread::sleep(Duration::from_millis(30));
thread::sleep(Duration::from_millis(50));
assert_eq!(num3.load(Ordering::Relaxed), 2);
}
#[test]
fn test_async_sync_middleware_multiple() {
let num1 = Arc::new(AtomicUsize::new(0));
let num2 = Arc::new(AtomicUsize::new(0));
let num3 = Arc::new(AtomicUsize::new(0));
let act_num1 = Arc::clone(&num1);
let act_num2 = Arc::clone(&num2);
let act_num3 = Arc::clone(&num3);
let mut srv = test::TestServer::with_factory(move || {
App::new()
.middleware(MiddlewareAsyncTest {
start: Arc::clone(&act_num1),
response: Arc::clone(&act_num2),
finish: Arc::clone(&act_num3),
})
.middleware(MiddlewareTest {
start: Arc::clone(&act_num1),
response: Arc::clone(&act_num2),
finish: Arc::clone(&act_num3),
})
.resource("/test", |r| r.f(|_| HttpResponse::Ok()))
});
let request = srv.get().uri(srv.url("/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
assert_eq!(num1.load(Ordering::Relaxed), 2);
assert_eq!(num2.load(Ordering::Relaxed), 2);
thread::sleep(Duration::from_millis(50));
assert_eq!(num3.load(Ordering::Relaxed), 2);
}
@ -577,10 +599,7 @@ fn test_async_scope_middleware() {
})
});
let request = srv.get()
.uri(srv.url("/scope/test"))
.finish()
.unwrap();
let request = srv.get().uri(srv.url("/scope/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
@ -618,10 +637,45 @@ fn test_async_scope_middleware_multiple() {
})
});
let request = srv.get()
.uri(srv.url("/scope/test"))
.finish()
.unwrap();
let request = srv.get().uri(srv.url("/scope/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
assert_eq!(num1.load(Ordering::Relaxed), 2);
assert_eq!(num2.load(Ordering::Relaxed), 2);
thread::sleep(Duration::from_millis(20));
assert_eq!(num3.load(Ordering::Relaxed), 2);
}
#[test]
fn test_async_async_scope_middleware_multiple() {
let num1 = Arc::new(AtomicUsize::new(0));
let num2 = Arc::new(AtomicUsize::new(0));
let num3 = Arc::new(AtomicUsize::new(0));
let act_num1 = Arc::clone(&num1);
let act_num2 = Arc::clone(&num2);
let act_num3 = Arc::clone(&num3);
let mut srv = test::TestServer::with_factory(move || {
App::new().scope("/scope", |scope| {
scope
.middleware(MiddlewareAsyncTest {
start: Arc::clone(&act_num1),
response: Arc::clone(&act_num2),
finish: Arc::clone(&act_num3),
})
.middleware(MiddlewareTest {
start: Arc::clone(&act_num1),
response: Arc::clone(&act_num2),
finish: Arc::clone(&act_num3),
})
.resource("/test", |r| r.f(|_| HttpResponse::Ok()))
})
});
let request = srv.get().uri(srv.url("/scope/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
@ -703,3 +757,42 @@ fn test_async_resource_middleware_multiple() {
thread::sleep(Duration::from_millis(40));
assert_eq!(num3.load(Ordering::Relaxed), 2);
}
#[test]
fn test_async_sync_resource_middleware_multiple() {
let num1 = Arc::new(AtomicUsize::new(0));
let num2 = Arc::new(AtomicUsize::new(0));
let num3 = Arc::new(AtomicUsize::new(0));
let act_num1 = Arc::clone(&num1);
let act_num2 = Arc::clone(&num2);
let act_num3 = Arc::clone(&num3);
let mut srv = test::TestServer::with_factory(move || {
let mw1 = MiddlewareAsyncTest {
start: Arc::clone(&act_num1),
response: Arc::clone(&act_num2),
finish: Arc::clone(&act_num3),
};
let mw2 = MiddlewareTest {
start: Arc::clone(&act_num1),
response: Arc::clone(&act_num2),
finish: Arc::clone(&act_num3),
};
App::new().resource("/test", move |r| {
r.middleware(mw1);
r.middleware(mw2);
r.h(|_| HttpResponse::Ok());
})
});
let request = srv.get().uri(srv.url("/test")).finish().unwrap();
let response = srv.execute(request.send()).unwrap();
assert!(response.status().is_success());
assert_eq!(num1.load(Ordering::Relaxed), 2);
assert_eq!(num2.load(Ordering::Relaxed), 2);
thread::sleep(Duration::from_millis(40));
assert_eq!(num3.load(Ordering::Relaxed), 2);
}

2
tests/test_server.rs
View file

@ -809,7 +809,7 @@ fn test_h2() {
})
});
let _res = core.run(tcp);
// assert_eq!(res.unwrap(), Bytes::from_static(STR.as_ref()));
// assert_eq!(_res.unwrap(), Bytes::from_static(STR.as_ref()));
}
#[test]