actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-27 17:22:57 +01:00
Code Issues Releases Wiki Activity

pass request by ref; added middleware support

Browse Source
This commit is contained in:
Nikolay Kim 2017-10-21 22:59:09 -07:00
parent 7364e088be
commit afe9459ce1
17 changed files with 470 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.toml
View File

@ -54,6 +54,9 @@ git = "https://github.com/fafhrd91/actix.git"
default-features = false
features = []
[dev-dependencies]
env_logger = "*"
[profile.release]
lto = true
opt-level = 3

8
examples/basic.rs
View File

@ -6,13 +6,13 @@ extern crate env_logger;
use actix_web::*;
/// somple handle
fn index(req: HttpRequest, payload: Payload, state: &()) -> HttpResponse {
fn index(req: &mut HttpRequest, payload: Payload, state: &()) -> HttpResponse {
println!("{:?}", req);
httpcodes::HTTPOk.into()
}
/// handle with path parameters like `/name/{name}/`
fn with_param(req: HttpRequest, payload: Payload, state: &()) -> HttpResponse {
fn with_param(req: &mut HttpRequest, payload: Payload, state: &()) -> HttpResponse {
println!("{:?}", req);
HttpResponse::builder(StatusCode::OK)
@ -22,10 +22,14 @@ fn with_param(req: HttpRequest, payload: Payload, state: &()) -> HttpResponse {
}
fn main() {
::std::env::set_var("RUST_LOG", "actix_web=info");
let _ = env_logger::init();
let sys = actix::System::new("ws-example");
HttpServer::new(
Application::default("/")
// enable logger
.middleware(Logger::new(None))
// register simple handler, handle all methods
.handler("/index.html", index)
// with path parameters

0
examples/websocket/static/index.html → examples/static/index.html
View File

0
examples/websocket/client.py → examples/websocket-client.py
View File

11
examples/websocket/src/main.rs → examples/websocket.rs
View File

@ -1,6 +1,7 @@
#![allow(unused_variables)]
extern crate actix;
extern crate actix_web;
extern crate env_logger;
use actix::*;
use actix_web::*;
@ -15,7 +16,8 @@ impl Actor for MyWebSocket {
impl Route for MyWebSocket {
type State = ();
fn request(req: HttpRequest, payload: Payload, ctx: &mut HttpContext<Self>) -> Reply<Self>
fn request(req: &mut HttpRequest,
payload: Payload, ctx: &mut HttpContext<Self>) -> Reply<Self>
{
match ws::handshake(&req) {
Ok(resp) => {
@ -59,12 +61,17 @@ impl Handler<ws::Message> for MyWebSocket {
}
fn main() {
::std::env::set_var("RUST_LOG", "actix_web=info");
let _ = env_logger::init();
let sys = actix::System::new("ws-example");
HttpServer::new(
Application::default("/")
// enable logger
.middleware(Logger::new(None))
// websocket route
.resource("/ws/", |r| r.get::<MyWebSocket>())
.route_handler("/", StaticFiles::new("static/", true)))
.route_handler("/", StaticFiles::new("examples/static/", true)))
.serve::<_, ()>("127.0.0.1:8080").unwrap();
println!("Started http server: 127.0.0.1:8080");

12
examples/websocket/Cargo.toml
View File

@ -1,12 +0,0 @@
[package]
name = "websocket-example"
version = "0.1.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
[[bin]]
name = "websocket"
path = "src/main.rs"
[dependencies]
actix = { git = "https://github.com/fafhrd91/actix.git" }
actix-web = { path = "../../" }

88
src/application.rs
View File

@ -12,6 +12,24 @@ use httpresponse::HttpResponse;
use server::HttpHandler;
#[allow(unused_variables)]
pub trait Middleware {
/// Method is called when request is ready.
fn start(&self, req: &mut HttpRequest) -> Result<(), HttpResponse> {
Ok(())
}
/// Method is called when handler returns response,
/// but before sending body stream to peer.
fn response(&self, req: &mut HttpRequest, resp: HttpResponse) -> HttpResponse {
resp
}
/// Http interation is finished
fn finish(&self, req: &mut HttpRequest, resp: &HttpResponse) {}
}
/// Application
pub struct Application<S> {
state: Rc<S>,
@ -19,6 +37,26 @@ pub struct Application<S> {
default: Resource<S>,
handlers: HashMap<String, Box<RouteHandler<S>>>,
router: RouteRecognizer<Resource<S>>,
middlewares: Rc<Vec<Box<Middleware>>>,
}
impl<S: 'static> Application<S> {
fn run(&self, req: &mut HttpRequest, payload: Payload) -> Task {
if let Some((params, h)) = self.router.recognize(req.path()) {
if let Some(params) = params {
req.set_match_info(params);
}
h.handle(req, payload, Rc::clone(&self.state))
} else {
for (prefix, handler) in &self.handlers {
if req.path().starts_with(prefix) {
return handler.handle(req, payload, Rc::clone(&self.state))
}
}
self.default.handle(req, payload, Rc::clone(&self.state))
}
}
}
impl<S: 'static> HttpHandler for Application<S> {
@ -27,21 +65,19 @@ impl<S: 'static> HttpHandler for Application<S> {
&self.prefix
}
fn handle(&self, req: HttpRequest, payload: Payload) -> Task {
if let Some((params, h)) = self.router.recognize(req.path()) {
if let Some(params) = params {
h.handle(
req.with_match_info(params), payload, Rc::clone(&self.state))
fn handle(&self, req: &mut HttpRequest, payload: Payload) -> Task {
// run middlewares
if !self.middlewares.is_empty() {
for middleware in self.middlewares.iter() {
if let Err(resp) = middleware.start(req) {
return Task::reply(resp)
};
}
let mut task = self.run(req, payload);
task.set_middlewares(Rc::clone(&self.middlewares));
task
} else {
h.handle(req, payload, Rc::clone(&self.state))
}
} else {
for (prefix, handler) in &self.handlers {
if req.path().starts_with(prefix) {
return handler.handle(req, payload, Rc::clone(&self.state))
}
}
self.default.handle(req, payload, Rc::clone(&self.state))
self.run(req, payload)
}
}
}
@ -56,7 +92,9 @@ impl Application<()> {
prefix: prefix.to_string(),
default: Resource::default(),
handlers: HashMap::new(),
resources: HashMap::new()})
resources: HashMap::new(),
middlewares: Vec::new(),
})
}
}
}
@ -73,7 +111,9 @@ impl<S> Application<S> where S: 'static {
prefix: prefix.to_string(),
default: Resource::default(),
handlers: HashMap::new(),
resources: HashMap::new()})
resources: HashMap::new(),
middlewares: Vec::new(),
})
}
}
}
@ -84,6 +124,7 @@ struct ApplicationBuilderParts<S> {
default: Resource<S>,
handlers: HashMap<String, Box<RouteHandler<S>>>,
resources: HashMap<String, Resource<S>>,
middlewares: Vec<Box<Middleware>>,
}
/// Application builder
@ -192,7 +233,7 @@ impl<S> ApplicationBuilder<S> where S: 'static {
/// }
/// ```
pub fn handler<P, F, R>(&mut self, path: P, handler: F) -> &mut Self
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Into<HttpResponse> + 'static,
P: ToString,
{
@ -217,6 +258,15 @@ impl<S> ApplicationBuilder<S> where S: 'static {
self
}
/// Construct application
pub fn middleware<T>(&mut self, mw: T) -> &mut Self
where T: Middleware + 'static
{
self.parts.as_mut().expect("Use after finish")
.middlewares.push(Box::new(mw));
self
}
/// Construct application
pub fn finish(&mut self) -> Application<S> {
let parts = self.parts.take().expect("Use after finish");
@ -243,7 +293,9 @@ impl<S> ApplicationBuilder<S> where S: 'static {
prefix: prefix.clone(),
default: parts.default,
handlers: handlers,
router: RouteRecognizer::new(prefix, routes) }
router: RouteRecognizer::new(prefix, routes),
middlewares: Rc::new(parts.middlewares),
}
}
}

