mirror of
https://github.com/actix/actix-extras.git
synced 2024-11-24 07:53:00 +01:00
Merge branch 'master' into master
This commit is contained in:
Darin
GitHub
commit
287c2b1d18
@ -3,7 +3,7 @@ sudo: required
|
||||
dist: trusty
|
||||
|
||||
cache:
|
||||
cargo: true
|
||||
# cargo: true
|
||||
apt: true
|
||||
|
||||
matrix:
|
||||
@ -26,7 +26,7 @@ before_install:
|
||||
|
||||
before_cache: |
|
||||
if [[ "$TRAVIS_RUST_VERSION" == "nightly-2019-04-02" ]]; then
|
||||
RUSTFLAGS="--cfg procmacro2_semver_exempt" cargo install cargo-tarpaulin
|
||||
RUSTFLAGS="--cfg procmacro2_semver_exempt" cargo install --version 0.6.11 cargo-tarpaulin
|
||||
fi
|
||||
|
||||
# Add clippy
|
||||
|
||||
17
CHANGES.md
17
CHANGES.md
@ -1,11 +1,22 @@
|
||||
# Changes
|
||||
|
||||
## [1.0.3] - unreleased
|
||||
## [1.0.4] - TBD
|
||||
|
||||
### Changed
|
||||
|
||||
* Upgrade `rand` dependency version to 0.7
|
||||
|
||||
## [1.0.3] - 2019-06-28
|
||||
|
||||
### Added
|
||||
|
||||
* Support asynchronous data factories #850
|
||||
|
||||
### Changed
|
||||
|
||||
* Use `encoding_rs` crate instead of unmaintained `encoding` crate
|
||||
|
||||
|
||||
## [1.0.2] - 2019-06-17
|
||||
|
||||
### Changed
|
||||
@ -17,7 +28,7 @@
|
||||
|
||||
## [1.0.1] - 2019-06-17
|
||||
|
||||
### Add
|
||||
### Added
|
||||
|
||||
* Add support for PathConfig #903
|
||||
|
||||
@ -42,7 +53,7 @@
|
||||
|
||||
## [1.0.0] - 2019-06-05
|
||||
|
||||
### Add
|
||||
### Added
|
||||
|
||||
* Add `Scope::configure()` method.
|
||||
|
||||
|
||||
12
Cargo.toml
12
Cargo.toml
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "actix-web"
|
||||
version = "1.0.2"
|
||||
version = "1.0.3"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Actix web is a simple, pragmatic and extremely fast web framework for Rust."
|
||||
readme = "README.md"
|
||||
@ -71,11 +71,11 @@ rust-tls = ["rustls", "actix-server/rust-tls"]
|
||||
[dependencies]
|
||||
actix-codec = "0.1.2"
|
||||
actix-service = "0.4.1"
|
||||
actix-utils = "0.4.1"
|
||||
actix-utils = "0.4.2"
|
||||
actix-router = "0.1.5"
|
||||
actix-rt = "0.2.2"
|
||||
actix-rt = "0.2.3"
|
||||
actix-web-codegen = "0.1.2"
|
||||
actix-http = "0.2.4"
|
||||
actix-http = "0.2.5"
|
||||
actix-server = "0.5.1"
|
||||
actix-server-config = "0.1.1"
|
||||
actix-threadpool = "0.1.1"
|
||||
@ -103,9 +103,9 @@ rustls = { version = "0.15", optional = true }
|
||||
|
||||
[dev-dependencies]
|
||||
actix = { version = "0.8.3" }
|
||||
actix-http = { version = "0.2.4", features=["ssl", "brotli", "flate2-zlib"] }
|
||||
actix-http = { version = "0.2.5", features=["ssl", "brotli", "flate2-zlib"] }
|
||||
actix-http-test = { version = "0.2.2", features=["ssl"] }
|
||||
rand = "0.6"
|
||||
rand = "0.7"
|
||||
env_logger = "0.6"
|
||||
serde_derive = "1.0"
|
||||
tokio-timer = "0.2.8"
|
||||
|
||||
@ -3,7 +3,7 @@
|
||||
## Documentation & community resources
|
||||
|
||||
* [User Guide](https://actix.rs/docs/)
|
||||
* [API Documentation](https://docs.rs/actix-identity/)
|
||||
* [API Documentation](https://docs.rs/actix-cors/)
|
||||
* [Chat on gitter](https://gitter.im/actix/actix)
|
||||
* Cargo package: [actix-session](https://crates.io/crates/actix-identity)
|
||||
* Cargo package: [actix-cors](https://crates.io/crates/actix-cors)
|
||||
* Minimum supported Rust version: 1.34 or later
|
||||
|
||||
@ -1,5 +1,10 @@
|
||||
# Changes
|
||||
|
||||
## [0.1.3] - 2019-06-28
|
||||
|
||||
* Do not set `Content-Length` header, let actix-http set it #930
|
||||
|
||||
|
||||
## [0.1.2] - 2019-06-13
|
||||
|
||||
* Content-Length is 0 for NamedFile HEAD request #914
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "actix-files"
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Static files support for actix web."
|
||||
readme = "README.md"
|
||||
@ -18,7 +18,7 @@ name = "actix_files"
|
||||
path = "src/lib.rs"
|
||||
|
||||
[dependencies]
|
||||
actix-web = { version = "1.0.0", default-features = false }
|
||||
actix-web = { version = "1.0.2", default-features = false }
|
||||
actix-http = "0.2.4"
|
||||
actix-service = "0.4.1"
|
||||
bitflags = "1"
|
||||
@ -32,4 +32,4 @@ percent-encoding = "1.0"
|
||||
v_htmlescape = "0.4"
|
||||
|
||||
[dev-dependencies]
|
||||
actix-web = { version = "1.0.0", features=["ssl"] }
|
||||
actix-web = { version = "1.0.2", features=["ssl"] }
|
||||
|
||||
@ -855,6 +855,8 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_named_file_content_length_headers() {
|
||||
use actix_web::body::{MessageBody, ResponseBody};
|
||||
|
||||
let mut srv = test::init_service(
|
||||
App::new().service(Files::new("test", ".").index_file("tests/test.binary")),
|
||||
);
|
||||
@ -866,14 +868,13 @@ mod tests {
|
||||
.to_request();
|
||||
let response = test::call_service(&mut srv, request);
|
||||
|
||||
let contentlength = response
|
||||
.headers()
|
||||
.get(header::CONTENT_LENGTH)
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(contentlength, "11");
|
||||
// let contentlength = response
|
||||
// .headers()
|
||||
// .get(header::CONTENT_LENGTH)
|
||||
// .unwrap()
|
||||
// .to_str()
|
||||
// .unwrap();
|
||||
// assert_eq!(contentlength, "11");
|
||||
|
||||
// Invalid range header
|
||||
let request = TestRequest::get()
|
||||
@ -890,14 +891,13 @@ mod tests {
|
||||
.to_request();
|
||||
let response = test::call_service(&mut srv, request);
|
||||
|
||||
let contentlength = response
|
||||
.headers()
|
||||
.get(header::CONTENT_LENGTH)
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(contentlength, "100");
|
||||
// let contentlength = response
|
||||
// .headers()
|
||||
// .get(header::CONTENT_LENGTH)
|
||||
// .unwrap()
|
||||
// .to_str()
|
||||
// .unwrap();
|
||||
// assert_eq!(contentlength, "100");
|
||||
|
||||
// chunked
|
||||
let request = TestRequest::get()
|
||||
@ -939,14 +939,14 @@ mod tests {
|
||||
.to_request();
|
||||
let response = test::call_service(&mut srv, request);
|
||||
|
||||
let contentlength = response
|
||||
.headers()
|
||||
.get(header::CONTENT_LENGTH)
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(contentlength, "100");
|
||||
// TODO: fix check
|
||||
// let contentlength = response
|
||||
// .headers()
|
||||
// .get(header::CONTENT_LENGTH)
|
||||
// .unwrap()
|
||||
// .to_str()
|
||||
// .unwrap();
|
||||
// assert_eq!(contentlength, "100");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@ -414,8 +414,6 @@ impl Responder for NamedFile {
|
||||
};
|
||||
};
|
||||
|
||||
resp.header(header::CONTENT_LENGTH, format!("{}", length));
|
||||
|
||||
if precondition_failed {
|
||||
return Ok(resp.status(StatusCode::PRECONDITION_FAILED).finish());
|
||||
} else if not_modified {
|
||||
|
||||
@ -1,10 +1,23 @@
|
||||
# Changes
|
||||
|
||||
## [0.2.5] - unreleased
|
||||
## [0.2.6] - TBD
|
||||
|
||||
### Changed
|
||||
|
||||
* Use `encoding_rs` crate instead of unmaintained `encoding` crate
|
||||
* Upgrade `rand` dependency version to 0.7
|
||||
|
||||
## [0.2.5] - 2019-06-28
|
||||
|
||||
### Added
|
||||
|
||||
* Add `on-connect` callback, `HttpServiceBuilder::on_connect()` #946
|
||||
|
||||
### Changed
|
||||
|
||||
* Use `encoding_rs` crate instead of unmaintained `encoding` crate
|
||||
|
||||
* Add `Copy` and `Clone` impls for `ws::Codec`
|
||||
|
||||
|
||||
## [0.2.4] - 2019-06-16
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "actix-http"
|
||||
version = "0.2.4"
|
||||
version = "0.2.5"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Actix http primitives"
|
||||
readme = "README.md"
|
||||
@ -44,10 +44,10 @@ fail = ["failure"]
|
||||
secure-cookies = ["ring"]
|
||||
|
||||
[dependencies]
|
||||
actix-service = "0.4.0"
|
||||
actix-service = "0.4.1"
|
||||
actix-codec = "0.1.2"
|
||||
actix-connect = "0.2.0"
|
||||
actix-utils = "0.4.1"
|
||||
actix-utils = "0.4.2"
|
||||
actix-server-config = "0.1.1"
|
||||
actix-threadpool = "0.1.0"
|
||||
|
||||
@ -70,7 +70,7 @@ language-tags = "0.2"
|
||||
log = "0.4"
|
||||
mime = "0.3"
|
||||
percent-encoding = "1.0"
|
||||
rand = "0.6"
|
||||
rand = "0.7"
|
||||
regex = "1.0"
|
||||
serde = "1.0"
|
||||
serde_json = "1.0"
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
use std::fmt;
|
||||
use std::marker::PhantomData;
|
||||
use std::rc::Rc;
|
||||
|
||||
use actix_codec::Framed;
|
||||
use actix_server_config::ServerConfig as SrvConfig;
|
||||
@ -10,6 +11,7 @@ use crate::config::{KeepAlive, ServiceConfig};
|
||||
use crate::error::Error;
|
||||
use crate::h1::{Codec, ExpectHandler, H1Service, UpgradeHandler};
|
||||
use crate::h2::H2Service;
|
||||
use crate::helpers::{Data, DataFactory};
|
||||
use crate::request::Request;
|
||||
use crate::response::Response;
|
||||
use crate::service::HttpService;
|
||||
@ -24,6 +26,7 @@ pub struct HttpServiceBuilder<T, S, X = ExpectHandler, U = UpgradeHandler<T>> {
|
||||
client_disconnect: u64,
|
||||
expect: X,
|
||||
upgrade: Option<U>,
|
||||
on_connect: Option<Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
_t: PhantomData<(T, S)>,
|
||||
}
|
||||
|
||||
@ -41,6 +44,7 @@ where
|
||||
client_disconnect: 0,
|
||||
expect: ExpectHandler,
|
||||
upgrade: None,
|
||||
on_connect: None,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -115,6 +119,7 @@ where
|
||||
client_disconnect: self.client_disconnect,
|
||||
expect: expect.into_new_service(),
|
||||
upgrade: self.upgrade,
|
||||
on_connect: self.on_connect,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -140,10 +145,24 @@ where
|
||||
client_disconnect: self.client_disconnect,
|
||||
expect: self.expect,
|
||||
upgrade: Some(upgrade.into_new_service()),
|
||||
on_connect: self.on_connect,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
/// Set on-connect callback.
|
||||
///
|
||||
/// It get called once per connection and result of the call
|
||||
/// get stored to the request's extensions.
|
||||
pub fn on_connect<F, I>(mut self, f: F) -> Self
|
||||
where
|
||||
F: Fn(&T) -> I + 'static,
|
||||
I: Clone + 'static,
|
||||
{
|
||||
self.on_connect = Some(Rc::new(move |io| Box::new(Data(f(io)))));
|
||||
self
|
||||
}
|
||||
|
||||
/// Finish service configuration and create *http service* for HTTP/1 protocol.
|
||||
pub fn h1<F, P, B>(self, service: F) -> H1Service<T, P, S, B, X, U>
|
||||
where
|
||||
@ -161,6 +180,7 @@ where
|
||||
H1Service::with_config(cfg, service.into_new_service())
|
||||
.expect(self.expect)
|
||||
.upgrade(self.upgrade)
|
||||
.on_connect(self.on_connect)
|
||||
}
|
||||
|
||||
/// Finish service configuration and create *http service* for HTTP/2 protocol.
|
||||
@ -199,5 +219,6 @@ where
|
||||
HttpService::with_config(cfg, service.into_new_service())
|
||||
.expect(self.expect)
|
||||
.upgrade(self.upgrade)
|
||||
.on_connect(self.on_connect)
|
||||
}
|
||||
}
|
||||
|
||||
@ -16,6 +16,8 @@ use crate::body::{Body, BodySize, MessageBody, ResponseBody};
|
||||
use crate::config::ServiceConfig;
|
||||
use crate::error::{DispatchError, Error};
|
||||
use crate::error::{ParseError, PayloadError};
|
||||
use crate::helpers::DataFactory;
|
||||
use crate::httpmessage::HttpMessage;
|
||||
use crate::request::Request;
|
||||
use crate::response::Response;
|
||||
|
||||
@ -81,6 +83,7 @@ where
|
||||
service: CloneableService<S>,
|
||||
expect: CloneableService<X>,
|
||||
upgrade: Option<CloneableService<U>>,
|
||||
on_connect: Option<Box<dyn DataFactory>>,
|
||||
flags: Flags,
|
||||
peer_addr: Option<net::SocketAddr>,
|
||||
error: Option<DispatchError>,
|
||||
@ -174,12 +177,13 @@ where
|
||||
U::Error: fmt::Display,
|
||||
{
|
||||
/// Create http/1 dispatcher.
|
||||
pub fn new(
|
||||
pub(crate) fn new(
|
||||
stream: T,
|
||||
config: ServiceConfig,
|
||||
service: CloneableService<S>,
|
||||
expect: CloneableService<X>,
|
||||
upgrade: Option<CloneableService<U>>,
|
||||
on_connect: Option<Box<dyn DataFactory>>,
|
||||
) -> Self {
|
||||
Dispatcher::with_timeout(
|
||||
stream,
|
||||
@ -190,11 +194,12 @@ where
|
||||
service,
|
||||
expect,
|
||||
upgrade,
|
||||
on_connect,
|
||||
)
|
||||
}
|
||||
|
||||
/// Create http/1 dispatcher with slow request timeout.
|
||||
pub fn with_timeout(
|
||||
pub(crate) fn with_timeout(
|
||||
io: T,
|
||||
codec: Codec,
|
||||
config: ServiceConfig,
|
||||
@ -203,6 +208,7 @@ where
|
||||
service: CloneableService<S>,
|
||||
expect: CloneableService<X>,
|
||||
upgrade: Option<CloneableService<U>>,
|
||||
on_connect: Option<Box<dyn DataFactory>>,
|
||||
) -> Self {
|
||||
let keepalive = config.keep_alive_enabled();
|
||||
let flags = if keepalive {
|
||||
@ -234,6 +240,7 @@ where
|
||||
service,
|
||||
expect,
|
||||
upgrade,
|
||||
on_connect,
|
||||
flags,
|
||||
ka_expire,
|
||||
ka_timer,
|
||||
@ -495,6 +502,11 @@ where
|
||||
let pl = self.codec.message_type();
|
||||
req.head_mut().peer_addr = self.peer_addr;
|
||||
|
||||
// on_connect data
|
||||
if let Some(ref on_connect) = self.on_connect {
|
||||
on_connect.set(&mut req.extensions_mut());
|
||||
}
|
||||
|
||||
if pl == MessageType::Stream && self.upgrade.is_some() {
|
||||
self.messages.push_back(DispatcherMessage::Upgrade(req));
|
||||
break;
|
||||
@ -851,6 +863,7 @@ mod tests {
|
||||
),
|
||||
CloneableService::new(ExpectHandler),
|
||||
None,
|
||||
None,
|
||||
);
|
||||
assert!(h1.poll().is_err());
|
||||
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
use std::fmt;
|
||||
use std::marker::PhantomData;
|
||||
use std::rc::Rc;
|
||||
|
||||
use actix_codec::Framed;
|
||||
use actix_server_config::{Io, IoStream, ServerConfig as SrvConfig};
|
||||
@ -11,6 +12,7 @@ use futures::{try_ready, Async, Future, IntoFuture, Poll, Stream};
|
||||
use crate::body::MessageBody;
|
||||
use crate::config::{KeepAlive, ServiceConfig};
|
||||
use crate::error::{DispatchError, Error, ParseError};
|
||||
use crate::helpers::DataFactory;
|
||||
use crate::request::Request;
|
||||
use crate::response::Response;
|
||||
|
||||
@ -24,6 +26,7 @@ pub struct H1Service<T, P, S, B, X = ExpectHandler, U = UpgradeHandler<T>> {
|
||||
cfg: ServiceConfig,
|
||||
expect: X,
|
||||
upgrade: Option<U>,
|
||||
on_connect: Option<Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
|
||||
@ -44,6 +47,7 @@ where
|
||||
srv: service.into_new_service(),
|
||||
expect: ExpectHandler,
|
||||
upgrade: None,
|
||||
on_connect: None,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -55,6 +59,7 @@ where
|
||||
srv: service.into_new_service(),
|
||||
expect: ExpectHandler,
|
||||
upgrade: None,
|
||||
on_connect: None,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -79,6 +84,7 @@ where
|
||||
cfg: self.cfg,
|
||||
srv: self.srv,
|
||||
upgrade: self.upgrade,
|
||||
on_connect: self.on_connect,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -94,9 +100,19 @@ where
|
||||
cfg: self.cfg,
|
||||
srv: self.srv,
|
||||
expect: self.expect,
|
||||
on_connect: self.on_connect,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
/// Set on connect callback.
|
||||
pub(crate) fn on_connect(
|
||||
mut self,
|
||||
f: Option<Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
) -> Self {
|
||||
self.on_connect = f;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, P, S, B, X, U> NewService for H1Service<T, P, S, B, X, U>
|
||||
@ -133,6 +149,7 @@ where
|
||||
fut_upg: self.upgrade.as_ref().map(|f| f.new_service(cfg)),
|
||||
expect: None,
|
||||
upgrade: None,
|
||||
on_connect: self.on_connect.clone(),
|
||||
cfg: Some(self.cfg.clone()),
|
||||
_t: PhantomData,
|
||||
}
|
||||
@ -157,6 +174,7 @@ where
|
||||
fut_upg: Option<U::Future>,
|
||||
expect: Option<X::Service>,
|
||||
upgrade: Option<U::Service>,
|
||||
on_connect: Option<Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
cfg: Option<ServiceConfig>,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
@ -205,6 +223,7 @@ where
|
||||
service,
|
||||
self.expect.take().unwrap(),
|
||||
self.upgrade.take(),
|
||||
self.on_connect.clone(),
|
||||
)))
|
||||
}
|
||||
}
|
||||
@ -214,6 +233,7 @@ pub struct H1ServiceHandler<T, P, S, B, X, U> {
|
||||
srv: CloneableService<S>,
|
||||
expect: CloneableService<X>,
|
||||
upgrade: Option<CloneableService<U>>,
|
||||
on_connect: Option<Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
cfg: ServiceConfig,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
@ -234,12 +254,14 @@ where
|
||||
srv: S,
|
||||
expect: X,
|
||||
upgrade: Option<U>,
|
||||
on_connect: Option<Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
) -> H1ServiceHandler<T, P, S, B, X, U> {
|
||||
H1ServiceHandler {
|
||||
srv: CloneableService::new(srv),
|
||||
expect: CloneableService::new(expect),
|
||||
upgrade: upgrade.map(|s| CloneableService::new(s)),
|
||||
cfg,
|
||||
on_connect,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -292,12 +314,21 @@ where
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
||||
let io = req.into_parts().0;
|
||||
|
||||
let on_connect = if let Some(ref on_connect) = self.on_connect {
|
||||
Some(on_connect(&io))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
Dispatcher::new(
|
||||
req.into_parts().0,
|
||||
io,
|
||||
self.cfg.clone(),
|
||||
self.srv.clone(),
|
||||
self.expect.clone(),
|
||||
self.upgrade.clone(),
|
||||
on_connect,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@ -22,6 +22,7 @@ use tokio_timer::Delay;
|
||||
use crate::body::{Body, BodySize, MessageBody, ResponseBody};
|
||||
use crate::config::ServiceConfig;
|
||||
use crate::error::{DispatchError, Error, ParseError, PayloadError, ResponseError};
|
||||
use crate::helpers::DataFactory;
|
||||
use crate::message::ResponseHead;
|
||||
use crate::payload::Payload;
|
||||
use crate::request::Request;
|
||||
@ -33,6 +34,7 @@ const CHUNK_SIZE: usize = 16_384;
|
||||
pub struct Dispatcher<T: IoStream, S: Service<Request = Request>, B: MessageBody> {
|
||||
service: CloneableService<S>,
|
||||
connection: Connection<T, Bytes>,
|
||||
on_connect: Option<Box<dyn DataFactory>>,
|
||||
config: ServiceConfig,
|
||||
peer_addr: Option<net::SocketAddr>,
|
||||
ka_expire: Instant,
|
||||
@ -49,9 +51,10 @@ where
|
||||
S::Response: Into<Response<B>>,
|
||||
B: MessageBody + 'static,
|
||||
{
|
||||
pub fn new(
|
||||
pub(crate) fn new(
|
||||
service: CloneableService<S>,
|
||||
connection: Connection<T, Bytes>,
|
||||
on_connect: Option<Box<dyn DataFactory>>,
|
||||
config: ServiceConfig,
|
||||
timeout: Option<Delay>,
|
||||
peer_addr: Option<net::SocketAddr>,
|
||||
@ -77,6 +80,7 @@ where
|
||||
config,
|
||||
peer_addr,
|
||||
connection,
|
||||
on_connect,
|
||||
ka_expire,
|
||||
ka_timer,
|
||||
_t: PhantomData,
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
use std::fmt::Debug;
|
||||
use std::marker::PhantomData;
|
||||
use std::{io, net};
|
||||
use std::{io, net, rc};
|
||||
|
||||
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
||||
use actix_server_config::{Io, IoStream, ServerConfig as SrvConfig};
|
||||
@ -16,6 +16,7 @@ use log::error;
|
||||
use crate::body::MessageBody;
|
||||
use crate::config::{KeepAlive, ServiceConfig};
|
||||
use crate::error::{DispatchError, Error, ParseError, ResponseError};
|
||||
use crate::helpers::DataFactory;
|
||||
use crate::payload::Payload;
|
||||
use crate::request::Request;
|
||||
use crate::response::Response;
|
||||
@ -26,6 +27,7 @@ use super::dispatcher::Dispatcher;
|
||||
pub struct H2Service<T, P, S, B> {
|
||||
srv: S,
|
||||
cfg: ServiceConfig,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
|
||||
@ -43,6 +45,7 @@ where
|
||||
|
||||
H2Service {
|
||||
cfg,
|
||||
on_connect: None,
|
||||
srv: service.into_new_service(),
|
||||
_t: PhantomData,
|
||||
}
|
||||
@ -52,10 +55,20 @@ where
|
||||
pub fn with_config<F: IntoNewService<S>>(cfg: ServiceConfig, service: F) -> Self {
|
||||
H2Service {
|
||||
cfg,
|
||||
on_connect: None,
|
||||
srv: service.into_new_service(),
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
/// Set on connect callback.
|
||||
pub(crate) fn on_connect(
|
||||
mut self,
|
||||
f: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
) -> Self {
|
||||
self.on_connect = f;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, P, S, B> NewService for H2Service<T, P, S, B>
|
||||
@ -79,6 +92,7 @@ where
|
||||
H2ServiceResponse {
|
||||
fut: self.srv.new_service(cfg).into_future(),
|
||||
cfg: Some(self.cfg.clone()),
|
||||
on_connect: self.on_connect.clone(),
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -88,6 +102,7 @@ where
|
||||
pub struct H2ServiceResponse<T, P, S: NewService, B> {
|
||||
fut: <S::Future as IntoFuture>::Future,
|
||||
cfg: Option<ServiceConfig>,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
|
||||
@ -107,6 +122,7 @@ where
|
||||
let service = try_ready!(self.fut.poll());
|
||||
Ok(Async::Ready(H2ServiceHandler::new(
|
||||
self.cfg.take().unwrap(),
|
||||
self.on_connect.clone(),
|
||||
service,
|
||||
)))
|
||||
}
|
||||
@ -116,6 +132,7 @@ where
|
||||
pub struct H2ServiceHandler<T, P, S, B> {
|
||||
srv: CloneableService<S>,
|
||||
cfg: ServiceConfig,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
|
||||
@ -127,9 +144,14 @@ where
|
||||
S::Response: Into<Response<B>>,
|
||||
B: MessageBody + 'static,
|
||||
{
|
||||
fn new(cfg: ServiceConfig, srv: S) -> H2ServiceHandler<T, P, S, B> {
|
||||
fn new(
|
||||
cfg: ServiceConfig,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
srv: S,
|
||||
) -> H2ServiceHandler<T, P, S, B> {
|
||||
H2ServiceHandler {
|
||||
cfg,
|
||||
on_connect,
|
||||
srv: CloneableService::new(srv),
|
||||
_t: PhantomData,
|
||||
}
|
||||
@ -161,11 +183,18 @@ where
|
||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
||||
let io = req.into_parts().0;
|
||||
let peer_addr = io.peer_addr();
|
||||
let on_connect = if let Some(ref on_connect) = self.on_connect {
|
||||
Some(on_connect(&io))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
H2ServiceHandlerResponse {
|
||||
state: State::Handshake(
|
||||
Some(self.srv.clone()),
|
||||
Some(self.cfg.clone()),
|
||||
peer_addr,
|
||||
on_connect,
|
||||
server::handshake(io),
|
||||
),
|
||||
}
|
||||
@ -181,6 +210,7 @@ where
|
||||
Option<CloneableService<S>>,
|
||||
Option<ServiceConfig>,
|
||||
Option<net::SocketAddr>,
|
||||
Option<Box<dyn DataFactory>>,
|
||||
Handshake<T, Bytes>,
|
||||
),
|
||||
}
|
||||
@ -216,12 +246,14 @@ where
|
||||
ref mut srv,
|
||||
ref mut config,
|
||||
ref peer_addr,
|
||||
ref mut on_connect,
|
||||
ref mut handshake,
|
||||
) => match handshake.poll() {
|
||||
Ok(Async::Ready(conn)) => {
|
||||
self.state = State::Incoming(Dispatcher::new(
|
||||
srv.take().unwrap(),
|
||||
conn,
|
||||
on_connect.take(),
|
||||
config.take().unwrap(),
|
||||
None,
|
||||
peer_addr.clone(),
|
||||
|
||||
@ -3,6 +3,8 @@ use std::{io, mem, ptr, slice};
|
||||
use bytes::{BufMut, BytesMut};
|
||||
use http::Version;
|
||||
|
||||
use crate::extensions::Extensions;
|
||||
|
||||
const DEC_DIGITS_LUT: &[u8] = b"0001020304050607080910111213141516171819\
|
||||
2021222324252627282930313233343536373839\
|
||||
4041424344454647484950515253545556575859\
|
||||
@ -180,6 +182,18 @@ impl<'a> io::Write for Writer<'a> {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) trait DataFactory {
|
||||
fn set(&self, ext: &mut Extensions);
|
||||
}
|
||||
|
||||
pub(crate) struct Data<T>(pub(crate) T);
|
||||
|
||||
impl<T: Clone + 'static> DataFactory for Data<T> {
|
||||
fn set(&self, ext: &mut Extensions) {
|
||||
ext.insert(self.0.clone())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
use std::marker::PhantomData;
|
||||
use std::{fmt, io, net};
|
||||
use std::{fmt, io, net, rc};
|
||||
|
||||
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
||||
use actix_server_config::{
|
||||
@ -15,6 +15,7 @@ use crate::body::MessageBody;
|
||||
use crate::builder::HttpServiceBuilder;
|
||||
use crate::config::{KeepAlive, ServiceConfig};
|
||||
use crate::error::{DispatchError, Error};
|
||||
use crate::helpers::DataFactory;
|
||||
use crate::request::Request;
|
||||
use crate::response::Response;
|
||||
use crate::{h1, h2::Dispatcher};
|
||||
@ -25,6 +26,7 @@ pub struct HttpService<T, P, S, B, X = h1::ExpectHandler, U = h1::UpgradeHandler
|
||||
cfg: ServiceConfig,
|
||||
expect: X,
|
||||
upgrade: Option<U>,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
|
||||
@ -61,6 +63,7 @@ where
|
||||
srv: service.into_new_service(),
|
||||
expect: h1::ExpectHandler,
|
||||
upgrade: None,
|
||||
on_connect: None,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -75,6 +78,7 @@ where
|
||||
srv: service.into_new_service(),
|
||||
expect: h1::ExpectHandler,
|
||||
upgrade: None,
|
||||
on_connect: None,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -104,6 +108,7 @@ where
|
||||
cfg: self.cfg,
|
||||
srv: self.srv,
|
||||
upgrade: self.upgrade,
|
||||
on_connect: self.on_connect,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
@ -127,9 +132,19 @@ where
|
||||
cfg: self.cfg,
|
||||
srv: self.srv,
|
||||
expect: self.expect,
|
||||
on_connect: self.on_connect,
|
||||
_t: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
/// Set on connect callback.
|
||||
pub(crate) fn on_connect(
|
||||
mut self,
|
||||
f: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
) -> Self {
|
||||
self.on_connect = f;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, P, S, B, X, U> NewService for HttpService<T, P, S, B, X, U>
|
||||
@ -167,6 +182,7 @@ where
|
||||
fut_upg: self.upgrade.as_ref().map(|f| f.new_service(cfg)),
|
||||
expect: None,
|
||||
upgrade: None,
|
||||
on_connect: self.on_connect.clone(),
|
||||
cfg: Some(self.cfg.clone()),
|
||||
_t: PhantomData,
|
||||
}
|
||||
@ -180,6 +196,7 @@ pub struct HttpServiceResponse<T, P, S: NewService, B, X: NewService, U: NewServ
|
||||
fut_upg: Option<U::Future>,
|
||||
expect: Option<X::Service>,
|
||||
upgrade: Option<U::Service>,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
cfg: Option<ServiceConfig>,
|
||||
_t: PhantomData<(T, P, B)>,
|
||||
}
|
||||
@ -229,6 +246,7 @@ where
|
||||
service,
|
||||
self.expect.take().unwrap(),
|
||||
self.upgrade.take(),
|
||||
self.on_connect.clone(),
|
||||
)))
|
||||
}
|
||||
}
|
||||
@ -239,6 +257,7 @@ pub struct HttpServiceHandler<T, P, S, B, X, U> {
|
||||
expect: CloneableService<X>,
|
||||
upgrade: Option<CloneableService<U>>,
|
||||
cfg: ServiceConfig,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
_t: PhantomData<(T, P, B, X)>,
|
||||
}
|
||||
|
||||
@ -259,9 +278,11 @@ where
|
||||
srv: S,
|
||||
expect: X,
|
||||
upgrade: Option<U>,
|
||||
on_connect: Option<rc::Rc<Fn(&T) -> Box<dyn DataFactory>>>,
|
||||
) -> HttpServiceHandler<T, P, S, B, X, U> {
|
||||
HttpServiceHandler {
|
||||
cfg,
|
||||
on_connect,
|
||||
srv: CloneableService::new(srv),
|
||||
expect: CloneableService::new(expect),
|
||||
upgrade: upgrade.map(|s| CloneableService::new(s)),
|
||||
@ -319,6 +340,13 @@ where
|
||||
|
||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
||||
let (io, _, proto) = req.into_parts();
|
||||
|
||||
let on_connect = if let Some(ref on_connect) = self.on_connect {
|
||||
Some(on_connect(&io))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
match proto {
|
||||
Protocol::Http2 => {
|
||||
let peer_addr = io.peer_addr();
|
||||
@ -332,6 +360,7 @@ where
|
||||
self.cfg.clone(),
|
||||
self.srv.clone(),
|
||||
peer_addr,
|
||||
on_connect,
|
||||
))),
|
||||
}
|
||||
}
|
||||
@ -342,6 +371,7 @@ where
|
||||
self.srv.clone(),
|
||||
self.expect.clone(),
|
||||
self.upgrade.clone(),
|
||||
on_connect,
|
||||
)),
|
||||
},
|
||||
_ => HttpServiceHandlerResponse {
|
||||
@ -352,6 +382,7 @@ where
|
||||
self.srv.clone(),
|
||||
self.expect.clone(),
|
||||
self.upgrade.clone(),
|
||||
on_connect,
|
||||
))),
|
||||
},
|
||||
}
|
||||
@ -380,6 +411,7 @@ where
|
||||
CloneableService<S>,
|
||||
CloneableService<X>,
|
||||
Option<CloneableService<U>>,
|
||||
Option<Box<dyn DataFactory>>,
|
||||
)>,
|
||||
),
|
||||
Handshake(
|
||||
@ -388,6 +420,7 @@ where
|
||||
ServiceConfig,
|
||||
CloneableService<S>,
|
||||
Option<net::SocketAddr>,
|
||||
Option<Box<dyn DataFactory>>,
|
||||
)>,
|
||||
),
|
||||
}
|
||||
@ -448,7 +481,8 @@ where
|
||||
} else {
|
||||
panic!()
|
||||
}
|
||||
let (io, buf, cfg, srv, expect, upgrade) = data.take().unwrap();
|
||||
let (io, buf, cfg, srv, expect, upgrade, on_connect) =
|
||||
data.take().unwrap();
|
||||
if buf[..14] == HTTP2_PREFACE[..] {
|
||||
let peer_addr = io.peer_addr();
|
||||
let io = Io {
|
||||
@ -460,6 +494,7 @@ where
|
||||
cfg,
|
||||
srv,
|
||||
peer_addr,
|
||||
on_connect,
|
||||
)));
|
||||
} else {
|
||||
self.state = State::H1(h1::Dispatcher::with_timeout(
|
||||
@ -471,6 +506,7 @@ where
|
||||
srv,
|
||||
expect,
|
||||
upgrade,
|
||||
on_connect,
|
||||
))
|
||||
}
|
||||
self.poll()
|
||||
@ -488,8 +524,10 @@ where
|
||||
} else {
|
||||
panic!()
|
||||
};
|
||||
let (_, cfg, srv, peer_addr) = data.take().unwrap();
|
||||
self.state = State::H2(Dispatcher::new(srv, conn, cfg, None, peer_addr));
|
||||
let (_, cfg, srv, peer_addr, on_connect) = data.take().unwrap();
|
||||
self.state = State::H2(Dispatcher::new(
|
||||
srv, conn, on_connect, cfg, None, peer_addr,
|
||||
));
|
||||
self.poll()
|
||||
}
|
||||
}
|
||||
|
||||
@ -37,7 +37,7 @@ pub enum Frame {
|
||||
Close(Option<CloseReason>),
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
#[derive(Debug, Copy, Clone)]
|
||||
/// WebSockets protocol codec
|
||||
pub struct Codec {
|
||||
max_size: usize,
|
||||
|
||||
@ -1,5 +1,9 @@
|
||||
# Changes
|
||||
|
||||
## [1.0.1] - 2019-06-28
|
||||
|
||||
* Allow to use custom ws codec with `WebsocketContext` #925
|
||||
|
||||
## [1.0.0] - 2019-05-29
|
||||
|
||||
* Update actix-http and actix-web
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "actix-web-actors"
|
||||
version = "1.0.0"
|
||||
version = "1.0.1"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Actix actors support for actix web framework."
|
||||
readme = "README.md"
|
||||
@ -19,8 +19,8 @@ path = "src/lib.rs"
|
||||
|
||||
[dependencies]
|
||||
actix = "0.8.3"
|
||||
actix-web = "1.0.0"
|
||||
actix-http = "0.2.4"
|
||||
actix-web = "1.0.3"
|
||||
actix-http = "0.2.5"
|
||||
actix-codec = "0.1.2"
|
||||
bytes = "0.4"
|
||||
futures = "0.1.25"
|
||||
|
||||
@ -177,9 +177,30 @@ where
|
||||
inner: ContextParts::new(mb.sender_producer()),
|
||||
messages: VecDeque::new(),
|
||||
};
|
||||
ctx.add_stream(WsStream::new(stream));
|
||||
ctx.add_stream(WsStream::new(stream, Codec::new()));
|
||||
|
||||
WebsocketContextFut::new(ctx, actor, mb)
|
||||
WebsocketContextFut::new(ctx, actor, mb, Codec::new())
|
||||
}
|
||||
|
||||
#[inline]
|
||||
/// Create a new Websocket context from a request, an actor, and a codec
|
||||
pub fn with_codec<S>(
|
||||
actor: A,
|
||||
stream: S,
|
||||
codec: Codec,
|
||||
) -> impl Stream<Item = Bytes, Error = Error>
|
||||
where
|
||||
A: StreamHandler<Message, ProtocolError>,
|
||||
S: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
||||
{
|
||||
let mb = Mailbox::default();
|
||||
let mut ctx = WebsocketContext {
|
||||
inner: ContextParts::new(mb.sender_producer()),
|
||||
messages: VecDeque::new(),
|
||||
};
|
||||
ctx.add_stream(WsStream::new(stream, codec));
|
||||
|
||||
WebsocketContextFut::new(ctx, actor, mb, codec)
|
||||
}
|
||||
|
||||
/// Create a new Websocket context
|
||||
@ -197,11 +218,11 @@ where
|
||||
inner: ContextParts::new(mb.sender_producer()),
|
||||
messages: VecDeque::new(),
|
||||
};
|
||||
ctx.add_stream(WsStream::new(stream));
|
||||
ctx.add_stream(WsStream::new(stream, Codec::new()));
|
||||
|
||||
let act = f(&mut ctx);
|
||||
|
||||
WebsocketContextFut::new(ctx, act, mb)
|
||||
WebsocketContextFut::new(ctx, act, mb, Codec::new())
|
||||
}
|
||||
}
|
||||
|
||||
@ -288,11 +309,11 @@ impl<A> WebsocketContextFut<A>
|
||||
where
|
||||
A: Actor<Context = WebsocketContext<A>>,
|
||||
{
|
||||
fn new(ctx: WebsocketContext<A>, act: A, mailbox: Mailbox<A>) -> Self {
|
||||
fn new(ctx: WebsocketContext<A>, act: A, mailbox: Mailbox<A>, codec: Codec) -> Self {
|
||||
let fut = ContextFut::new(ctx, act, mailbox);
|
||||
WebsocketContextFut {
|
||||
fut,
|
||||
encoder: Codec::new(),
|
||||
encoder: codec,
|
||||
buf: BytesMut::new(),
|
||||
closed: false,
|
||||
}
|
||||
@ -353,10 +374,10 @@ impl<S> WsStream<S>
|
||||
where
|
||||
S: Stream<Item = Bytes, Error = PayloadError>,
|
||||
{
|
||||
fn new(stream: S) -> Self {
|
||||
fn new(stream: S, codec: Codec) -> Self {
|
||||
Self {
|
||||
stream,
|
||||
decoder: Codec::new(),
|
||||
decoder: codec,
|
||||
buf: BytesMut::new(),
|
||||
closed: false,
|
||||
}
|
||||
|
||||
@ -1,5 +1,14 @@
|
||||
# Changes
|
||||
|
||||
## [0.2.2] - 2019-07-01
|
||||
|
||||
### Changed
|
||||
|
||||
* Always append a colon after username in basic auth
|
||||
|
||||
* Upgrade `rand` dependency version to 0.7
|
||||
|
||||
|
||||
## [0.2.1] - 2019-06-05
|
||||
|
||||
### Added
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "awc"
|
||||
version = "0.2.1"
|
||||
version = "0.2.2"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Actix http client."
|
||||
readme = "README.md"
|
||||
@ -49,7 +49,7 @@ futures = "0.1.25"
|
||||
log =" 0.4"
|
||||
mime = "0.3"
|
||||
percent-encoding = "1.0"
|
||||
rand = "0.6"
|
||||
rand = "0.7"
|
||||
serde = "1.0"
|
||||
serde_json = "1.0"
|
||||
serde_urlencoded = "0.5.3"
|
||||
@ -66,5 +66,5 @@ actix-server = { version = "0.5.1", features=["ssl"] }
|
||||
brotli2 = { version="0.3.2" }
|
||||
flate2 = { version="1.0.2" }
|
||||
env_logger = "0.6"
|
||||
rand = "0.6"
|
||||
rand = "0.7"
|
||||
tokio-tcp = "0.1"
|
||||
@ -115,7 +115,7 @@ impl ClientBuilder {
|
||||
{
|
||||
let auth = match password {
|
||||
Some(password) => format!("{}:{}", username, password),
|
||||
None => format!("{}", username),
|
||||
None => format!("{}:", username),
|
||||
};
|
||||
self.header(
|
||||
header::AUTHORIZATION,
|
||||
@ -164,7 +164,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap(),
|
||||
"Basic dXNlcm5hbWU="
|
||||
"Basic dXNlcm5hbWU6"
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@ -280,7 +280,7 @@ impl ClientRequest {
|
||||
{
|
||||
let auth = match password {
|
||||
Some(password) => format!("{}:{}", username, password),
|
||||
None => format!("{}", username),
|
||||
None => format!("{}:", username),
|
||||
};
|
||||
self.header(
|
||||
header::AUTHORIZATION,
|
||||
@ -664,7 +664,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap(),
|
||||
"Basic dXNlcm5hbWU="
|
||||
"Basic dXNlcm5hbWU6"
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@ -195,7 +195,7 @@ impl WebsocketsRequest {
|
||||
{
|
||||
let auth = match password {
|
||||
Some(password) => format!("{}:{}", username, password),
|
||||
None => format!("{}", username),
|
||||
None => format!("{}:", username),
|
||||
};
|
||||
self.header(AUTHORIZATION, format!("Basic {}", base64::encode(&auth)))
|
||||
}
|
||||
@ -443,7 +443,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.to_str()
|
||||
.unwrap(),
|
||||
"Basic dXNlcm5hbWU="
|
||||
"Basic dXNlcm5hbWU6"
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
66
src/app.rs
66
src/app.rs
@ -8,7 +8,7 @@ use actix_service::boxed::{self, BoxedNewService};
|
||||
use actix_service::{
|
||||
apply_transform, IntoNewService, IntoTransform, NewService, Transform,
|
||||
};
|
||||
use futures::IntoFuture;
|
||||
use futures::{Future, IntoFuture};
|
||||
|
||||
use crate::app_service::{AppEntry, AppInit, AppRoutingFactory};
|
||||
use crate::config::{AppConfig, AppConfigInner, ServiceConfig};
|
||||
@ -23,6 +23,7 @@ use crate::service::{
|
||||
};
|
||||
|
||||
type HttpNewService = BoxedNewService<(), ServiceRequest, ServiceResponse, Error, ()>;
|
||||
type FnDataFactory = Box<Fn() -> Box<dyn Future<Item = Box<DataFactory>, Error = ()>>>;
|
||||
|
||||
/// Application builder - structure that follows the builder pattern
|
||||
/// for building application instances.
|
||||
@ -32,6 +33,7 @@ pub struct App<T, B> {
|
||||
default: Option<Rc<HttpNewService>>,
|
||||
factory_ref: Rc<RefCell<Option<AppRoutingFactory>>>,
|
||||
data: Vec<Box<DataFactory>>,
|
||||
data_factories: Vec<FnDataFactory>,
|
||||
config: AppConfigInner,
|
||||
external: Vec<ResourceDef>,
|
||||
_t: PhantomData<(B)>,
|
||||
@ -44,6 +46,7 @@ impl App<AppEntry, Body> {
|
||||
App {
|
||||
endpoint: AppEntry::new(fref.clone()),
|
||||
data: Vec::new(),
|
||||
data_factories: Vec::new(),
|
||||
services: Vec::new(),
|
||||
default: None,
|
||||
factory_ref: fref,
|
||||
@ -100,6 +103,31 @@ where
|
||||
self
|
||||
}
|
||||
|
||||
/// Set application data factory. This function is
|
||||
/// similar to `.data()` but it accepts data factory. Data object get
|
||||
/// constructed asynchronously during application initialization.
|
||||
pub fn data_factory<F, Out>(mut self, data: F) -> Self
|
||||
where
|
||||
F: Fn() -> Out + 'static,
|
||||
Out: IntoFuture + 'static,
|
||||
Out::Error: std::fmt::Debug,
|
||||
{
|
||||
self.data_factories.push(Box::new(move || {
|
||||
Box::new(
|
||||
data()
|
||||
.into_future()
|
||||
.map_err(|e| {
|
||||
log::error!("Can not construct data instance: {:?}", e);
|
||||
})
|
||||
.map(|data| {
|
||||
let data: Box<dyn DataFactory> = Box::new(Data::new(data));
|
||||
data
|
||||
}),
|
||||
)
|
||||
}));
|
||||
self
|
||||
}
|
||||
|
||||
/// Set application data. Application data could be accessed
|
||||
/// by using `Data<T>` extractor where `T` is data type.
|
||||
pub fn register_data<U: 'static>(mut self, data: Data<U>) -> Self {
|
||||
@ -349,6 +377,7 @@ where
|
||||
App {
|
||||
endpoint,
|
||||
data: self.data,
|
||||
data_factories: self.data_factories,
|
||||
services: self.services,
|
||||
default: self.default,
|
||||
factory_ref: self.factory_ref,
|
||||
@ -423,6 +452,7 @@ where
|
||||
fn into_new_service(self) -> AppInit<T, B> {
|
||||
AppInit {
|
||||
data: Rc::new(self.data),
|
||||
data_factories: Rc::new(self.data_factories),
|
||||
endpoint: self.endpoint,
|
||||
services: Rc::new(RefCell::new(self.services)),
|
||||
external: RefCell::new(self.external),
|
||||
@ -490,24 +520,24 @@ mod tests {
|
||||
assert_eq!(resp.status(), StatusCode::CREATED);
|
||||
}
|
||||
|
||||
// #[test]
|
||||
// fn test_data_factory() {
|
||||
// let mut srv =
|
||||
// init_service(App::new().data_factory(|| Ok::<_, ()>(10usize)).service(
|
||||
// web::resource("/").to(|_: web::Data<usize>| HttpResponse::Ok()),
|
||||
// ));
|
||||
// let req = TestRequest::default().to_request();
|
||||
// let resp = block_on(srv.call(req)).unwrap();
|
||||
// assert_eq!(resp.status(), StatusCode::OK);
|
||||
#[test]
|
||||
fn test_data_factory() {
|
||||
let mut srv =
|
||||
init_service(App::new().data_factory(|| Ok::<_, ()>(10usize)).service(
|
||||
web::resource("/").to(|_: web::Data<usize>| HttpResponse::Ok()),
|
||||
));
|
||||
let req = TestRequest::default().to_request();
|
||||
let resp = block_on(srv.call(req)).unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
|
||||
// let mut srv =
|
||||
// init_service(App::new().data_factory(|| Ok::<_, ()>(10u32)).service(
|
||||
// web::resource("/").to(|_: web::Data<usize>| HttpResponse::Ok()),
|
||||
// ));
|
||||
// let req = TestRequest::default().to_request();
|
||||
// let resp = block_on(srv.call(req)).unwrap();
|
||||
// assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
|
||||
// }
|
||||
let mut srv =
|
||||
init_service(App::new().data_factory(|| Ok::<_, ()>(10u32)).service(
|
||||
web::resource("/").to(|_: web::Data<usize>| HttpResponse::Ok()),
|
||||
));
|
||||
let req = TestRequest::default().to_request();
|
||||
let resp = block_on(srv.call(req)).unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
|
||||
}
|
||||
|
||||
fn md<S, B>(
|
||||
req: ServiceRequest,
|
||||
|
||||
@ -25,6 +25,7 @@ type BoxedResponse = Either<
|
||||
FutureResult<ServiceResponse, Error>,
|
||||
Box<Future<Item = ServiceResponse, Error = Error>>,
|
||||
>;
|
||||
type FnDataFactory = Box<Fn() -> Box<dyn Future<Item = Box<DataFactory>, Error = ()>>>;
|
||||
|
||||
/// Service factory to convert `Request` to a `ServiceRequest<S>`.
|
||||
/// It also executes data factories.
|
||||
@ -40,6 +41,7 @@ where
|
||||
{
|
||||
pub(crate) endpoint: T,
|
||||
pub(crate) data: Rc<Vec<Box<DataFactory>>>,
|
||||
pub(crate) data_factories: Rc<Vec<FnDataFactory>>,
|
||||
pub(crate) config: RefCell<AppConfig>,
|
||||
pub(crate) services: Rc<RefCell<Vec<Box<ServiceFactory>>>>,
|
||||
pub(crate) default: Option<Rc<HttpNewService>>,
|
||||
@ -119,16 +121,12 @@ where
|
||||
let rmap = Rc::new(rmap);
|
||||
rmap.finish(rmap.clone());
|
||||
|
||||
// create app data container
|
||||
let mut data = Extensions::new();
|
||||
for f in self.data.iter() {
|
||||
f.create(&mut data);
|
||||
}
|
||||
|
||||
AppInitResult {
|
||||
endpoint: None,
|
||||
endpoint_fut: self.endpoint.new_service(&()),
|
||||
data: Rc::new(data),
|
||||
data: self.data.clone(),
|
||||
data_factories: Vec::new(),
|
||||
data_factories_fut: self.data_factories.iter().map(|f| f()).collect(),
|
||||
config,
|
||||
rmap,
|
||||
_t: PhantomData,
|
||||
@ -144,7 +142,9 @@ where
|
||||
endpoint_fut: T::Future,
|
||||
rmap: Rc<ResourceMap>,
|
||||
config: AppConfig,
|
||||
data: Rc<Extensions>,
|
||||
data: Rc<Vec<Box<DataFactory>>>,
|
||||
data_factories: Vec<Box<DataFactory>>,
|
||||
data_factories_fut: Vec<Box<dyn Future<Item = Box<DataFactory>, Error = ()>>>,
|
||||
_t: PhantomData<B>,
|
||||
}
|
||||
|
||||
@ -159,21 +159,43 @@ where
|
||||
>,
|
||||
{
|
||||
type Item = AppInitService<T::Service, B>;
|
||||
type Error = T::InitError;
|
||||
type Error = ();
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
// async data factories
|
||||
let mut idx = 0;
|
||||
while idx < self.data_factories_fut.len() {
|
||||
match self.data_factories_fut[idx].poll()? {
|
||||
Async::Ready(f) => {
|
||||
self.data_factories.push(f);
|
||||
self.data_factories_fut.remove(idx);
|
||||
}
|
||||
Async::NotReady => idx += 1,
|
||||
}
|
||||
}
|
||||
|
||||
if self.endpoint.is_none() {
|
||||
if let Async::Ready(srv) = self.endpoint_fut.poll()? {
|
||||
self.endpoint = Some(srv);
|
||||
}
|
||||
}
|
||||
|
||||
if self.endpoint.is_some() {
|
||||
if self.endpoint.is_some() && self.data_factories_fut.is_empty() {
|
||||
// create app data container
|
||||
let mut data = Extensions::new();
|
||||
for f in self.data.iter() {
|
||||
f.create(&mut data);
|
||||
}
|
||||
|
||||
for f in &self.data_factories {
|
||||
f.create(&mut data);
|
||||
}
|
||||
|
||||
Ok(Async::Ready(AppInitService {
|
||||
service: self.endpoint.take().unwrap(),
|
||||
rmap: self.rmap.clone(),
|
||||
config: self.config.clone(),
|
||||
data: self.data.clone(),
|
||||
data: Rc::new(data),
|
||||
pool: HttpRequestPool::create(),
|
||||
}))
|
||||
} else {
|
||||
|
||||
@ -64,7 +64,7 @@ where
|
||||
RT.with(move |rt| rt.borrow_mut().get_mut().block_on(f.into_future()))
|
||||
}
|
||||
|
||||
/// Runs the provided function, blocking the current thread until the resul
|
||||
/// Runs the provided function, blocking the current thread until the result
|
||||
/// future completes.
|
||||
///
|
||||
/// This function can be used to synchronously block the current thread
|
||||
|
||||
@ -180,16 +180,14 @@ where
|
||||
.map(|c| (c.limit, c.ehandler.clone(), c.content_type.clone()))
|
||||
.unwrap_or((32768, None, None));
|
||||
|
||||
let path = req.path().to_string();
|
||||
|
||||
Box::new(
|
||||
JsonBody::new(req, payload, ctype)
|
||||
.limit(limit)
|
||||
.map_err(move |e| {
|
||||
log::debug!(
|
||||
"Failed to deserialize Json from payload. \
|
||||
Request path: {:?}",
|
||||
path
|
||||
Request path: {}",
|
||||
req2.path()
|
||||
);
|
||||
if let Some(err) = err {
|
||||
(*err)(e, &req2)
|
||||
@ -324,14 +322,11 @@ where
|
||||
};
|
||||
}
|
||||
|
||||
let mut len = None;
|
||||
if let Some(l) = req.headers().get(&CONTENT_LENGTH) {
|
||||
if let Ok(s) = l.to_str() {
|
||||
if let Ok(l) = s.parse::<usize>() {
|
||||
len = Some(l)
|
||||
}
|
||||
}
|
||||
}
|
||||
let len = req
|
||||
.headers()
|
||||
.get(&CONTENT_LENGTH)
|
||||
.and_then(|l| l.to_str().ok())
|
||||
.and_then(|s| s.parse::<usize>().ok());
|
||||
let payload = Decompress::from_headers(payload.take(), req.headers());
|
||||
|
||||
JsonBody {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user