actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-06-25 06:39:22 +02:00
Code Issues Releases Wiki Activity

Use immutable reference of service state. Update awc dns resolver. (#1905)

Browse Source
This commit is contained in:
fakeshadow
2021-02-06 17:00:40 -08:00
committed by GitHub
parent 20cf0094e5
commit 41bc04b1c4
65 changed files with 497 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
actix-http/src/client/config.rs
View File

@ -18,7 +18,7 @@ pub(crate) struct ConnectorConfig {
impl Default for ConnectorConfig {
fn default() -> Self {
Self {
timeout: Duration::from_secs(1),
timeout: Duration::from_secs(5),
conn_lifetime: Duration::from_secs(75),
conn_keep_alive: Duration::from_secs(15),
disconnect_timeout: Some(Duration::from_millis(3000)),

3
actix-http/src/client/connection.rs
View File

@ -5,7 +5,8 @@ use std::{fmt, io, time};
use actix_codec::{AsyncRead, AsyncWrite, Framed, ReadBuf};
use bytes::Bytes;
use futures_util::future::{err, Either, FutureExt, LocalBoxFuture, Ready};
use futures_core::future::LocalBoxFuture;
use futures_util::future::{err, Either, FutureExt, Ready};
use h2::client::SendRequest;
use pin_project::pin_project;

14
actix-http/src/client/connector.rs
View File

@ -100,9 +100,9 @@ impl Connector<(), ()> {
fn build_ssl(protocols: Vec<Vec<u8>>) -> SslConnector {
let mut config = ClientConfig::new();
config.set_protocols(&protocols);
config
.root_store
.add_server_trust_anchors(&actix_tls::accept::rustls::TLS_SERVER_ROOTS);
config.root_store.add_server_trust_anchors(
&actix_tls::connect::ssl::rustls::TLS_SERVER_ROOTS,
);
SslConnector::Rustls(Arc::new(config))
}
@ -392,11 +392,11 @@ mod connect_impl {
Ready<Result<IoConnection<Io>, ConnectError>>,
>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.tcp_pool.poll_ready(cx)
}
fn call(&mut self, req: Connect) -> Self::Future {
fn call(&self, req: Connect) -> Self::Future {
match req.uri.scheme_str() {
Some("https") | Some("wss") => {
Either::Right(err(ConnectError::SslIsNotSupported))
@ -460,11 +460,11 @@ mod connect_impl {
InnerConnectorResponseB<T2, Io1, Io2>,
>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.tcp_pool.poll_ready(cx)
}
fn call(&mut self, req: Connect) -> Self::Future {
fn call(&self, req: Connect) -> Self::Future {
match req.uri.scheme_str() {
Some("https") | Some("wss") => Either::Right(InnerConnectorResponseB {
fut: self.ssl_pool.call(req),

3
actix-http/src/client/error.rs
View File

@ -1,6 +1,5 @@
use std::io;
use actix_tls::connect::resolver::ResolveError;
use derive_more::{Display, From};
#[cfg(feature = "openssl")]
@ -23,7 +22,7 @@ pub enum ConnectError {
/// Failed to resolve the hostname
#[display(fmt = "Failed resolving hostname: {}", _0)]
Resolver(ResolveError),
Resolver(Box<dyn std::error::Error>),
/// No dns records
#[display(fmt = "No dns records found for the input")]

23
actix-http/src/client/pool.rs
View File

@ -13,7 +13,8 @@ use actix_utils::task::LocalWaker;
use ahash::AHashMap;
use bytes::Bytes;
use futures_channel::oneshot;
use futures_util::future::{poll_fn, FutureExt, LocalBoxFuture};
use futures_core::future::LocalBoxFuture;
use futures_util::future::{poll_fn, FutureExt};
use h2::client::{Connection, SendRequest};
use http::uri::Authority;
use indexmap::IndexSet;
@ -45,7 +46,7 @@ impl From<Authority> for Key {
}
/// Connections pool
pub(crate) struct ConnectionPool<T, Io: 'static>(Rc<RefCell<T>>, Rc<RefCell<Inner<Io>>>);
pub(crate) struct ConnectionPool<T, Io: 'static>(Rc<T>, Rc<RefCell<Inner<Io>>>);
impl<T, Io> ConnectionPool<T, Io>
where
@ -53,7 +54,7 @@ where
T: Service<Connect, Response = (Io, Protocol), Error = ConnectError> + 'static,
{
pub(crate) fn new(connector: T, config: ConnectorConfig) -> Self {
let connector_rc = Rc::new(RefCell::new(connector));
let connector_rc = Rc::new(connector);
let inner_rc = Rc::new(RefCell::new(Inner {
config,
acquired: 0,
@ -98,12 +99,12 @@ where
type Error = ConnectError;
type Future = LocalBoxFuture<'static, Result<IoConnection<Io>, ConnectError>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.0.poll_ready(cx)
}
fn call(&mut self, req: Connect) -> Self::Future {
let mut connector = self.0.clone();
fn call(&self, req: Connect) -> Self::Future {
let connector = self.0.clone();
let inner = self.1.clone();
let fut = async move {
@ -325,7 +326,7 @@ where
{
if let Some(timeout) = self.config.disconnect_timeout {
if let ConnectionType::H1(io) = conn.io {
actix_rt::spawn(CloseConnection::new(io, timeout))
actix_rt::spawn(CloseConnection::new(io, timeout));
}
}
} else {
@ -340,7 +341,7 @@ where
if let ConnectionType::H1(io) = io {
actix_rt::spawn(CloseConnection::new(
io, timeout,
))
));
}
}
continue;
@ -372,7 +373,7 @@ where
self.acquired -= 1;
if let Some(timeout) = self.config.disconnect_timeout {
if let ConnectionType::H1(io) = io {
actix_rt::spawn(CloseConnection::new(io, timeout))
actix_rt::spawn(CloseConnection::new(io, timeout));
}
}
self.check_availability();
@ -428,7 +429,7 @@ struct ConnectorPoolSupport<T, Io>
where
Io: AsyncRead + AsyncWrite + Unpin + 'static,
{
connector: T,
connector: Rc<T>,
inner: Rc<RefCell<Inner<Io>>>,
}
@ -535,7 +536,7 @@ where
rx: Some(rx),
inner: Some(inner),
config,
})
});
}
}

22
actix-http/src/h1/dispatcher.rs
View File

@ -1,5 +1,4 @@
use std::{
cell::RefCell,
collections::VecDeque,
fmt,
future::Future,
@ -91,7 +90,7 @@ where
U: Service<(Request, Framed<T, Codec>), Response = ()>,
U::Error: fmt::Display,
{
flow: Rc<RefCell<HttpFlow<S, X, U>>>,
flow: Rc<HttpFlow<S, X, U>>,
on_connect_data: OnConnectData,
flags: Flags,
peer_addr: Option<net::SocketAddr>,
@ -177,7 +176,7 @@ where
pub(crate) fn new(
stream: T,
config: ServiceConfig,
services: Rc<RefCell<HttpFlow<S, X, U>>>,
services: Rc<HttpFlow<S, X, U>>,
on_connect_data: OnConnectData,
peer_addr: Option<net::SocketAddr>,
) -> Self {
@ -200,7 +199,7 @@ where
config: ServiceConfig,
read_buf: BytesMut,
timeout: Option<Sleep>,
services: Rc<RefCell<HttpFlow<S, X, U>>>,
services: Rc<HttpFlow<S, X, U>>,
on_connect_data: OnConnectData,
peer_addr: Option<net::SocketAddr>,
) -> Self {
@ -377,7 +376,7 @@ where
Poll::Ready(Ok(req)) => {
self.as_mut().send_continue();
this = self.as_mut().project();
let fut = this.flow.borrow_mut().service.call(req);
let fut = this.flow.service.call(req);
this.state.set(State::ServiceCall(fut));
continue;
}
@ -467,12 +466,12 @@ where
if req.head().expect() {
// set dispatcher state so the future is pinned.
let mut this = self.as_mut().project();
let task = this.flow.borrow_mut().expect.call(req);
let task = this.flow.expect.call(req);
this.state.set(State::ExpectCall(task));
} else {
// the same as above.
let mut this = self.as_mut().project();
let task = this.flow.borrow_mut().service.call(req);
let task = this.flow.service.call(req);
this.state.set(State::ServiceCall(task));
};
@ -485,7 +484,7 @@ where
Poll::Ready(Ok(req)) => {
self.as_mut().send_continue();
let mut this = self.as_mut().project();
let task = this.flow.borrow_mut().service.call(req);
let task = this.flow.service.call(req);
this.state.set(State::ServiceCall(task));
continue;
}
@ -556,9 +555,7 @@ where
// merge on_connect_ext data into request extensions
this.on_connect_data.merge_into(&mut req);
if pl == MessageType::Stream
&& this.flow.borrow().upgrade.is_some()
{
if pl == MessageType::Stream && this.flow.upgrade.is_some() {
this.messages.push_back(DispatcherMessage::Upgrade(req));
break;
}
@ -811,9 +808,8 @@ where
let framed = Framed::from_parts(parts);
let upgrade = inner_p
.flow
.borrow_mut()
.upgrade
.take()
.as_ref()
.unwrap()
.call((req, framed));
self.as_mut()

8
actix-http/src/h1/expect.rs
View File

@ -1,4 +1,4 @@
use std::task::{Context, Poll};
use std::task::Poll;
use actix_service::{Service, ServiceFactory};
use futures_util::future::{ready, Ready};
@ -26,11 +26,9 @@ impl Service<Request> for ExpectHandler {
type Error = Error;
type Future = Ready<Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, req: Request) -> Self::Future {
fn call(&self, req: Request) -> Self::Future {
ready(Ok(req))
// TODO: add some way to trigger error
// Err(error::ErrorExpectationFailed("test"))

16
actix-http/src/h1/service.rs
View File

@ -1,4 +1,3 @@
use std::cell::RefCell;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
@ -367,7 +366,7 @@ where
X: Service<Request>,
U: Service<(Request, Framed<T, Codec>)>,
{
flow: Rc<RefCell<HttpFlow<S, X, U>>>,
flow: Rc<HttpFlow<S, X, U>>,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
cfg: ServiceConfig,
_phantom: PhantomData<B>,
@ -417,9 +416,9 @@ where
type Error = DispatchError;
type Future = Dispatcher<T, S, B, X, U>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut flow = self.flow.borrow_mut();
let ready = flow
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let ready = self
.flow
.expect
.poll_ready(cx)
.map_err(|e| {
@ -429,7 +428,8 @@ where
})?
.is_ready();
let ready = flow
let ready = self
.flow
.service
.poll_ready(cx)
.map_err(|e| {
@ -440,7 +440,7 @@ where
.is_ready()
&& ready;
let ready = if let Some(ref mut upg) = flow.upgrade {
let ready = if let Some(ref upg) = self.flow.upgrade {
upg.poll_ready(cx)
.map_err(|e| {
let e = e.into();
@ -460,7 +460,7 @@ where
}
}
fn call(&mut self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
fn call(&self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
let on_connect_data =
OnConnectData::from_io(&io, self.on_connect_ext.as_deref());

8
actix-http/src/h1/upgrade.rs
View File

@ -1,4 +1,4 @@
use std::task::{Context, Poll};
use std::task::Poll;
use actix_codec::Framed;
use actix_service::{Service, ServiceFactory};
@ -28,11 +28,9 @@ impl<T> Service<(Request, Framed<T, Codec>)> for UpgradeHandler {
type Error = Error;
type Future = Ready<Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
actix_service::always_ready!();
fn call(&mut self, _: (Request, Framed<T, Codec>)) -> Self::Future {
fn call(&self, _: (Request, Framed<T, Codec>)) -> Self::Future {
ready(Ok(()))
}
}

9
actix-http/src/h2/dispatcher.rs
View File

@ -1,4 +1,3 @@
use std::cell::RefCell;
use std::future::Future;
use std::marker::PhantomData;
use std::net;
@ -37,7 +36,7 @@ where
S: Service<Request>,
B: MessageBody,
{
flow: Rc<RefCell<HttpFlow<S, X, U>>>,
flow: Rc<HttpFlow<S, X, U>>,
connection: Connection<T, Bytes>,
on_connect_data: OnConnectData,
config: ServiceConfig,
@ -56,7 +55,7 @@ where
B: MessageBody,
{
pub(crate) fn new(
services: Rc<RefCell<HttpFlow<S, X, U>>>,
flow: Rc<HttpFlow<S, X, U>>,
connection: Connection<T, Bytes>,
on_connect_data: OnConnectData,
config: ServiceConfig,
@ -80,7 +79,7 @@ where
};
Dispatcher {
flow: services,
flow,
config,
peer_addr,
connection,
@ -138,7 +137,7 @@ where
let svc = ServiceResponse::<S::Future, S::Response, S::Error, B> {
state: ServiceResponseState::ServiceCall(
this.flow.borrow_mut().service.call(req),
this.flow.service.call(req),
Some(res),
),
config: this.config.clone(),

11
actix-http/src/h2/service.rs
View File

@ -1,4 +1,3 @@
use std::cell::RefCell;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
@ -249,7 +248,7 @@ pub struct H2ServiceHandler<T, S, B>
where
S: Service<Request>,
{
flow: Rc<RefCell<HttpFlow<S, (), ()>>>,
flow: Rc<HttpFlow<S, (), ()>>,
cfg: ServiceConfig,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_phantom: PhantomData<B>,
@ -290,15 +289,15 @@ where
type Error = DispatchError;
type Future = H2ServiceHandlerResponse<T, S, B>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.flow.borrow_mut().service.poll_ready(cx).map_err(|e| {
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.flow.service.poll_ready(cx).map_err(|e| {
let e = e.into();
error!("Service readiness error: {:?}", e);
DispatchError::Service(e)
})
}
fn call(&mut self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
fn call(&self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
let on_connect_data =
OnConnectData::from_io(&io, self.on_connect_ext.as_deref());
@ -321,7 +320,7 @@ where
{
Incoming(Dispatcher<T, S, B, (), ()>),
Handshake(
Option<Rc<RefCell<HttpFlow<S, (), ()>>>>,
Option<Rc<HttpFlow<S, (), ()>>>,
Option<ServiceConfig>,
Option<net::SocketAddr>,
OnConnectData,

2
actix-http/src/header/into_value.rs
View File

@ -126,6 +126,6 @@ impl IntoHeaderValue for Mime {
#[inline]
fn try_into_value(self) -> Result<HeaderValue, Self::Error> {
HeaderValue::try_from(format!("{}", self))
HeaderValue::from_str(self.as_ref())
}
}

24
actix-http/src/service.rs
View File

@ -1,4 +1,3 @@
use std::cell::RefCell;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
@ -440,7 +439,7 @@ where
X: Service<Request>,
U: Service<(Request, Framed<T, h1::Codec>)>,
{
flow: Rc<RefCell<HttpFlow<S, X, U>>>,
flow: Rc<HttpFlow<S, X, U>>,
cfg: ServiceConfig,
on_connect_ext: Option<Rc<ConnectCallback<T>>>,
_phantom: PhantomData<B>,
@ -454,12 +453,12 @@ pub(super) struct HttpFlow<S, X, U> {
}
impl<S, X, U> HttpFlow<S, X, U> {
pub(super) fn new(service: S, expect: X, upgrade: Option<U>) -> Rc<RefCell<Self>> {
Rc::new(RefCell::new(Self {
pub(super) fn new(service: S, expect: X, upgrade: Option<U>) -> Rc<Self> {
Rc::new(Self {
service,
expect,
upgrade,
}))
})
}
}
@ -509,9 +508,9 @@ where
type Error = DispatchError;
type Future = HttpServiceHandlerResponse<T, S, B, X, U>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut flow = self.flow.borrow_mut();
let ready = flow
fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let ready = self
.flow
.expect
.poll_ready(cx)
.map_err(|e| {
@ -521,7 +520,8 @@ where
})?
.is_ready();
let ready = flow
let ready = self
.flow
.service
.poll_ready(cx)
.map_err(|e| {
@ -532,7 +532,7 @@ where
.is_ready()
&& ready;
let ready = if let Some(ref mut upg) = flow.upgrade {
let ready = if let Some(ref upg) = self.flow.upgrade {
upg.poll_ready(cx)
.map_err(|e| {
let e = e.into();
@ -553,7 +553,7 @@ where
}
fn call(
&mut self,
&self,
(io, proto, peer_addr): (T, Protocol, Option<net::SocketAddr>),
) -> Self::Future {
let on_connect_data =
@ -604,7 +604,7 @@ where
Option<(
Handshake<T, Bytes>,
ServiceConfig,
Rc<RefCell<HttpFlow<S, X, U>>>,
Rc<HttpFlow<S, X, U>>,
OnConnectData,
Option<net::SocketAddr>,
)>,