3
src/dev.rs
View File

@ -10,13 +10,14 @@
pub use ws;
pub use httpcodes;
pub use error::ParseError;
pub use application::{Application, ApplicationBuilder};
pub use application::{Application, ApplicationBuilder, Middleware};
pub use httprequest::HttpRequest;
pub use httpresponse::{Body, HttpResponse, HttpResponseBuilder};
pub use payload::{Payload, PayloadItem, PayloadError};
pub use resource::{Reply, Resource};
pub use route::{Route, RouteFactory, RouteHandler};
pub use recognizer::Params;
pub use logger::Logger;
pub use server::HttpServer;
pub use context::HttpContext;
pub use staticfiles::StaticFiles;

2
src/httpcodes.rs
View File

@ -61,7 +61,7 @@ impl StaticResponse {
}
impl<S> RouteHandler<S> for StaticResponse {
fn handle(&self, _: HttpRequest, _: Payload, _: Rc<S>) -> Task {
fn handle(&self, _: &mut HttpRequest, _: Payload, _: Rc<S>) -> Task {
Task::reply(HttpResponse::new(self.0, Body::Empty))
}
}

35
src/httprequest.rs
View File

@ -4,7 +4,7 @@ use std::collections::HashMap;
use bytes::BytesMut;
use futures::{Async, Future, Stream, Poll};
use url::form_urlencoded;
use http::{header, Method, Version, Uri, HeaderMap};
use http::{header, Method, Version, Uri, HeaderMap, Extensions};
use {Cookie, CookieParseError};
use {HttpRange, HttpRangeParseError};
@ -22,6 +22,7 @@ pub struct HttpRequest {
headers: HeaderMap,
params: Params,
cookies: Vec<Cookie<'static>>,
extensions: Extensions,
}
impl HttpRequest {
@ -35,9 +36,28 @@ impl HttpRequest {
headers: headers,
params: Params::empty(),
cookies: Vec::new(),
extensions: Extensions::new(),
}
}
pub(crate) fn for_error() -> HttpRequest {
HttpRequest {
method: Method::GET,
uri: Uri::default(),
version: Version::HTTP_11,
headers: HeaderMap::new(),
params: Params::empty(),
cookies: Vec::new(),
extensions: Extensions::new(),
}
}
/// Protocol extensions.
#[inline]
pub fn extensions(&mut self) -> &mut Extensions {
&mut self.extensions
}
/// Read the Request Uri.
#[inline]
pub fn uri(&self) -> &Uri { &self.uri }
@ -111,16 +131,9 @@ impl HttpRequest {
#[inline]
pub fn match_info(&self) -> &Params { &self.params }
/// Create new request with Params object.
pub fn with_match_info(self, params: Params) -> Self {
HttpRequest {
method: self.method,
uri: self.uri,
version: self.version,
headers: self.headers,
params: params,
cookies: self.cookies,
}
/// Set request Params.
pub fn set_match_info(&mut self, params: Params) {
self.params = params;
}
/// Checks if a connection should be kept alive.

4
src/lib.rs
View File

@ -32,6 +32,7 @@ mod date;
mod decode;
mod httprequest;
mod httpresponse;
mod logger;
mod payload;
mod resource;
mod recognizer;
@ -48,13 +49,14 @@ pub mod dev;
pub mod httpcodes;
pub mod multipart;
pub use error::ParseError;
pub use application::{Application, ApplicationBuilder};
pub use application::{Application, ApplicationBuilder, Middleware};
pub use httprequest::{HttpRequest, UrlEncoded};
pub use httpresponse::{Body, HttpResponse, HttpResponseBuilder};
pub use payload::{Payload, PayloadItem, PayloadError};
pub use route::{Route, RouteFactory, RouteHandler};
pub use resource::{Reply, Resource};
pub use recognizer::{Params, RouteRecognizer};
pub use logger::Logger;
pub use server::HttpServer;
pub use context::HttpContext;
pub use staticfiles::StaticFiles;

274
src/logger.rs Normal file
View File

@ -0,0 +1,274 @@
//! Request logging middleware
use std::fmt;
use std::str::Chars;
use std::iter::Peekable;
use std::fmt::{Display, Formatter};
use time;
use application::Middleware;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
/// `Middleware` for logging request and response info to the terminal.
pub struct Logger {
format: Format,
}
impl Logger {
/// Create `Logger` middlewares with the specified `format`.
/// If a `None` is passed in, uses the default format:
///
/// ```ignore
/// {method} {uri} -> {status} ({response-time} ms)
/// ```
///
/// ```rust,ignore
/// let app = Application::default("/")
/// .middleware(Logger::new(None))
/// .finish()
/// ```
pub fn new(format: Option<Format>) -> Logger {
let format = format.unwrap_or_default();
Logger { format: format.clone() }
}
}
struct StartTime(time::Tm);
impl Logger {
fn initialise(&self, req: &mut HttpRequest) {
req.extensions().insert(StartTime(time::now()));
}
fn log(&self, req: &mut HttpRequest, resp: &HttpResponse) {
let entry_time = req.extensions().get::<StartTime>().unwrap().0;
let response_time = time::now() - entry_time;
let response_time_ms = (response_time.num_seconds() * 1000) as f64 + (response_time.num_nanoseconds().unwrap_or(0) as f64) / 1000000.0;
{
let render = |fmt: &mut Formatter, text: &FormatText| {
match *text {
FormatText::Str(ref string) => fmt.write_str(string),
FormatText::Method => req.method().fmt(fmt),
FormatText::URI => req.uri().fmt(fmt),
FormatText::Status => resp.status().fmt(fmt),
FormatText::ResponseTime =>
fmt.write_fmt(format_args!("{} ms", response_time_ms)),
FormatText::RemoteAddr => Ok(()), //req.remote_addr.fmt(fmt),
FormatText::RequestTime => {
entry_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ%z")
.unwrap()
.fmt(fmt)
}
}
};
info!("{}", self.format.display_with(&render));
//println!("{}", self.format.display_with(&render));
}
}
}
impl Middleware for Logger {
fn start(&self, req: &mut HttpRequest) -> Result<(), HttpResponse> {
self.initialise(req);
Ok(())
}
fn finish(&self, req: &mut HttpRequest, resp: &HttpResponse) {
self.log(req, resp);
}
}
use self::FormatText::{Method, URI, Status, ResponseTime, RemoteAddr, RequestTime};
/// A formatting style for the `Logger`, consisting of multiple
/// `FormatText`s concatenated into one line.
#[derive(Clone)]
#[doc(hidden)]
pub struct Format(Vec<FormatText>);
impl Default for Format {
/// Return the default formatting style for the `Logger`:
///
/// ```ignore
/// {method} {uri} -> {status} ({response-time})
/// // This will be written as: {method} {uri} -> {status} ({response-time})
/// ```
fn default() -> Format {
Format::new("{method} {uri} {status} ({response-time})").unwrap()
}
}
impl Format {
/// Create a `Format` from a format string, which can contain the fields
/// `{method}`, `{uri}`, `{status}`, `{response-time}`, `{ip-addr}` and
/// `{request-time}`.
///
/// Returns `None` if the format string syntax is incorrect.
pub fn new(s: &str) -> Option<Format> {
let parser = FormatParser::new(s.chars().peekable());
let mut results = Vec::new();
for unit in parser {
match unit {
Some(unit) => results.push(unit),
None => return None
}
}
Some(Format(results))
}
}
pub(crate) trait ContextDisplay<'a> {
type Item;
type Display: fmt::Display;
fn display_with(&'a self,
render: &'a Fn(&mut Formatter, &Self::Item) -> Result<(), fmt::Error>)
-> Self::Display;
}
impl<'a> ContextDisplay<'a> for Format {
type Item = FormatText;
type Display = FormatDisplay<'a>;
fn display_with(&'a self,
render: &'a Fn(&mut Formatter, &FormatText) -> Result<(), fmt::Error>)
-> FormatDisplay<'a> {
FormatDisplay {
format: self,
render: render,
}
}
}
struct FormatParser<'a> {
// The characters of the format string.
chars: Peekable<Chars<'a>>,
// A reusable buffer for parsing style attributes.
object_buffer: String,
finished: bool
}
impl<'a> FormatParser<'a> {
fn new(chars: Peekable<Chars>) -> FormatParser {
FormatParser {
chars: chars,
// No attributes are longer than 14 characters, so we can avoid reallocating.
object_buffer: String::with_capacity(14),
finished: false
}
}
}
// Some(None) means there was a parse error and this FormatParser should be abandoned.
impl<'a> Iterator for FormatParser<'a> {
type Item = Option<FormatText>;
fn next(&mut self) -> Option<Option<FormatText>> {
// If the parser has been cancelled or errored for some reason.
if self.finished { return None }
// Try to parse a new FormatText.
match self.chars.next() {
// Parse a recognized object.
//
// The allowed forms are:
// - {method}
// - {uri}
// - {status}
// - {response-time}
// - {ip-addr}
// - {request-time}
Some('{') => {
self.object_buffer.clear();
let mut chr = self.chars.next();
while chr != None {
match chr.unwrap() {
// Finished parsing, parse buffer.
'}' => break,
c => self.object_buffer.push(c.clone())
}
chr = self.chars.next();
}
let text = match self.object_buffer.as_ref() {
"method" => Method,
"uri" => URI,
"status" => Status,
"response-time" => ResponseTime,
"request-time" => RequestTime,
"ip-addr" => RemoteAddr,
_ => {
// Error, so mark as finished.
self.finished = true;
return Some(None);
}
};
Some(Some(text))
}
// Parse a regular string part of the format string.
Some(c) => {
let mut buffer = String::new();
buffer.push(c);
loop {
match self.chars.peek() {
// Done parsing.
Some(&'{') | None => return Some(Some(FormatText::Str(buffer))),
Some(_) => {
buffer.push(self.chars.next().unwrap())
}
}
}
},
// Reached end of the format string.
None => None
}
}
}
/// A string of text to be logged. This is either one of the data
/// fields supported by the `Logger`, or a custom `String`.
#[derive(Clone)]
#[doc(hidden)]
pub enum FormatText {
Str(String),
Method,
URI,
Status,
ResponseTime,
RemoteAddr,
RequestTime
}
pub(crate) struct FormatDisplay<'a> {
format: &'a Format,
render: &'a Fn(&mut Formatter, &FormatText) -> Result<(), fmt::Error>,
}
impl<'a> fmt::Display for FormatDisplay<'a> {
fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
let Format(ref format) = *self.format;
for unit in format {
(self.render)(fmt, unit)?;
}
Ok(())
}
}

