actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-23 23:51:06 +01:00
Code Issues Releases Wiki Activity

cleanup warnings

Browse Source
This commit is contained in:
Nikolay Kim 2018-07-04 21:01:27 +06:00
parent 4c5a63965e
commit 6fd686ef98
39 changed files with 116 additions and 226 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.toml
View File

@ -50,7 +50,7 @@ flate2-c = ["flate2/miniz-sys"]
flate2-rust = ["flate2/rust_backend"]
[dependencies]
# actix = "0.6.1"
# actix = "0.7.0"
actix = { git="https://github.com/actix/actix.git" }
base64 = "0.9"

20
src/application.rs
View File

@ -7,13 +7,12 @@ use http::{Method, StatusCode};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::Middleware;
use param::Params;
use pipeline::{HandlerType, Pipeline, PipelineHandler};
use pred::Predicate;
use resource::ResourceHandler;
use router::{Resource, RouteInfo, Router};
use scope::Scope;
use server::{HttpHandler, HttpHandlerTask, IntoHttpHandler, Request, ServerSettings};
use server::{HttpHandler, HttpHandlerTask, IntoHttpHandler, Request};
/// Application
pub struct HttpApplication<S = ()> {
@ -102,23 +101,20 @@ impl<S: 'static> HttpApplication<S> {
continue 'outer;
}
}
let info = self
.router
.route_info_params(params, self.inner.prefix as u16);
return (info, HandlerType::Handler(idx));
return (
self.router.route_info_params(params),
HandlerType::Handler(idx),
);
}
}
}
}
(
self.router.default_route_info(self.inner.prefix as u16),
HandlerType::Default,
)
(self.router.default_route_info(), HandlerType::Default)
}
}
#[cfg(test)]
pub(crate) fn run(&self, mut req: Request) -> AsyncResult<HttpResponse> {
pub(crate) fn run(&self, req: Request) -> AsyncResult<HttpResponse> {
let (info, tp) = self.get_handler(&req);
let req = HttpRequest::new(req, Rc::clone(&self.state), info);
@ -129,7 +125,7 @@ impl<S: 'static> HttpApplication<S> {
impl<S: 'static> HttpHandler for HttpApplication<S> {
type Task = Pipeline<S, Inner<S>>;
fn handle(&self, mut msg: Request) -> Result<Pipeline<S, Inner<S>>, Request> {
fn handle(&self, msg: Request) -> Result<Pipeline<S, Inner<S>>, Request> {
let m = {
let path = msg.path();
path.starts_with(&self.prefix)

20
src/client/connector.rs
View File

@ -5,8 +5,8 @@ use std::{fmt, io, mem, time};
use actix::resolver::{Connect as ResolveConnect, Resolver, ResolverError};
use actix::{
fut, Actor, ActorContext, ActorFuture, ActorResponse, Addr, AsyncContext, Context,
ContextFutureSpawner, Handler, Message, Recipient, StreamHandler2, Supervised,
fut, Actor, ActorFuture, ActorResponse, Addr, AsyncContext, Context,
ContextFutureSpawner, Handler, Message, Recipient, StreamHandler, Supervised,
SystemService, WrapFuture,
};
@ -220,7 +220,7 @@ impl Actor for ClientConnector {
self.resolver = Some(Resolver::from_registry())
}
self.collect_periodic(ctx);
ctx.add_stream2(self.acq_rx.take().unwrap());
ctx.add_stream(self.acq_rx.take().unwrap());
ctx.spawn(Maintenance);
}
}
@ -769,20 +769,17 @@ impl Handler<Connect> for ClientConnector {
}
}
impl StreamHandler2<AcquiredConnOperation, ()> for ClientConnector {
fn handle(
&mut self, msg: Result<Option<AcquiredConnOperation>, ()>,
ctx: &mut Context<Self>,
) {
impl StreamHandler<AcquiredConnOperation, ()> for ClientConnector {
fn handle(&mut self, msg: AcquiredConnOperation, _: &mut Context<Self>) {
let now = Instant::now();
match msg {
Ok(Some(AcquiredConnOperation::Close(conn))) => {
AcquiredConnOperation::Close(conn) => {
self.release_key(&conn.key);
self.to_close.push(conn);
self.stats.closed += 1;
}
Ok(Some(AcquiredConnOperation::Release(conn))) => {
AcquiredConnOperation::Release(conn) => {
self.release_key(&conn.key);
// check connection lifetime and the return to available pool
@ -793,10 +790,9 @@ impl StreamHandler2<AcquiredConnOperation, ()> for ClientConnector {
.push_back(Conn(Instant::now(), conn));
}
}
Ok(Some(AcquiredConnOperation::ReleaseKey(key))) => {
AcquiredConnOperation::ReleaseKey(key) => {
self.release_key(&key);
}
_ => ctx.stop(),
}
// check keep-alive

4
src/client/response.rs
View File

@ -1,13 +1,11 @@
use std::cell::RefCell;
use std::{fmt, str};
use bytes::Bytes;
use cookie::Cookie;
use futures::{Async, Poll, Stream};
use http::header::{self, HeaderValue};
use http::{HeaderMap, StatusCode, Version};
use error::{CookieParseError, PayloadError};
use error::CookieParseError;
use httpmessage::HttpMessage;
use super::pipeline::Pipeline;

2
src/context.rs
View File

@ -105,7 +105,7 @@ where
{
#[inline]
/// Create a new HTTP Context from a request and an actor
pub fn new(req: HttpRequest<S>, actor: A) -> Body {
pub fn create(req: HttpRequest<S>, actor: A) -> Body {
let mb = Mailbox::default();
let ctx = HttpContext {
inner: ContextParts::new(mb.sender_producer()),

2
src/extractor.rs
View File

@ -768,7 +768,7 @@ mod tests {
routes.push((Resource::new("index", "/{key}/{value}/"), Some(resource)));
let (router, _) = Router::new("", routes);
let mut req = TestRequest::with_uri("/name/user1/?id=test")
let req = TestRequest::with_uri("/name/user1/?id=test")
.finish_with_router(router.clone());
let info = router.recognize(&req).unwrap().1;
let req = req.with_route_info(info);

5
src/fs.rs
View File

@ -2,7 +2,6 @@
use std::fmt::Write;
use std::fs::{DirEntry, File, Metadata};
use std::io::{Read, Seek};
use std::marker::PhantomData;
use std::ops::{Deref, DerefMut};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
@ -20,9 +19,7 @@ use mime_guess::{get_mime_type, guess_mime_type};
use percent_encoding::{utf8_percent_encode, DEFAULT_ENCODE_SET};
use error::Error;
use handler::{
AsyncResult, AsyncResultItem, Handler, Responder, RouteHandler, WrapHandler,
};
use handler::{AsyncResult, Handler, Responder, RouteHandler, WrapHandler};
use header;
use http::{ContentEncoding, Method, StatusCode};
use httpmessage::HttpMessage;

1
src/handler.rs
View File

@ -10,7 +10,6 @@ use http::StatusCode;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use resource::ResourceHandler;
use server::Request;
/// Trait defines object that could be registered as route handler
#[allow(unused_variables)]

10
src/httpmessage.rs
View File

@ -16,7 +16,6 @@ use error::{
use header::Header;
use json::JsonBody;
use multipart::Multipart;
use payload::Payload;
/// Trait that implements general purpose operations on http messages
pub trait HttpMessage: Sized {
@ -613,10 +612,7 @@ mod tests {
use encoding::all::ISO_8859_2;
use encoding::Encoding;
use futures::Async;
use http::{Method, Uri, Version};
use httprequest::HttpRequest;
use mime;
use std::str::FromStr;
use test::TestRequest;
#[test]
@ -765,7 +761,7 @@ mod tests {
#[test]
fn test_urlencoded() {
let mut req = TestRequest::with_header(
let req = TestRequest::with_header(
header::CONTENT_TYPE,
"application/x-www-form-urlencoded",
).header(header::CONTENT_LENGTH, "11")
@ -780,7 +776,7 @@ mod tests {
})
);
let mut req = TestRequest::with_header(
let req = TestRequest::with_header(
header::CONTENT_TYPE,
"application/x-www-form-urlencoded; charset=utf-8",
).header(header::CONTENT_LENGTH, "11")
@ -818,7 +814,7 @@ mod tests {
_ => unreachable!("error"),
}
let mut req = TestRequest::default()
let req = TestRequest::default()
.set_payload(Bytes::from_static(b"11111111111111"))
.finish();
match req.body().limit(5).poll().err().unwrap() {

38
src/httprequest.rs
View File

@ -1,22 +1,18 @@
//! HTTP Request message related code.
use std::cell::{Cell, Ref, RefCell, RefMut};
use std::cell::{Ref, RefMut};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::ops::Deref;
use std::rc::Rc;
use std::{cmp, fmt, io, str};
use std::{fmt, str};
use bytes::Bytes;
use cookie::Cookie;
use failure;
use futures::{Async, Poll, Stream};
use futures_cpupool::CpuPool;
use http::{header, HeaderMap, Method, StatusCode, Uri, Version};
use tokio_io::AsyncRead;
use url::{form_urlencoded, Url};
use body::Body;
use error::{CookieParseError, PayloadError, UrlGenerationError};
use error::{CookieParseError, UrlGenerationError};
use extensions::Extensions;
use handler::FromRequest;
use httpmessage::HttpMessage;
@ -24,19 +20,12 @@ use httpresponse::{HttpResponse, HttpResponseBuilder};
use info::ConnectionInfo;
use param::Params;
use payload::Payload;
use router::{Resource, RouteInfo, Router};
use server::message::{MessageFlags, Request};
use uri::Url as InnerUrl;
use router::{Resource, RouteInfo};
use server::Request;
struct Query(HashMap<String, String>);
struct Cookies(Vec<Cookie<'static>>);
#[derive(Debug, Copy, Clone, PartialEq)]
pub(crate) enum RouterResource {
Notset,
Normal(u16),
}
/// An HTTP Request
pub struct HttpRequest<S = ()> {
req: Rc<Request>,
@ -167,11 +156,6 @@ impl<S> HttpRequest<S> {
self.request().inner.url.path()
}
#[inline]
pub(crate) fn url(&self) -> &InnerUrl {
&self.request().inner.url
}
/// Get *ConnectionInfo* for the correct request.
#[inline]
pub fn connection_info(&self) -> Ref<ConnectionInfo> {
@ -248,7 +232,6 @@ impl<S> HttpRequest<S> {
for (key, val) in form_urlencoded::parse(self.query_string().as_ref()) {
query.insert(key.as_ref().to_string(), val.to_string());
}
let mut req = self.clone();
self.extensions_mut().insert(Query(query));
}
Ref::map(self.extensions(), |ext| &ext.get::<Query>().unwrap().0)
@ -270,7 +253,6 @@ impl<S> HttpRequest<S> {
#[inline]
pub fn cookies(&self) -> Result<Ref<Vec<Cookie<'static>>>, CookieParseError> {
if self.extensions().get::<Cookies>().is_none() {
let mut req = self.clone();
let mut cookies = Vec::new();
for hdr in self.request().inner.headers.get_all(header::COOKIE) {
let s = str::from_utf8(hdr.as_bytes()).map_err(CookieParseError::from)?;
@ -379,7 +361,7 @@ impl<S> fmt::Debug for HttpRequest<S> {
mod tests {
use super::*;
use resource::ResourceHandler;
use router::Resource;
use router::{Resource, Router};
use test::TestRequest;
#[test]
@ -448,7 +430,7 @@ mod tests {
let routes =
vec![(Resource::new("index", "/user/{name}.{ext}"), Some(resource))];
let (router, _) = Router::new("/", routes);
let info = router.default_route_info(0);
let info = router.default_route_info();
assert!(info.has_route("/user/test.html"));
assert!(!info.has_route("/test/unknown"));
@ -476,7 +458,7 @@ mod tests {
resource.name("index");
let routes = vec![(Resource::new("index", "/user/{name}.html"), Some(resource))];
let (router, _) = Router::new("/prefix/", routes);
let info = router.default_route_info(0);
let info = router.default_route_info();
assert!(info.has_route("/user/test.html"));
assert!(!info.has_route("/prefix/user/test.html"));
@ -495,7 +477,7 @@ mod tests {
resource.name("index");
let routes = vec![(Resource::new("index", "/index.html"), Some(resource))];
let (router, _) = Router::new("/prefix/", routes);
let info = router.default_route_info(0);
let info = router.default_route_info();
assert!(info.has_route("/index.html"));
assert!(!info.has_route("/prefix/index.html"));
@ -518,7 +500,7 @@ mod tests {
None,
)];
let router = Router::new::<()>("", routes).0;
let info = router.default_route_info(0);
let info = router.default_route_info();
assert!(!info.has_route("https://youtube.com/watch/unknown"));
let req = TestRequest::default().finish_with_router(router);

2
src/httpresponse.rs
View File

@ -1042,8 +1042,6 @@ mod tests {
use body::Binary;
use http;
use http::header::{HeaderValue, CONTENT_TYPE, COOKIE};
use http::{Method, Uri};
use std::str::FromStr;
use time::Duration;
use test::TestRequest;

8
src/info.rs
View File

@ -1,9 +1,4 @@
use std::rc::Rc;
use std::str::FromStr;
use http::header::{self, HeaderName};
use httpmessage::HttpMessage;
use httprequest::HttpRequest;
use server::Request;
const X_FORWARDED_FOR: &[u8] = b"x-forwarded-for";
@ -165,7 +160,6 @@ impl ConnectionInfo {
#[cfg(test)]
mod tests {
use super::*;
use http::header::HeaderValue;
use test::TestRequest;
#[test]
@ -214,7 +208,7 @@ mod tests {
assert_eq!(info.host(), "192.0.2.60");
assert_eq!(info.remote(), None);
let mut req = TestRequest::default()
let req = TestRequest::default()
.header(X_FORWARDED_PROTO, "https")
.request();
let mut info = ConnectionInfo::default();

4
src/json.rs
View File

@ -1,4 +1,4 @@
use bytes::{Bytes, BytesMut};
use bytes::BytesMut;
use futures::{Future, Poll, Stream};
use http::header::CONTENT_LENGTH;
use std::fmt;
@ -10,7 +10,7 @@ use serde::de::DeserializeOwned;
use serde::Serialize;
use serde_json;
use error::{Error, JsonPayloadError, PayloadError};
use error::{Error, JsonPayloadError};
use handler::{FromRequest, Responder};
use http::StatusCode;
use httpmessage::HttpMessage;

1
src/middleware/csrf.rs
View File

@ -50,7 +50,6 @@ use std::collections::HashSet;
use bytes::Bytes;
use error::{ResponseError, Result};
use http::{header, HeaderMap, HttpTryFrom, Uri};
use httpmessage::HttpMessage;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Middleware, Started};

5
src/middleware/errhandlers.rs
View File

@ -87,10 +87,7 @@ mod tests {
use http::StatusCode;
use httpmessage::HttpMessage;
use middleware::Started;
use test;
use server::Request;
use test::TestRequest;
use test::{self, TestRequest};
fn render_500<S>(_: &HttpRequest<S>, resp: HttpResponse) -> Result<Response> {
let mut builder = resp.into_builder();

6
src/middleware/identity.rs
View File

@ -46,7 +46,6 @@
//! ));
//! }
//! ```
use std::cell::RefCell;
use std::rc::Rc;
use cookie::{Cookie, CookieJar, Key};
@ -59,7 +58,6 @@ use http::header::{self, HeaderValue};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Middleware, Response, Started};
use server::Request;
/// The helper trait to obtain your identity from a request.
///
@ -109,13 +107,13 @@ impl<S> RequestIdentity for HttpRequest<S> {
}
fn remember(&self, identity: String) {
if let Some(mut id) = self.extensions_mut().get_mut::<IdentityBox>() {
if let Some(id) = self.extensions_mut().get_mut::<IdentityBox>() {
return id.0.as_mut().remember(identity);
}
}
fn forget(&self) {
if let Some(mut id) = self.extensions_mut().get_mut::<IdentityBox>() {
if let Some(id) = self.extensions_mut().get_mut::<IdentityBox>() {
return id.0.forget();
}
}

10
src/middleware/logger.rs
View File

@ -11,7 +11,6 @@ use httpmessage::HttpMessage;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Finished, Middleware, Started};
use server::Request;
/// `Middleware` for logging request and response info to the terminal.
///
@ -309,13 +308,12 @@ impl<'a> fmt::Display for FormatDisplay<'a> {
#[cfg(test)]
mod tests {
use super::*;
use http::header::{self, HeaderMap};
use http::{Method, StatusCode, Uri, Version};
use std::str::FromStr;
use test::TestRequest;
use time;
use super::*;
use http::{header, StatusCode};
use test::TestRequest;
#[test]
fn test_logger() {
let logger = Logger::new("%% %{User-Agent}i %{X-Test}o %{HOME}e %D test");

1
src/middleware/mod.rs
View File

@ -4,7 +4,6 @@ use futures::Future;
use error::{Error, Result};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use server::Request;
mod logger;

1
src/middleware/session.rs
View File

@ -83,7 +83,6 @@ use handler::FromRequest;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Middleware, Response, Started};
use server::Request;
/// The helper trait to obtain your session data from a request.
///

4
src/param.rs
View File

@ -57,10 +57,6 @@ impl Params {
self.segments.clear();
}
pub(crate) fn set_url(&mut self, url: Url) {
self.url = url;
}
pub(crate) fn set_tail(&mut self, tail: u16) {
self.tail = tail;
}

14
src/pipeline.rs
View File

@ -6,7 +6,6 @@ use futures::sync::oneshot;
use futures::{Async, Future, Poll, Stream};
use log::Level::Debug;
use application::Inner;
use body::{Body, BodyStream};
use context::{ActorHttpContext, Frame};
use error::Error;
@ -15,7 +14,7 @@ use header::ContentEncoding;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::{Finished, Middleware, Response, Started};
use server::{HttpHandlerTask, Request, Writer, WriterState};
use server::{HttpHandlerTask, Writer, WriterState};
#[doc(hidden)]
#[derive(Debug, Clone, Copy)]
@ -84,17 +83,6 @@ struct PipelineInfo<S: 'static> {
}
impl<S: 'static> PipelineInfo<S> {
fn new(req: HttpRequest<S>) -> PipelineInfo<S> {
PipelineInfo {
req,
count: 0,
error: None,
context: None,
disconnected: None,
encoding: ContentEncoding::Auto,
}
}
fn poll_context(&mut self) -> Poll<(), Error> {
if let Some(ref mut context) = self.context {
match context.poll() {

7
src/pred.rs
View File

@ -4,8 +4,6 @@ use std::marker::PhantomData;
use http;
use http::{header, HttpTryFrom};
use httpmessage::HttpMessage;
use httprequest::HttpRequest;
use server::message::Request;
/// Trait defines resource route predicate.
@ -239,11 +237,8 @@ impl<S: 'static> Predicate<S> for HostPredicate<S> {
#[cfg(test)]
mod tests {
use std::str::FromStr;
use super::*;
use http::header::{self, HeaderMap};
use http::{Method, Uri, Version};
use http::{header, Method};
use test::TestRequest;
#[test]

1
src/resource.rs
View File

@ -12,7 +12,6 @@ use httpresponse::HttpResponse;
use middleware::Middleware;
use pred;
use route::Route;
use server::Request;
#[derive(Copy, Clone)]
pub(crate) struct RouteId(usize);

1
src/route.rs
View File

@ -16,7 +16,6 @@ use middleware::{
Started as MiddlewareStarted,
};
use pred::Predicate;
use server::Request;
use with::{With, WithAsync};
/// Resource route definition

24
src/router.rs
View File

@ -3,11 +3,9 @@ use std::hash::{Hash, Hasher};
use std::rc::Rc;
use regex::{escape, Regex};
use smallvec::SmallVec;
use url::Url;
use error::UrlGenerationError;
use httprequest::HttpRequest;
use param::{ParamItem, Params};
use resource::ResourceHandler;
use server::Request;
@ -25,7 +23,6 @@ pub struct Router(Rc<Inner>);
pub struct RouteInfo {
router: Rc<Inner>,
resource: RouterResource,
prefix: u16,
params: Params,
}
@ -51,14 +48,8 @@ impl RouteInfo {
&self.params
}
#[doc(hidden)]
#[inline]
pub fn prefix_len(&self) -> u16 {
self.prefix
}
#[inline]
pub(crate) fn merge(&self, mut params: Params) -> RouteInfo {
pub(crate) fn merge(&self, params: &Params) -> RouteInfo {
let mut p = self.params.clone();
p.set_tail(params.tail);
for item in &params.segments {
@ -69,7 +60,6 @@ impl RouteInfo {
params: p,
router: self.router.clone(),
resource: self.resource,
prefix: self.prefix,
}
}
@ -167,10 +157,6 @@ impl Router {
&self.0.prefix
}
pub(crate) fn get_resource(&self, idx: usize) -> &Resource {
&self.0.patterns[idx]
}
pub(crate) fn route_info(&self, req: &Request, prefix: u16) -> RouteInfo {
let mut params = Params::with_url(req.url());
params.set_tail(prefix);
@ -179,22 +165,19 @@ impl Router {
params,
router: self.0.clone(),
resource: RouterResource::Notset,
prefix: 0,
}
}
pub(crate) fn route_info_params(&self, params: Params, prefix: u16) -> RouteInfo {
pub(crate) fn route_info_params(&self, params: Params) -> RouteInfo {
RouteInfo {
params,
prefix,
router: self.0.clone(),
resource: RouterResource::Notset,
}
}
pub(crate) fn default_route_info(&self, prefix: u16) -> RouteInfo {
pub(crate) fn default_route_info(&self) -> RouteInfo {
RouteInfo {
prefix,
router: self.0.clone(),
resource: RouterResource::Notset,
params: Params::new(),
@ -215,7 +198,6 @@ impl Router {
params,
router: self.0.clone(),
resource: RouterResource::Normal(idx as u16),
prefix: self.0.prefix_len as u16,
},
));
}

5
src/scope.rs
View File

@ -341,7 +341,7 @@ impl<S: 'static> RouteHandler<S> for Scope<S> {
// recognize resources
for &(ref pattern, ref resource) in self.resources.iter() {
if let Some(params) = pattern.match_with_params(req, tail, false) {
let req2 = req.with_route_info(req.route().merge(params));
let req2 = req.with_route_info(req.route().merge(&params));
if let Some(id) = resource.get_route_id(&req2) {
if self.middlewares.is_empty() {
return resource.handle(id, &req2);
@ -358,10 +358,9 @@ impl<S: 'static> RouteHandler<S> for Scope<S> {
}
// nested scopes
let len = req.route().prefix_len() as usize;
'outer: for &(ref prefix, ref handler, ref filters) in &self.nested {
if let Some(params) = prefix.match_prefix_with_params(req, tail) {
let req2 = req.with_route_info(req.route().merge(params));
let req2 = req.with_route_info(req.route().merge(&params));
let state = req.state();
for filter in filters {

3
src/server/error.rs
View File

@ -2,7 +2,6 @@ use futures::{Async, Poll};
use super::{helpers, HttpHandlerTask, Writer};
use http::{StatusCode, Version};
use httpresponse::HttpResponse;
use Error;
pub(crate) struct ServerError(Version, StatusCode);
@ -16,7 +15,7 @@ impl ServerError {
impl HttpHandlerTask for ServerError {
fn poll_io(&mut self, io: &mut Writer) -> Poll<bool, Error> {
{
let mut bytes = io.buffer();
let bytes = io.buffer();
helpers::write_status_line(self.0, self.1.as_u16(), bytes);
}
io.set_date();

3
src/server/h1.rs
View File

@ -9,10 +9,7 @@ use tokio_timer::Delay;
use error::{Error, PayloadError};
use http::{StatusCode, Version};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use payload::{Payload, PayloadStatus, PayloadWriter};
use pipeline::Pipeline;
use super::error::ServerError;
use super::h1decoder::{DecoderError, H1Decoder, Message};

6
src/server/h1writer.rs
View File

@ -149,20 +149,18 @@ impl<T: AsyncWrite, H: 'static> Writer for H1Writer<T, H> {
let mut buffer = self.buffer.as_mut();
let reason = msg.reason().as_bytes();
let mut is_bin = if let Body::Binary(ref bytes) = body {
if let Body::Binary(ref bytes) = body {
buffer.reserve(
256
+ msg.headers().len() * AVERAGE_HEADER_SIZE
+ bytes.len()
+ reason.len(),
);
true
} else {
buffer.reserve(
256 + msg.headers().len() * AVERAGE_HEADER_SIZE + reason.len(),
);
false
};
}
// status line
helpers::write_status_line(version, msg.status().as_u16(), &mut buffer);

4
src/server/h2.rs
View File

@ -15,11 +15,7 @@ use tokio_timer::Delay;
use error::{Error, PayloadError};
use http::{StatusCode, Version};
use httpmessage::HttpMessage;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use payload::{Payload, PayloadStatus, PayloadWriter};
use pipeline::Pipeline;
use uri::Url;
use super::error::ServerError;

4
src/server/message.rs
View File

@ -172,7 +172,8 @@ impl Request {
}
#[inline]
pub(crate) fn reset(&mut self) {
/// Reset request instance
pub fn reset(&mut self) {
self.inner.headers.clear();
self.inner.extensions.borrow_mut().clear();
self.inner.flags.set(MessageFlags::empty());
@ -210,6 +211,7 @@ impl RequestPool {
}
#[inline]
/// Release request instance
pub fn release(&self, mut msg: Request) {
let v = &mut self.0.borrow_mut();
if v.len() < 128 {

1
src/server/mod.rs
View File

@ -33,7 +33,6 @@ use actix::Message;
use body::Binary;
use error::Error;
use header::ContentEncoding;
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
/// max buffer size 64k

8
src/server/output.rs
View File

@ -10,12 +10,10 @@ use bytes::BytesMut;
use flate2::write::{DeflateEncoder, GzEncoder};
#[cfg(feature = "flate2")]
use flate2::Compression;
use http::header::{
HeaderValue, ACCEPT_ENCODING, CONTENT_ENCODING, CONTENT_LENGTH, TRANSFER_ENCODING,
};
use http::{HttpTryFrom, Method, Version};
use http::header::{ACCEPT_ENCODING, CONTENT_LENGTH};
use http::Version;
use super::message::{InnerRequest, Request};
use super::message::InnerRequest;
use body::{Binary, Body};
use header::ContentEncoding;
use httpresponse::HttpResponse;

5
src/server/settings.rs
View File

@ -11,7 +11,6 @@ use parking_lot::Mutex;
use time;
use super::channel::Node;
use super::helpers;
use super::message::{Request, RequestPool};
use super::KeepAlive;
use body::Body;
@ -161,7 +160,6 @@ pub(crate) struct WorkerSettings<H> {
channels: Cell<usize>,
node: Box<Node<()>>,
date: UnsafeCell<Date>,
settings: ServerSettings,
}
impl<H> WorkerSettings<H> {
@ -177,13 +175,12 @@ impl<H> WorkerSettings<H> {
WorkerSettings {
h: RefCell::new(h),
bytes: Rc::new(SharedBytesPool::new()),
messages: RequestPool::pool(settings.clone()),
messages: RequestPool::pool(settings),
channels: Cell::new(0),
node: Box::new(Node::head()),
date: UnsafeCell::new(Date::new()),
keep_alive,
ka_enabled,
settings,
}
}

86
src/server/srv.rs
View File

@ -5,7 +5,7 @@ use std::{io, net, thread};
use actix::{
fut, signal, Actor, ActorFuture, Addr, Arbiter, AsyncContext, Context, Handler,
Response, StreamHandler2, System, WrapFuture,
Response, StreamHandler, System, WrapFuture,
};
use futures::sync::mpsc;
@ -447,7 +447,7 @@ impl<H: IntoHttpHandler> HttpServer<H> {
// start http server actor
let signals = self.subscribe_to_signals();
let addr = Actor::create(move |ctx| {
ctx.add_stream2(rx);
ctx.add_stream(rx);
self
});
if let Some(signals) = signals {
@ -613,51 +613,55 @@ impl<H: IntoHttpHandler> Handler<signal::Signal> for HttpServer<H> {
}
/// Commands from accept threads
impl<H: IntoHttpHandler> StreamHandler2<ServerCommand, ()> for HttpServer<H> {
fn handle(&mut self, msg: Result<Option<ServerCommand>, ()>, _: &mut Context<Self>) {
if let Ok(Some(ServerCommand::WorkerDied(idx, socks))) = msg {
let mut found = false;
for i in 0..self.workers.len() {
if self.workers[i].0 == idx {
self.workers.swap_remove(i);
found = true;
break;
}
}
impl<H: IntoHttpHandler> StreamHandler<ServerCommand, ()> for HttpServer<H> {
fn finished(&mut self, _: &mut Context<Self>) {}
if found {
error!("Worker has died {:?}, restarting", idx);
let (tx, rx) = mpsc::unbounded::<Conn<net::TcpStream>>();
let mut new_idx = self.workers.len();
'found: loop {
for i in 0..self.workers.len() {
if self.workers[i].0 == new_idx {
new_idx += 1;
continue 'found;
}
fn handle(&mut self, msg: ServerCommand, _: &mut Context<Self>) {
match msg {
ServerCommand::WorkerDied(idx, socks) => {
let mut found = false;
for i in 0..self.workers.len() {
if self.workers[i].0 == idx {
self.workers.swap_remove(i);
found = true;
break;
}
break;
}
let ka = self.keep_alive;
let factory = Arc::clone(&self.factory);
let host = self.host.clone();
let addr = socks[0].addr;
if found {
error!("Worker has died {:?}, restarting", idx);
let (tx, rx) = mpsc::unbounded::<Conn<net::TcpStream>>();
let addr = Arbiter::start(move |ctx: &mut Context<_>| {
let settings = ServerSettings::new(Some(addr), &host, false);
let apps: Vec<_> =
(*factory)().into_iter().map(|h| h.into_handler()).collect();
ctx.add_message_stream(rx);
Worker::new(apps, socks, ka, settings)
});
for item in &self.accept {
let _ = item.1.send(Command::Worker(new_idx, tx.clone()));
let _ = item.0.set_readiness(mio::Ready::readable());
let mut new_idx = self.workers.len();
'found: loop {
for i in 0..self.workers.len() {
if self.workers[i].0 == new_idx {
new_idx += 1;
continue 'found;
}
}
break;
}
let ka = self.keep_alive;
let factory = Arc::clone(&self.factory);
let host = self.host.clone();
let addr = socks[0].addr;
let addr = Arbiter::start(move |ctx: &mut Context<_>| {
let settings = ServerSettings::new(Some(addr), &host, false);
let apps: Vec<_> =
(*factory)().into_iter().map(|h| h.into_handler()).collect();
ctx.add_message_stream(rx);
Worker::new(apps, socks, ka, settings)
});
for item in &self.accept {
let _ = item.1.send(Command::Worker(new_idx, tx.clone()));
let _ = item.0.set_readiness(mio::Ready::readable());
}
self.workers.push((new_idx, addr));
}
self.workers.push((new_idx, addr));
}
}
}

10
src/test.rs
View File

@ -22,7 +22,7 @@ use client::{ClientConnector, ClientRequest, ClientRequestBuilder};
use error::Error;
use handler::{AsyncResultItem, Handler, Responder};
use header::{Header, IntoHeaderValue};
use httprequest::{HttpRequest, RouterResource};
use httprequest::HttpRequest;
use httpresponse::HttpResponse;
use middleware::Middleware;
use param::Params;
@ -548,7 +548,7 @@ impl<S: 'static> TestRequest<S> {
*req.inner.payload.borrow_mut() = payload;
let mut req =
HttpRequest::new(req, Rc::new(state), router.route_info_params(params, 0));
HttpRequest::new(req, Rc::new(state), router.route_info_params(params));
req.set_cookies(cookies);
req
}
@ -574,7 +574,7 @@ impl<S: 'static> TestRequest<S> {
req.inner.headers = headers;
*req.inner.payload.borrow_mut() = payload;
let mut req =
HttpRequest::new(req, Rc::new(state), router.route_info_params(params, 0));
HttpRequest::new(req, Rc::new(state), router.route_info_params(params));
req.set_cookies(cookies);
req
}
@ -582,14 +582,12 @@ impl<S: 'static> TestRequest<S> {
/// Complete request creation and generate server `Request` instance
pub fn request(self) -> Request {
let TestRequest {
state,
method,
uri,
version,
headers,
params,
cookies,
payload,
..
} = self;
let mut req = Request::new(ServerSettings::default());
req.inner.method = method;

4
src/ws/client.rs
View File

@ -23,8 +23,8 @@ use httpmessage::HttpMessage;
use payload::PayloadHelper;
use client::{
ClientConnector, ClientRequest, ClientRequestBuilder, ClientResponse,
HttpResponseParserError, Pipeline, SendRequest, SendRequestError,
ClientConnector, ClientRequest, ClientRequestBuilder, HttpResponseParserError,
Pipeline, SendRequest, SendRequestError,
};
use super::frame::Frame;

2
src/ws/context.rs
View File

@ -92,7 +92,7 @@ where
{
#[inline]
/// Create a new Websocket context from a request and an actor
pub fn new<P>(req: HttpRequest<S>, actor: A, stream: WsStream<P>) -> Body
pub fn create<P>(req: HttpRequest<S>, actor: A, stream: WsStream<P>) -> Body
where
A: StreamHandler<Message, ProtocolError>,
P: Stream<Item = Bytes, Error = PayloadError> + 'static,

8
src/ws/mod.rs
View File

@ -45,7 +45,7 @@ use bytes::Bytes;
use futures::{Async, Poll, Stream};
use http::{header, Method, StatusCode};
use super::actix::{Actor, AsyncContext, StreamHandler};
use super::actix::{Actor, StreamHandler};
use body::Binary;
use error::{Error, PayloadError, ResponseError};
@ -179,7 +179,7 @@ where
let mut resp = handshake(req)?;
let stream = WsStream::new(req.payload());
let body = WebsocketContext::new(req.clone(), actor, stream);
let body = WebsocketContext::create(req.clone(), actor, stream);
Ok(resp.body(body))
}
@ -357,10 +357,8 @@ pub trait WsWriter {
#[cfg(test)]
mod tests {
use std::str::FromStr;
use super::*;
use http::{header, HeaderMap, Method, Uri, Version};
use http::{header, Method};
use test::TestRequest;
#[test]