6
src/resource.rs
View File

@ -58,7 +58,7 @@ impl<S> Resource<S> where S: 'static {
/// Register handler for specified method.
pub fn handler<F, R>(&mut self, method: Method, handler: F)
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Into<HttpResponse> + 'static,
{
self.routes.insert(method, Box::new(FnHandler::new(handler)));
@ -66,7 +66,7 @@ impl<S> Resource<S> where S: 'static {
/// Register async handler for specified method.
pub fn async<F, R>(&mut self, method: Method, handler: F)
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Stream<Item=Frame, Error=()> + 'static,
{
self.routes.insert(method, Box::new(StreamHandler::new(handler)));
@ -119,7 +119,7 @@ impl<S> Resource<S> where S: 'static {
impl<S: 'static> RouteHandler<S> for Resource<S> {
fn handle(&self, req: HttpRequest, payload: Payload, state: Rc<S>) -> Task {
fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task {
if let Some(handler) = self.routes.get(req.method()) {
handler.handle(req, payload, state)
} else {

23
src/route.rs
View File

@ -27,7 +27,7 @@ pub enum Frame {
#[allow(unused_variables)]
pub trait RouteHandler<S>: 'static {
/// Handle request
fn handle(&self, req: HttpRequest, payload: Payload, state: Rc<S>) -> Task;
fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task;
/// Set route prefix
fn set_prefix(&mut self, prefix: String) {}
@ -73,7 +73,8 @@ pub trait Route: Actor {
/// request/response or websocket connection.
/// In that case `HttpContext::start` and `HttpContext::write` has to be used
/// for writing response.
fn request(req: HttpRequest, payload: Payload, ctx: &mut Self::Context) -> Reply<Self>;
fn request(req: &mut HttpRequest,
payload: Payload, ctx: &mut Self::Context) -> Reply<Self>;
/// This method creates `RouteFactory` for this actor.
fn factory() -> RouteFactory<Self, Self::State> {
@ -88,7 +89,7 @@ impl<A, S> RouteHandler<S> for RouteFactory<A, S>
where A: Actor<Context=HttpContext<A>> + Route<State=S>,
S: 'static
{
fn handle(&self, req: HttpRequest, payload: Payload, state: Rc<A::State>) -> Task
fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<A::State>) -> Task
{
let mut ctx = HttpContext::new(state);
@ -105,7 +106,7 @@ impl<A, S> RouteHandler<S> for RouteFactory<A, S>
/// Fn() route handler
pub(crate)
struct FnHandler<S, R, F>
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Into<HttpResponse>,
S: 'static,
{
@ -114,7 +115,7 @@ struct FnHandler<S, R, F>
}
impl<S, R, F> FnHandler<S, R, F>
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Into<HttpResponse> + 'static,
S: 'static,
{
@ -124,11 +125,11 @@ impl<S, R, F> FnHandler<S, R, F>
}
impl<S, R, F> RouteHandler<S> for FnHandler<S, R, F>
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Into<HttpResponse> + 'static,
S: 'static,
{
fn handle(&self, req: HttpRequest, payload: Payload, state: Rc<S>) -> Task
fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task
{
Task::reply((self.f)(req, payload, &state).into())
}
@ -137,7 +138,7 @@ impl<S, R, F> RouteHandler<S> for FnHandler<S, R, F>
/// Async route handler
pub(crate)
struct StreamHandler<S, R, F>
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Stream<Item=Frame, Error=()> + 'static,
S: 'static,
{
@ -146,7 +147,7 @@ struct StreamHandler<S, R, F>
}
impl<S, R, F> StreamHandler<S, R, F>
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Stream<Item=Frame, Error=()> + 'static,
S: 'static,
{
@ -156,11 +157,11 @@ impl<S, R, F> StreamHandler<S, R, F>
}
impl<S, R, F> RouteHandler<S> for StreamHandler<S, R, F>
where F: Fn(HttpRequest, Payload, &S) -> R + 'static,
where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
R: Stream<Item=Frame, Error=()> + 'static,
S: 'static,
{
fn handle(&self, req: HttpRequest, payload: Payload, state: Rc<S>) -> Task
fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task
{
Task::with_stream(
(self.f)(req, payload, &state).map_err(

22
src/server.rs
View File

@ -1,5 +1,6 @@
use std::{io, mem, net};
use std::{io, net};
use std::rc::Rc;
use std::cell::UnsafeCell;
use std::time::Duration;
use std::marker::PhantomData;
use std::collections::VecDeque;
@ -10,7 +11,7 @@ use tokio_core::reactor::Timeout;
use tokio_core::net::{TcpListener, TcpStream};
use tokio_io::{AsyncRead, AsyncWrite};
use task::{Task, RequestInfo};
use task::Task;
use reader::{Reader, ReaderError};
use payload::Payload;
use httpcodes::HTTPNotFound;
@ -21,7 +22,7 @@ pub trait HttpHandler: 'static {
/// Http handler prefix
fn prefix(&self) -> &str;
/// Handle request
fn handle(&self, req: HttpRequest, payload: Payload) -> Task;
fn handle(&self, req: &mut HttpRequest, payload: Payload) -> Task;
}
/// An HTTP Server
@ -148,7 +149,7 @@ impl<T, A, H> Handler<IoStream<T, A>, io::Error> for HttpServer<T, A, H>
struct Entry {
task: Task,
req: RequestInfo,
req: UnsafeCell<HttpRequest>,
eof: bool,
error: bool,
finished: bool,
@ -213,9 +214,7 @@ impl<T, A, H> Future for HttpChannel<T, A, H>
}
// this is anoying
let req: &RequestInfo = unsafe {
mem::transmute(&self.items[idx].req)
};
let req = unsafe {self.items[idx].req.get().as_mut().unwrap()};
match self.items[idx].task.poll_io(&mut self.stream, req)
{
Ok(Async::Ready(ready)) => {
@ -280,23 +279,22 @@ impl<T, A, H> Future for HttpChannel<T, A, H>
// read incoming data
if !self.error && self.items.len() < MAX_PIPELINED_MESSAGES {
match self.reader.parse(&mut self.stream) {
Ok(Async::Ready((req, payload))) => {
Ok(Async::Ready((mut req, payload))) => {
// stop keepalive timer
self.keepalive_timer.take();
// start request processing
let info = RequestInfo::new(&req);
let mut task = None;
for h in self.router.iter() {
if req.path().starts_with(h.prefix()) {
task = Some(h.handle(req, payload));
task = Some(h.handle(&mut req, payload));
break
}
}
self.items.push_back(
Entry {task: task.unwrap_or_else(|| Task::reply(HTTPNotFound)),
req: info,
req: UnsafeCell::new(req),
eof: false,
error: false,
finished: false});
@ -313,7 +311,7 @@ impl<T, A, H> Future for HttpChannel<T, A, H>
if let ReaderError::Error(err) = err {
self.items.push_back(
Entry {task: Task::reply(err),
req: RequestInfo::for_error(),
req: UnsafeCell::new(HttpRequest::for_error()),
eof: false,
error: false,
finished: false});

11
src/staticfiles.rs
View File

@ -48,14 +48,19 @@ impl StaticFiles {
pub fn new<D: Into<PathBuf>>(dir: D, index: bool) -> StaticFiles {
let dir = dir.into();
let (dir, access) = if let Ok(dir) = dir.canonicalize() {
let (dir, access) = match dir.canonicalize() {
Ok(dir) => {
if dir.is_dir() {
(dir, true)
} else {
warn!("Is not directory `{:?}`", dir);
(dir, false)
}
} else {
},
Err(err) => {
warn!("Static files directory `{:?}` error: {}", dir, err);
(dir, false)
}
};
StaticFiles {
@ -134,7 +139,7 @@ impl<S: 'static> RouteHandler<S> for StaticFiles {
}
}
fn handle(&self, req: HttpRequest, payload: Payload, state: Rc<S>) -> Task {
fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task {
if !self.accessible {
Task::reply(HTTPNotFound)
} else {

72
src/task.rs
View File

@ -1,4 +1,5 @@
use std::{cmp, io};
use std::rc::Rc;
use std::fmt::Write;
use std::collections::VecDeque;
@ -11,6 +12,7 @@ use tokio_io::{AsyncRead, AsyncWrite};
use date;
use route::Frame;
use application::Middleware;
use httprequest::HttpRequest;
use httpresponse::{Body, HttpResponse};
@ -44,26 +46,6 @@ impl TaskIOState {
}
}
pub(crate) struct RequestInfo {
version: Version,
keep_alive: bool,
}
impl RequestInfo {
pub fn new(req: &HttpRequest) -> Self {
RequestInfo {
version: req.version(),
keep_alive: req.keep_alive(),
}
}
pub fn for_error() -> Self {
RequestInfo {
version: Version::HTTP_11,
keep_alive: false,
}
}
}
pub struct Task {
state: TaskRunningState,
iostate: TaskIOState,
@ -73,7 +55,8 @@ pub struct Task {
buffer: BytesMut,
upgrade: bool,
keepalive: bool,
prepared: bool,
prepared: Option<HttpResponse>,
middlewares: Option<Rc<Vec<Box<Middleware>>>>,
}
impl Task {
@ -92,7 +75,8 @@ impl Task {
buffer: BytesMut::new(),
upgrade: false,
keepalive: false,
prepared: false,
prepared: None,
middlewares: None,
}
}
@ -108,7 +92,8 @@ impl Task {
buffer: BytesMut::new(),
upgrade: false,
keepalive: false,
prepared: false,
prepared: None,
middlewares: None,
}
}
@ -116,15 +101,31 @@ impl Task {
self.keepalive && !self.upgrade
}
fn prepare(&mut self, req: &RequestInfo, mut msg: HttpResponse)
pub(crate) fn set_middlewares(&mut self, middlewares: Rc<Vec<Box<Middleware>>>) {
self.middlewares = Some(middlewares);
}
fn prepare(&mut self, req: &mut HttpRequest, msg: HttpResponse)
{
trace!("Prepare message status={:?}", msg.status);
// run middlewares
let mut msg = if let Some(middlewares) = self.middlewares.take() {
let mut msg = msg;
for middleware in middlewares.iter() {
msg = middleware.response(req, msg);
}
self.middlewares = Some(middlewares);
msg
} else {
msg
};
// prepare task
let mut extra = 0;
let body = msg.replace_body(Body::Empty);
let version = msg.version().unwrap_or_else(|| req.version);
self.keepalive = msg.keep_alive().unwrap_or_else(|| req.keep_alive);
self.prepared = true;
let version = msg.version().unwrap_or_else(|| req.version());
self.keepalive = msg.keep_alive().unwrap_or_else(|| req.keep_alive());
match body {
Body::Empty => {
@ -219,12 +220,14 @@ impl Task {
if let Body::Binary(ref bytes) = body {
self.buffer.extend(bytes);
self.prepared = Some(msg);
return
}
msg.replace_body(body);
self.prepared = Some(msg);
}
pub(crate) fn poll_io<T>(&mut self, io: &mut T, info: &RequestInfo) -> Poll<bool, ()>
pub(crate) fn poll_io<T>(&mut self, io: &mut T, req: &mut HttpRequest) -> Poll<bool, ()>
where T: AsyncRead + AsyncWrite
{
trace!("POLL-IO frames:{:?}", self.frames.len());
@ -248,10 +251,10 @@ impl Task {
trace!("IO Frame: {:?}", frame);
match frame {
Frame::Message(response) => {
self.prepare(info, response);
self.prepare(req, response);
}
Frame::Payload(Some(chunk)) => {
if self.prepared {
if self.prepared.is_some() {
// TODO: add warning, write after EOF
self.encoder.encode(&mut self.buffer, chunk.as_ref());
} else {
@ -295,6 +298,15 @@ impl Task {
// response is completed
if self.buffer.is_empty() && self.iostate.is_done() {
// run middlewares
if let Some(ref mut resp) = self.prepared {
if let Some(middlewares) = self.middlewares.take() {
for middleware in middlewares.iter() {
middleware.finish(req, resp);
}
}
}
Ok(Async::Ready(self.state.is_done()))
} else {
Ok(Async::NotReady)